refactor: isolate worker stdio and stream tool logs

agent/engine.py

@@ -1,16 +1,13 @@
-import os
-import time
-import traceback
-from threading import Event, Thread
-
 from chipcompiler.data import StateEnum, WorkspaceStep
-from chipcompiler.engine.flow import EngineFlow, get_process_rss_mb, track_current_process_memory
-from chipcompiler.utility.log import redirect_stdio_to_file
+from chipcompiler.engine.flow import EngineFlow
 
 from .tools import run_step as run_agent_step
 
 
 class AgentEngineFlow(EngineFlow):
+    """Flow Agent engine: the canonical step lifecycle lives in EngineFlow;
+    only the tool runner and the agent's result vocabulary differ."""
+
     def build_default_steps(self):
         super().build_default_steps()
         steps = self.workspace.flow.data["steps"]
@@ -20,76 +17,16 @@ def build_default_steps(self):
         steps.insert(filler_index, self.init_flow_step("DRC", "ecc", StateEnum.Unstart))
         self.save()
 
-    def run_step(self, workspace_step: WorkspaceStep | str, *, rerun: bool = False) -> StateEnum:
-        if isinstance(workspace_step, str):
-            workspace_step = self.get_workspace_step(workspace_step)
-        if workspace_step is None:
-            return StateEnum.Invalid
-        step_tag = f"{workspace_step.name}({workspace_step.tool})"
-        if not rerun and self.check_state(
-            name=workspace_step.name, tool=workspace_step.tool, state=StateEnum.Success
-        ):
-            self.workspace.logger.info("[SKIP] %s already succeeded", step_tag)
-            self.clear_db_engine_after_step(workspace_step, StateEnum.Success)
-            return StateEnum.Success
-
-        start_time = time.time()
-        timing_constraints = self.timing_constraint_facts()
-        self.set_state(name=workspace_step.name, tool=workspace_step.tool, state=StateEnum.Ongoing)
-        self._redirect_step_stdio(workspace_step)
-        start_memory, peak_memory, stop_monitor, monitor = self._start_memory_monitor()
-        result = False
-        try:
-            result = run_agent_step(
-                workspace=self.workspace, step=workspace_step, ecc_module=self.engine_db.engine
-            )
-            self.workspace.logger.info("[STEP] %s finished result=%s", step_tag, result)
-        except Exception:
-            self.workspace.logger.error("[STEP] %s failed with exception", step_tag)
-            traceback.print_exc()
-        finally:
-            self._stop_memory_monitor(stop_monitor, monitor)
-
-        elapsed = time.time() - start_time
-        state = self._step_state(workspace_step, result)
-        self._finish_step(
-            workspace_step,
-            state,
-            elapsed,
-            timing_constraints,
-            max(0, round(peak_memory[0] - start_memory, 3)),
-        )
-        return state
-
-    def _redirect_step_stdio(self, workspace_step: WorkspaceStep) -> None:
-        log_file = workspace_step.log.file or ""
-        if not log_file:
-            return
-        try:
-            log_file = os.path.abspath(log_file)
-            os.makedirs(os.path.dirname(log_file) or ".", exist_ok=True)
-            redirect_stdio_to_file(log_file)
-        except Exception:
-            traceback.print_exc()
-
-    def _start_memory_monitor(self) -> tuple[float, list[float], Event, Thread]:
-        start_memory = get_process_rss_mb(os.getpid())
-        peak_memory = [start_memory]
-        stop_monitor = Event()
-        monitor = Thread(
-            target=track_current_process_memory,
-            args=(os.getpid(), stop_monitor, peak_memory),
-            daemon=True,
+    def _invoke_step_tool(self, workspace_step: WorkspaceStep):
+        return run_agent_step(
+            workspace=self.workspace, step=workspace_step, ecc_module=self.engine_db.engine
         )
-        monitor.start()
-        return start_memory, peak_memory, stop_monitor, monitor
 
-    @staticmethod
-    def _stop_memory_monitor(stop_monitor: Event, monitor: Thread) -> None:
-        stop_monitor.set()
-        monitor.join()
-
-    def _step_state(self, workspace_step: WorkspaceStep, result: object) -> StateEnum:
+    def _derive_step_state(
+        self, workspace_step: WorkspaceStep, result, *, raised: bool
+    ) -> StateEnum:
+        if raised:
+            return StateEnum.Imcomplete
         if result is StateEnum.Invalid:
             return StateEnum.Invalid
         if result is True or result is StateEnum.Success:
@@ -99,52 +36,3 @@ def _step_state(self, workspace_step: WorkspaceStep, result: object) -> StateEnu
                 else StateEnum.Imcomplete
             )
         return StateEnum.Imcomplete
