fix: preserve service readiness truth

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>

ouroboros/tools/services.py

@@ -32,6 +32,7 @@ from ouroboros.utils import append_jsonl, utc_now_iso
 from ouroboros.workspace_executor import executor_ref_from_ctx
 from ouroboros.workspace_executor import kill_all_services as executor_kill_all_services
 from ouroboros.workspace_executor import map_host_path as executor_map_host_path
+from ouroboros.workspace_executor import _read_local_service_marker
 from ouroboros.workspace_executor import service_logs as executor_service_logs
 from ouroboros.workspace_executor import service_status as executor_service_status
 from ouroboros.workspace_executor import start_service as executor_start_service
@@ -52,6 +53,10 @@ class ServiceRecord:
     started_at: float = field(default_factory=time.time)
     readiness: Dict[str, Any] = field(default_factory=dict)
     ready: bool = False
+    ready_observed_at: str = ""
+    readiness_log_offset: int = 0
+    readiness_log_carry: bytes = b""
+    readiness_log_identity: tuple[int, int] | None = None
     outputs: List[str] = field(default_factory=list)
     cwd_root: str = ""
     cwd_base: str = ""
@@ -353,15 +358,26 @@ def _refresh_ready(record: ServiceRecord) -> bool:
     if record.proc.poll() is not None:
         record.ready = False
         return False
+    if record.ready:
+        return True
     readiness = record.readiness or {}
     contains = str(readiness.get("stdout_contains") or readiness.get("log_contains") or "").strip()
     if not contains:
         record.ready = True
+        record.ready_observed_at = record.ready_observed_at or utc_now_iso()
         return True
-    record.ready = contains in _tail(record.log_path, 20_000)
+    record.ready = _readiness_marker_observed(record, contains)
+    if record.ready:
+        record.ready_observed_at = record.ready_observed_at or utc_now_iso()
     return record.ready
 
 
+def _readiness_marker_observed(record: ServiceRecord, marker: str) -> bool:
+    """Scan only unseen service-log bytes without turning the display tail into truth."""
+
+    return _read_local_service_marker(record, record.log_path, marker)
+
+
 def _start_service(
     ctx: ToolContext,
     cmd: List[str],
@@ -543,6 +559,7 @@ def _status_payload(record: ServiceRecord) -> Dict[str, Any]:
         "pgid": record.pgid,
         "state": state,
         "ready": bool(record.ready),
+        "ready_observed_at": record.ready_observed_at or None,
         "returncode": rc,
         "uptime_sec": round(max(0.0, time.time() - record.started_at), 3),
         "cwd": record.cwd,

tests/test_workspace_executor.py

@@ -571,7 +571,11 @@ def test_executor_local_service_lifecycle_hides_private_snapshot(tmp_path, monke
             "start_service",
             {
                 "name": "svc",
-                "cmd": [sys.executable, "-c", "import time; print('READY', flush=True); time.sleep(30)"],
+                "cmd": [
+                    sys.executable,
+                    "-c",
+                    "import os,time; os.write(1, b'READY\\n' + b'x' * 25000); time.sleep(30)",
+                ],
                 "readiness": {"log_contains": "READY", "timeout_sec": 5},
             },
         )
@@ -582,13 +586,70 @@ def test_executor_local_service_lifecycle_hides_private_snapshot(tmp_path, monke
     stopped = json.loads(stopped_raw)
 
     assert started["ready"] is True
+    assert started["ready_observed_at"]
     assert status["state"] == "running"
-    assert "READY" in logs["tail"]
+    assert "READY" not in logs["tail"]
+    assert "x" in logs["tail"]
     assert stopped["state"] == "stopped"
     assert "_before_outputs" not in stopped_raw
     assert bootstrap_calls
 
 
+def test_executor_readiness_scans_before_large_log_suffix(tmp_path, monkeypatch):
+    import ouroboros.workspace_executor as workspace_executor
+
+    log_path = tmp_path / "executor-service.log"
+    log_path.write_bytes(b"READY\n" + (b"x" * 25_000))
+    record = SimpleNamespace(
+        executor=SimpleNamespace(kind="local"),
+        backend_log_path=str(log_path),
+        local_proc=SimpleNamespace(poll=lambda: None),
+        ready=False,
+    )
+    monkeypatch.setattr(workspace_executor.time, "sleep", lambda _seconds: None)
+
+    workspace_executor._wait_readiness(
+        record,
+        {"log_contains": "READY", "timeout_sec": 0.05},
+    )
+
+    assert record.ready is True
+
+
+def test_executor_terminal_payload_clears_readiness(tmp_path):
+    import ouroboros.workspace_executor as workspace_executor
+
+    record = SimpleNamespace(
+        service_id="task:svc",
+        name="svc",
+        task_id="task",
+        executor=SimpleNamespace(
+            executor_id="local-service",
+            kind="local",
+            network="host",
+        ),
+        backend_pid="4321",
+        backend_cwd="/workspace",
+        host_cwd=tmp_path,
+        cwd_root="active_workspace",
+        cwd_base=str(tmp_path),
+        cwd_source="active_workspace",
+        skill_name="",
+        cmd=["service"],
+        outputs=[],
+        keep_alive=False,
+        backend_log_path=str(tmp_path / "service.log"),
+        started_at=workspace_executor.time.time(),
+        ready=True,
+    )
+
+    payload = workspace_executor._service_payload(record, state="exited")
+
+    assert payload["state"] == "exited"
+    assert payload["ready"] is False
+    assert record.ready is False
+
+
 def test_start_service_with_executor_ref_uses_local_for_unmapped_task_drive_cwd(tmp_path, monkeypatch):
     import ouroboros.safety as safety_mod
     from ouroboros.tool_access import resource_root_path