feat: add event-boundary timing instrumentation for PoW

aiopslab/orchestrator/tasks/analysis.py

@@ -10,13 +10,16 @@
 from aiopslab.orchestrator.actions.analysis import AnalysisActions
 from aiopslab.service.apps.base import Application
 from aiopslab.session import SessionItem
+from aiopslab.timing import ANALYSIS_COMPLETED
 from aiopslab.utils.actions import get_actions
 from aiopslab.utils.status import InvalidActionError
 
 
 class AnalysisTask(Task):
     """An AIOps root cause analysis task."""
 
+    timing_completion_event = ANALYSIS_COMPLETED
+
     def __init__(self, app: Application):
         super().__init__()
         self.app = app

aiopslab/orchestrator/tasks/base.py

@@ -6,6 +6,7 @@
 from aiopslab.service.kubectl import KubeCtl
 from aiopslab.orchestrator.evaluators.quantitative import *
 from aiopslab.orchestrator.evaluators.qualitative import LLMJudge
+from aiopslab.timing import EvaluationTiming
 
 
 config = Config(BASE_DIR / "config.yml")
@@ -14,9 +15,21 @@
 class Task:
     """Base class for all tasks."""
 
+    timing_completion_event = None
+
     def __init__(self):
         self.results = {}
         self.kubectl = KubeCtl()
+        self.timing: EvaluationTiming | None = None
+
+    def set_timing(self, timing: EvaluationTiming):
+        """Attach the session timing recorder to this task."""
+        self.timing = timing
+
+    def mark_timing_completion(self):
+        """Record this task's completion event when an event boundary is defined."""
+        if self.timing is not None and self.timing_completion_event:
+            self.timing.mark(self.timing_completion_event)
 
     def get_task_description(self):
         raise NotImplementedError("Subclasses must implement this method.")

aiopslab/orchestrator/tasks/detection.py