-
-    def _finish_step(
-        self,
-        workspace_step: WorkspaceStep,
-        state: StateEnum,
-        elapsed: float,
-        timing_constraints: dict,
-        peak_memory_mb: float,
-    ) -> None:
-        runtime = f"{int(elapsed // 3600)}:{int((elapsed % 3600) // 60)}:{int(elapsed % 60)}"
-        self.set_state(
-            name=workspace_step.name,
-            tool=workspace_step.tool,
-            state=state,
-            runtime=runtime,
-            peak_memory=peak_memory_mb,
-        )
-        if state == StateEnum.Success:
-            self._save_agent_step_facts(
-                workspace_step,
-                state,
-                elapsed,
-                peak_memory_mb,
-                timing_constraints,
-            )
-        self.clear_db_engine_after_step(workspace_step, state)
-
-    def _save_agent_step_facts(
-        self,
-        workspace_step: WorkspaceStep,
-        state: StateEnum,
-        elapsed: float,
-        peak_memory: float,
-        timing_constraints: dict,
-    ) -> None:
-        from chipcompiler.tools import build_step_metrics, save_layout_image
-
-        if self.save_step_flow_facts(
-            workspace_step=workspace_step,
-            state=state,
-            runtime_seconds=elapsed,
-            peak_memory_mb=peak_memory,
-            timing_constraints=timing_constraints,
-        ):
-            try:
-                build_step_metrics(workspace=self.workspace, step=workspace_step)
-            except Exception:
-                self.workspace.logger.exception("[QOR] failed to refresh analysis")
-        save_layout_image(workspace=self.workspace, step=workspace_step)

agent/test/test_engine.py

@@ -38,3 +38,98 @@ def run_step(**_kwargs):
 
     assert flow.run_step(step) is expected_state
     assert flow.check_state("route", "ecc", expected_state)
