fix(workflows): read the service wire on a detached start

Every durable approval continuation reported "control delivery unreachable"
with "Workflow service emitted an unknown record before detached start", while
the runner admitted the continuation and ran the turn to completion. Two
executions reached terminal/completed and still carried
continuation_delivery_failed.

The strict start parser required the first NDJSON record to carry kind "event"
or a successful kind "result". That is the RUNNER's vocabulary. The API does
not call the runner. It calls the deployed workflow service, which re-frames
every runner record as an agenta event, {"type", "data"}, with no kind field
anywhere. So the strict branch rejected the first record of every
continuation, always.

_stream_service_started now accepts the first JSON object as the started
handshake. A new _detached_start_failure rejects only an explicit failure
frame, and reads one in either vocabulary: the service's {"type": "error"},
and the runner's {"kind": "result", "result": {"ok": false}} where a
deployment forwards runner records verbatim.

The tests replay the record sequences from the browser pass of 2026-09-04.

Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk

Author: mmabrouk

api/oss/src/core/workflows/service.py

@@ -756,12 +756,17 @@ async def _stream_service_started(
     ) -> WorkflowServiceDetachedResponse:
         """Stream the service ``/invoke`` and return on the FIRST record (the started handshake).
 
-        The runner emits NDJSON ``{"kind": "event"|"result", ...}`` records the moment each is
-        built; the first one means the run is accepted and owned (the alive-held handshake). We
-        return then and close the connection — the runner owns the run (alive watchdog) and
-        persists independently (producer-driven ingest), so draining to completion is unnecessary.
-        The read timeout is generous (sandbox cold-start can take seconds); we are NOT awaiting
-        the whole run, so it is not the batch 60s-whole-run budget.
+        The deployed workflow service emits one NDJSON record the moment each is built; the first
+        one means the run is accepted and owned (the alive-held handshake). We return then and
+        close the connection — the runner owns the run (alive watchdog) and persists independently
+        (producer-driven ingest), so draining to completion is unnecessary. The read timeout is
+        generous (sandbox cold-start can take seconds); we are NOT awaiting the whole run, so it
+        is not the batch 60s-whole-run budget.
+
+        ``strict_first_record`` (a durable control command) surfaces an explicit failure frame
+        instead of reporting it as a start. It does NOT require a particular record shape: see
+        ``_detached_start_failure`` for why the two producers on this stream disagree about the
+        vocabulary, and why anything unrecognised is a start.
         """
         headers = inject(
             {
@@ -813,22 +818,12 @@ async def _stream_service_started(
                         raise WorkflowDetachedStartFailed(
                             "Workflow service emitted a non-object record before detached start."
                         )
-                    kind = record.get("kind") if isinstance(record, dict) else None
-                    if strict_first_record and kind == "result":
-                        result = record.get("result")
-                        if not isinstance(result, dict) or result.get("ok") is not True:
-                            detail = (
-                                result.get("error")
-                                if isinstance(result, dict)
-                                else "malformed result record"
-                            )
+                    if strict_first_record and isinstance(record, dict):
+                        failure = WorkflowsService._detached_start_failure(record)
+                        if failure is not None:
                             raise WorkflowDetachedStartFailed(
-                                f"Workflow service rejected detached start: {detail}"
+                                f"Workflow service rejected detached start: {failure}"
                             )
-                    elif strict_first_record and kind != "event":
-                        raise WorkflowDetachedStartFailed(
-                            "Workflow service emitted an unknown record before detached start."
-                        )
                     record_run_id = (
                         record.get("run_id") if isinstance(record, dict) else None
                     )
@@ -844,6 +839,48 @@ async def _stream_service_started(
             "Workflow service closed the stream before emitting a started record."
         )
 
+    @staticmethod
+    def _detached_start_failure(record: dict) -> Optional[str]:
+        """Read an explicit failure out of the FIRST record, in either wire vocabulary.
+
+        Two producers can answer this stream and they do not share a vocabulary.
+
+        * The deployed workflow SERVICE is the ordinary case. It streams agenta event frames,
+          ``{"type": ..., "data": {...}}``, and its failure frame is ``{"type": "error"}``. There
+          is no ``kind`` anywhere on that wire.
+        * The agent RUNNER's own NDJSON, ``{"kind": "event"|"result"}``, reaches this parser only
+          where a deployment forwards the runner stream verbatim. Its failure is a terminal
+          ``{"kind": "result", "result": {"ok": false}}``.
+
+        Everything else is the started handshake. Rejecting an unrecognised record instead is what
+        made EVERY durable continuation report `unreachable` while the runner was in fact already
+        running the turn: the service's first frame carries ``type``, never ``kind``.
+
+        A runner that refuses a continuation outright never reaches here at all. The SDK turns its
+        ``ok: false`` result into an exception inside the already-committed ASGI response, so the
+        service closes the stream having written nothing, and the caller raises the
+        "closed the stream" failure above.
+        """
+        if record.get("kind") == "result":
+            result = record.get("result")
+            if isinstance(result, dict) and result.get("ok") is True:
+                return None
+            detail = (
+                result.get("error")
+                if isinstance(result, dict)
+                else "malformed result record"
+            )
+            return str(detail or "the runner rejected the run")
+
+        if record.get("type") == "error":
+            data = record.get("data")
+            message = data.get("message") if isinstance(data, dict) else None
+            code = data.get("code") if isinstance(data, dict) else None
+            detail = str(message or "the service reported an error")
+            return f"{detail} ({code})" if code else detail
+
+        return None
+
     @staticmethod
     def _coerce_invoke_response(
         *,

api/oss/tests/pytest/unit/workflows/test_invoke_detached.py

@@ -134,7 +134,8 @@ async def test_stream_service_started_raises_on_http_error():
         "[]",
         '{"kind": "result", "result": {"ok": false, "error": "rejected"}}',
         '{"kind": "result"}',
-        '{"kind": "unknown"}',
+        # The service wire's own failure frame (an agenta `error` event).
+        '{"type": "error", "data": {"type": "error", "message": "no key", "code": "auth"}}',
     ],
 )
 async def test_stream_service_started_rejects_failure_or_malformed_first_record(line):
@@ -164,6 +165,81 @@ async def test_stream_service_started_keeps_legacy_best_effort_for_ordinary_trig
     assert result.run_id == "run-x"
 
 
+@pytest.mark.parametrize(
+    "line",
+    [
+        # The record sequence a durable continuation really produced (browser pass,
+        # 2026-09-04 17:35Z, session d99f32ae / command 01a06d7d): the runner admitted the
+        # continuation and its first event was a `tool_call`. The DEPLOYED SERVICE re-frames
+        # every runner record as an agenta event, `{"type", "data"}` — there is no `kind` on
+        # that wire, and reading the first frame as a runner record called every one of those
+        # deliveries unreachable while the turn ran to completion underneath the card.
+        '{"type": "tool_call", "data": {"type": "tool_call", "id": "t1", "name": "Bash"}}',
+        '{"type": "interaction_response", "data": {"type": "interaction_response"}}',
+        '{"type": "message", "data": {"type": "message", "text": "ok"}}',
+        # An unrecognised record is a start, not a failure: only an explicit failure frame is.
+        '{"kind": "unknown"}',
+        '{"type": "error_recovered", "data": {}}',
+    ],
+)
+async def test_stream_service_started_accepts_a_service_event_frame_as_the_start(line):
+    response = _FakeStreamResponse(lines=[line, '{"type": "done", "data": {}}'])
+    with patch("httpx.AsyncClient", return_value=_FakeAsyncClient(response)):
+        result = await _service()._stream_service_started(
+            url="http://svc/invoke",
+            credentials="Secret tok",
+            payload={},
+            run_id="run-x",
+            strict_first_record=True,
+        )
+    assert result.accepted is True
+    assert result.run_id == "run-x"
+    assert response.consumed == 1
+
+
+async def test_stream_service_started_reports_a_runner_refusal_verbatim():
+    """Case (b) of the same browser pass, command 01a06d7a.
+
+    The runner refuses a continuation it cannot prove it owns and writes
+    ``{"kind": "result", ok: false}``. Where a deployment forwards that record verbatim the
+    caller must surface the reason, not report a start.
+    """
+    refusal = (
+        '{"kind": "result", "result": {"ok": false, "error": '
+        '"Continuation could not establish alive ownership; retry delivery."}}'
+    )
+    response = _FakeStreamResponse(lines=[refusal])
+    with patch("httpx.AsyncClient", return_value=_FakeAsyncClient(response)):
+        with pytest.raises(WorkflowDetachedStartFailed) as failure:
+            await _service()._stream_service_started(
+                url="http://svc/invoke",
+                credentials="Secret tok",
+                payload={},
+                run_id="run-x",
+                strict_first_record=True,
+            )
+    assert "alive ownership" in str(failure.value)
+
+
+async def test_stream_service_started_reports_an_empty_stream_as_a_failed_start():
+    """The same refusal as it actually reaches the API through the SDK service.
+
+    The SDK turns the runner's ``ok: false`` result into an exception inside an ASGI response
+    whose 200 is already committed, so the service closes the stream having written nothing.
+    """
+    response = _FakeStreamResponse(lines=[])
+    with patch("httpx.AsyncClient", return_value=_FakeAsyncClient(response)):
+        with pytest.raises(WorkflowDetachedStartFailed) as failure:
+            await _service()._stream_service_started(
+                url="http://svc/invoke",
+                credentials="Secret tok",
+                payload={},
+                run_id="run-x",
+                strict_first_record=True,
+            )
+    assert "closed the stream" in str(failure.value)
+
+
 async def test_stream_service_started_accepts_success_result_record():
     response = _FakeStreamResponse(
         lines=['{"kind": "result", "result": {"ok": true}}'],