@@ -1,78 +1,81 @@
-# Copyright (c) Microsoft Corporation.
-# Licensed under the MIT License.
-
-"""Define and query information about an AIOps Detection task."""
-
-import textwrap
-from typing import Any
-
-
-from aiopslab.orchestrator.tasks.base import Task
-from aiopslab.orchestrator.actions.detection import DetectionActions
-from aiopslab.service.apps.base import Application
-from aiopslab.session import SessionItem
-from aiopslab.utils.actions import get_actions
-from aiopslab.utils.status import InvalidActionError
-
-
-class DetectionTask(Task):
-    """An AIOps anomaly detection task."""
-
-    def __init__(self, app: Application):
-        super().__init__()
-        self.app = app
-        self.helm_configs = self.app.helm_configs
-        self.app_summary = self.app.get_app_summary()
-        self.actions = DetectionActions()
-
-        self.task_desc = """\
-            You are an expert DevOps engineer assigned to detect anomalies in a deployed service.
-
-            Service Details:
-            {app_summary}
-
-            You will begin by analyzing the service's state and telemetry, and then submit your solution:
-            1. str: "Yes" or "No" to indicate whether there are anomalies detected
-            """
-
-        self.instructions = """\
-            You will respond with one of the above APIs as your next action.
-            Please respond in the following format in a markdown code block:
-            ```\n<API_NAME>(<API_PARAM1>, <API_PARAM2> ...)\n```
-
-            For instance, if you want to list files in current directory, your response must be exactly:
-
-            ```\nexec_shell("ls -l")\n```
-
-            If you decide that there are no anomalies:
-
-            ```\nsubmit(\"No\")\n```
-
-            Or, if anomalies are found:
-
-            ```\nsubmit(\"Yes\")\n```
-
-            Please respond with only a single API call (a.k.a., action) per turn without any additional words, labels, or prefixes.
-            """
-
-    def get_task_description(self):
-        return textwrap.dedent(self.task_desc).format(app_summary=self.app_summary)
-
-    def get_instructions(self):
-        return textwrap.dedent(self.instructions)
-
-    def get_available_actions(self):
-        return get_actions(task="detection")
-
-    def perform_action(self, action_name, *args, **kwargs):
-        action_method = getattr(self.actions, action_name, None)
-
-        if action_method is not None and callable(action_method):
-            return action_method(*args, **kwargs)
-        else:
-            raise InvalidActionError(action_name)
-
-    def eval(self, soln: Any, trace: list[SessionItem], duration: float):
-        self.add_result("TTD", duration)
-        self.common_eval(trace)
-        return self.results
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Define and query information about an AIOps Detection task."""
+
+import textwrap
+from typing import Any
+
+
+from aiopslab.orchestrator.tasks.base import Task
+from aiopslab.orchestrator.actions.detection import DetectionActions
+from aiopslab.service.apps.base import Application
+from aiopslab.session import SessionItem
+from aiopslab.timing import DETECTION_COMPLETED
+from aiopslab.utils.actions import get_actions
+from aiopslab.utils.status import InvalidActionError
+
+
+class DetectionTask(Task):
+    """An AIOps anomaly detection task."""
+
+    timing_completion_event = DETECTION_COMPLETED
+
+    def __init__(self, app: Application):
+        super().__init__()
+        self.app = app
+        self.helm_configs = self.app.helm_configs
+        self.app_summary = self.app.get_app_summary()
+        self.actions = DetectionActions()
+
+        self.task_desc = """\
+            You are an expert DevOps engineer assigned to detect anomalies in a deployed service.
+
+            Service Details:
+            {app_summary}
+
+            You will begin by analyzing the service's state and telemetry, and then submit your solution:
+            1. str: "Yes" or "No" to indicate whether there are anomalies detected
+            """
+
+        self.instructions = """\
+            You will respond with one of the above APIs as your next action.
+            Please respond in the following format in a markdown code block:
+            ```\n<API_NAME>(<API_PARAM1>, <API_PARAM2> ...)\n```
+
+            For instance, if you want to list files in current directory, your response must be exactly:
+
+            ```\nexec_shell("ls -l")\n```
+
+            If you decide that there are no anomalies:
+
+            ```\nsubmit(\"No\")\n```
+
+            Or, if anomalies are found:
+
+            ```\nsubmit(\"Yes\")\n```
+
+            Please respond with only a single API call (a.k.a., action) per turn without any additional words, labels, or prefixes.
+            """
+
+    def get_task_description(self):
+        return textwrap.dedent(self.task_desc).format(app_summary=self.app_summary)
+
+    def get_instructions(self):
+        return textwrap.dedent(self.instructions)
+
+    def get_available_actions(self):
+        return get_actions(task="detection")
+
+    def perform_action(self, action_name, *args, **kwargs):
+        action_method = getattr(self.actions, action_name, None)
+
+        if action_method is not None and callable(action_method):
+            return action_method(*args, **kwargs)
+        else:
+            raise InvalidActionError(action_name)
+
+    def eval(self, soln: Any, trace: list[SessionItem], duration: float):
+        self.add_result("TTD", duration)
+        self.common_eval(trace)
+        return self.results

aiopslab/orchestrator/tasks/localization.py

@@ -11,13 +11,16 @@
 from aiopslab.orchestrator.actions.localization import LocalizationActions
 from aiopslab.service.apps.base import Application
 from aiopslab.session import SessionItem
+from aiopslab.timing import LOCALIZATION_COMPLETED
 from aiopslab.utils.actions import get_actions
 from aiopslab.utils.status import InvalidActionError
 
 
 class LocalizationTask(Task):
     """An AIOps fault localization task."""
 
+    timing_completion_event = LOCALIZATION_COMPLETED
+
     def __init__(self, app: Application):
         super().__init__()
         self.app = app
@@ -54,7 +57,8 @@ def __init__(self, app: Application):
 
             Or, if no faults are found:
 
-            ```\nsubmit([])\n```
+            ```\nsubmit([])
+            ```
 
             Please respond with only a single API call (a.k.a., action) per turn without any additional words, labels, or prefixes.
             """

aiopslab/orchestrator/tasks/mitigation.py

@@ -10,13 +10,18 @@
 from aiopslab.orchestrator.actions.mitigation import MitigationActions
 from aiopslab.service.apps.base import Application
 from aiopslab.session import SessionItem
+from aiopslab.timing import MITIGATION_COMPLETED
 from aiopslab.utils.actions import get_actions
 from aiopslab.utils.status import InvalidActionError
 
 
 class MitigationTask(Task):
     """An AIOps anomaly mitigation task."""
 
+    # Mitigation completion is recorded after the task-specific recovery oracle
+    # confirms that the environment is healthy, not merely when submit() is called.
+    timing_completion_event = MITIGATION_COMPLETED
+
     def __init__(self, app: Application):
         super().__init__()
         self.app = app

aiopslab/session.py

@@ -10,6 +10,7 @@
 from pydantic import BaseModel
 
 from aiopslab.paths import RESULTS_DIR
+from aiopslab.timing import EvaluationTiming
 
 
 class SessionItem(BaseModel):
@@ -29,6 +30,7 @@ def __init__(self, results_dir=None) -> None:
         self.end_time = None
         self.agent_name = None
         self.results_dir = results_dir
+        self.timing = EvaluationTiming()
 
     def set_problem(self, problem, pid=None):
         """Set the problem instance for the session.
@@ -60,7 +62,7 @@ def set_agent(self, agent_name):
         """Set the agent name for the session.
 
         Args:
-            agent_name (str): The name of the agent.
+            agent_name (str): The name of the agent (default: "agent").
         """
         self.agent_name = agent_name
 
@@ -108,6 +110,7 @@ def to_dict(self):
             "problem_id": self.pid,
             "start_time": self.start_time,
             "end_time": self.end_time,
+            "timing_events": self.timing.to_dict(),
             "trace": [item.model_dump() for item in self.history],
             "results": self.results,
         }
@@ -137,4 +140,6 @@ def from_json(self, filename: str):
         self.start_time = data.get("start_time")
         self.end_time = data.get("end_time")
         self.results = data.get("results")
-        self.history = [SessionItem.model_validate(item) for item in data.get("trace")]
+        self.history = [SessionItem.model_validate(item) for item in data.get("trace", [])]
+        for event, timestamp in data.get("timing_events", {}).items():
+            self.timing.mark(event, timestamp)

aiopslab/timing.py

@@ -0,0 +1,45 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Event-boundary timing primitives for AIOps evaluation."""
+
+from __future__ import annotations
+
+import time
+from typing import Dict
+
+
+FAULT_OCCURRED = "fault_occurred"
+DETECTION_COMPLETED = "detection_completed"
+LOCALIZATION_COMPLETED = "localization_completed"
+ANALYSIS_COMPLETED = "analysis_completed"
+MITIGATION_COMPLETED = "mitigation_completed"
+
+
+class EvaluationTiming:
+    """Record benchmark lifecycle events and derive elapsed intervals."""
+
+    def __init__(self) -> None:
+        self.events: Dict[str, float] = {}
+
+    def mark(self, event: str, timestamp: float | None = None) -> float:
+        """Record an event timestamp and return the stored value."""
+        value = time.time() if timestamp is None else timestamp
+        self.events[event] = value
+        return value
+
+    def get(self, event: str) -> float | None:
+        """Return an event timestamp, if recorded."""
+        return self.events.get(event)
+
+    def elapsed(self, start_event: str, end_event: str) -> float | None:
+        """Return elapsed time between two recorded events."""
+        start = self.get(start_event)
+        end = self.get(end_event)
+        if start is None or end is None:
+            return None
+        return end - start
+
+    def to_dict(self) -> Dict[str, float]:
+        """Return a JSON-serializable snapshot of recorded events."""
+        return dict(self.events)

docs/evaluation_timing.md

@@ -0,0 +1,32 @@
+# Event-Boundary Evaluation Timing
+
+AIOpsLab currently exposes `TTD`, `TTL`, `TTA`, and `TTM` from the session duration. That duration is useful for backward compatibility, but it does not by itself identify the benchmark lifecycle boundaries described by the research-aligned timing model.
+
+This PoW adds event instrumentation without changing the historical metric fields yet.
+
+## Canonical boundaries
+
+| Event | Meaning |
+|---|---|
+| `fault_occurred` | Fault injection has completed and the fault is active. |
+| `detection_completed` | A valid detection submission has been accepted. |
+| `localization_completed` | A valid localization submission has been accepted. |
+| `analysis_completed` | A valid RCA/analysis submission has been accepted. |
+| `mitigation_completed` | Mitigation has reached an oracle-confirmed completion point. |
+
+## Derived intervals
+
+The intended event-boundary model is:
+
+- `TTD = detection_completed - fault_occurred`
+- `TTL = localization_completed - fault_occurred`
+- `TTA = analysis_completed - fault_occurred`
+- `TTM = mitigation_completed - detection_completed`
+
+The last four formulas are treated as the PoW target model; the primary AIOpsLab research source explicitly establishes TTD and TTM, while TTL/TTA remain operational extensions until independently sourced.
+
+## Compatibility rule
+
+Existing result fields are not rewritten by this PoW. Historical session-duration values remain valid as historical/legacy measurements. New event timestamps are persisted separately so a later PR can migrate the metric fields only after regression coverage demonstrates that all benchmark task types expose the required boundaries.
+
+No-op problems must not be interpreted as instantaneous fault detection merely because a session duration exists; future metric migration should represent the absence of a fault boundary explicitly.

tests/unit/test_timing.py

@@ -0,0 +1,16 @@
+from aiopslab.timing import EvaluationTiming
+
+
+def test_elapsed_uses_explicit_event_boundaries():
+    timing = EvaluationTiming()
+    timing.mark("fault_occurred", 10.0)
+    timing.mark("detection_completed", 12.5)
+
+    assert timing.elapsed("fault_occurred", "detection_completed") == 2.5
+
+
+def test_missing_boundary_does_not_fabricate_duration():
+    timing = EvaluationTiming()
+    timing.mark("fault_occurred", 10.0)
+
+    assert timing.elapsed("fault_occurred", "mitigation_completed") is None