+
+
+def _marker_workspace(tmp_path):
+    workspace = Workspace(directory=tmp_path, flow=Flow(path=tmp_path / "flow.json"))
+    flow = AgentEngineFlow(workspace)
+    workspace.flow.data = {"steps": [{"name": "route", "tool": "ecc", "state": "Unstart"}]}
+    step = EccStep(name="route", directory=tmp_path, tool="ecc")
+    flow.workspace_steps = [step]
+    flow.engine_db = SimpleNamespace(engine=None)
+    return flow, step
+
+
+def test_agent_engine_emits_markers_around_step_writes(monkeypatch, tmp_path):
+    """The agent executor frames the step stream with markers and never
+    redirects stdio into a step log file."""
+    flow, step = _marker_workspace(tmp_path)
+    events = []
+    monkeypatch.setattr(flow, "check_step_result", lambda **_kwargs: True)
+    monkeypatch.setattr(
+        "agent.engine.run_agent_step",
+        lambda **kwargs: events.append(("tool", None)) or True,
+    )
+    monkeypatch.setattr(flow, "save_step_flow_facts", lambda **_kwargs: False)
+    monkeypatch.setattr(
+        "chipcompiler.tools.save_layout_image",
+        lambda **_kwargs: events.append(("layout", None)),
+    )
+    monkeypatch.setattr(
+        "chipcompiler.runtime.log_stream.emit_step_marker",
+        lambda event, *, step, tool: events.append(("marker", event)),
+    )
+
+    assert flow.run_step(step) is StateEnum.Success
+    end_index = events.index(("marker", "end"))
+    assert events.index(("marker", "begin")) < events.index(("tool", None))
+    assert events.index(("layout", None)) < end_index
+
+
+def test_agent_engine_suppresses_end_marker_when_final_save_fails(monkeypatch, tmp_path):
+    """A failed final save downgrades the record and suppresses the end
+    marker, matching the base engine's authoritative-save contract."""
+    flow, step = _marker_workspace(tmp_path)
+    markers = []
+    monkeypatch.setattr(flow, "check_step_result", lambda **_kwargs: True)
+    monkeypatch.setattr("agent.engine.run_agent_step", lambda **kwargs: True)
+    monkeypatch.setattr(
+        "chipcompiler.runtime.log_stream.emit_step_marker",
+        lambda event, *, step, tool: markers.append(event),
+    )
+
+    real_save = flow.save
+    save_calls = []
+
+    def save_failing_on_final():
+        save_calls.append(len(save_calls) + 1)
+        if len(save_calls) == 1:
+            return real_save()  # the Ongoing save persists
+        return False  # the one final save fails
+
+    monkeypatch.setattr(flow, "save", save_failing_on_final)
+
+    assert flow.run_step(step) is StateEnum.Imcomplete
+    assert save_calls == [1, 2]
+    assert markers == ["begin"]
+    record = flow.get_step("route", "ecc")
+    assert record["state"] == StateEnum.Imcomplete.value
+
+
+def test_agent_engine_inherits_lifecycle_and_never_opens_step_logs(monkeypatch, tmp_path):
+    """The inherited base lifecycle drives the agent tool hook end to end,
+    and no step log file is opened even when the step declares a log path."""
+    flow, step = _marker_workspace(tmp_path)
+    declared_log = tmp_path / "route_ecc" / "log" / "route.log"
+    step.log.file = declared_log
+    tool_calls = []
+    monkeypatch.setattr(flow, "check_step_result", lambda **_kwargs: True)
+    monkeypatch.setattr(
+        "agent.engine.run_agent_step",
+        lambda **kwargs: tool_calls.append(kwargs) or True,
+    )
+    monkeypatch.setattr(flow, "save_step_flow_facts", lambda **_kwargs: False)
+    monkeypatch.setattr("chipcompiler.tools.save_layout_image", lambda **_kwargs: True)
+    markers = []
+    monkeypatch.setattr(
+        "chipcompiler.runtime.log_stream.emit_step_marker",
+        lambda event, *, step, tool: markers.append(event),
+    )
+
+    assert flow.run_step(step) is StateEnum.Success
+    # The agent runner hook was invoked through the base lifecycle.
+    assert len(tool_calls) == 1
+    assert tool_calls[0]["step"] is step
+    assert markers == ["begin", "end"]
+    assert not declared_log.exists()
+    assert flow.check_state("route", "ecc", StateEnum.Success)

agent/test/test_workspace_api.py

@@ -1,6 +1,10 @@
+import json
+import os
 from pathlib import Path
 from types import SimpleNamespace
 
+import pytest
+
 from agent.requests import CandidateRerunRequest
 from agent.workspace_api import FlowAgentRuntimeApi, _candidate_step_artifact_dirs
 from chipcompiler.data import StateEnum
@@ -112,6 +116,61 @@ def test_candidate_rerun_uses_the_agent_flow_and_replays_its_receipts(monkeypatc
     assert not list(cts_output.iterdir())
 
 
+def test_candidate_step_exception_reconciles_the_record(monkeypatch, tmp_path, capfd):
+    """A run_step raising after its begin marker must downgrade the record
+    before the exception propagates — not leave Success over a partial log."""
+    (tmp_path / "home").mkdir()
+    steps = [{"name": "place", "tool": "dreamplace", "state": "Success"}]
+    (tmp_path / "home" / "flow.json").write_text(json.dumps({"steps": steps}))
+    workspace = SimpleNamespace(
+        directory=tmp_path,
+        flow=SimpleNamespace(data={"steps": [dict(s) for s in steps]}),
+    )
+    place_output = tmp_path / "place_dreamplace" / "output"
+    place_output.mkdir(parents=True)
+    step = SimpleNamespace(
+        name="place",
+        tool="dreamplace",
+        output=EccOutput(dir=place_output),
+        analysis={},
+    )
+    flow = _Flow(workspace, (step,))
+
+    from chipcompiler.runtime.log_stream import emit_step_marker
+
+    def raising_run_step(step, *, rerun):
+        flow.run_calls.append((step.name, rerun))
+        emit_step_marker("begin", step=step.name, tool=step.tool)
+        os.write(2, b"partial\n")
+        raise RuntimeError("layout save blew up")
+
+    flow.run_step = raising_run_step
+    flow.set_state = lambda name, tool, state: (
+        workspace.flow.data["steps"][0].update({"state": state.value})
+    )
+
+    api = FlowAgentRuntimeApi(_EccApi(workspace))
+    monkeypatch.setattr("agent.workspace_api.build_agent_flow_for_workspace", lambda _ws: flow)
+    monkeypatch.setattr(
+        "agent.workspace_api._init_db_engine_for_workspace_step", lambda _flow, _step: None
+    )
+
+    with pytest.raises(RuntimeError, match="layout save blew up"):
+        api.candidate_rerun(
+            CandidateRerunRequest(
+                workspace_id="workspace-1",
+                target_step="place",
+                end_step="place",
+                candidate_id=None,
+                patch=None,
+                execution_scope="single_step",
+            )
+        )
+
+    assert workspace.flow.data["steps"][0]["state"] == StateEnum.Imcomplete.value
+    assert "ECC-STEP" not in capfd.readouterr().err
+
+
 class _EccApi:
     def __init__(self, workspace):
         self.session = SimpleNamespace(workspace=workspace, db_handle=None)

agent/workspace_api.py

@@ -3,6 +3,7 @@
 from hashlib import sha256
 from pathlib import Path
 
+from chipcompiler.data import StateEnum
 from chipcompiler.runtime.requests import WorkspaceIdRequest
 from chipcompiler.runtime.workspace_api import (
     RuntimeApiError,
@@ -262,8 +263,40 @@ def _clear_candidate_artifact_dir(workspace_root: Path, directory: Path, step_na
 
 
 def _run_candidate_step(flow, step) -> None:
+    from chipcompiler.runtime.log_stream import archive_own_step_logs
+
     _init_db_engine_for_workspace_step(flow, step)
-    state = flow.run_step(step, rerun=True)
+    # In-process execution is still executor+client in one process: route the
+    # own fd-2 stream through the reader so markers are consumed and the
+    # step's bytes land in its archive (echoed to the real stderr).
+    reader = None
+    try:
+        with archive_own_step_logs(flow.workspace.directory) as active_reader:
+            reader = active_reader
+            state = flow.run_step(step, rerun=True)
+    except BaseException:
+        # A step raising after its begin marker (post-processing, the end
+        # write) leaves the reader holding an active step while flow.json may
+        # already say Success; reconcile before propagating.
+        if reader is not None and (
+            reader.state.error is not None or reader.state.active_step is not None
+        ):
+            flow.set_state(step.name, step.tool, StateEnum.Imcomplete)
+        raise
+    # An archive failure or unmatched begin must not report success while the
+    # step's log is missing; downgrade so a later rerun rebuilds it. A None
+    # reader means an outer client owns the stream (passthrough) — nothing to
+    # reconcile here.
+    if reader is not None and (
+        reader.state.error is not None or reader.state.active_step is not None
+    ):
+        # set_state owns the authoritative save; a failed save is logged there.
+        flow.set_state(step.name, step.tool, StateEnum.Imcomplete)
+        raise RuntimeApiError(
+            "command_failed",
+            f"candidate rerun step {step.name} log archival failed: "
+            f"{reader.state.error or 'unmatched begin marker'}",
+        )
     if _state_value(state) != "Success":
         raise RuntimeApiError(
             "command_failed",