fix(mcp): persist handoff results durably to survive transport timeouts

src/cli_agent_orchestrator/api/main.py

@@ -41,9 +41,12 @@
 from cli_agent_orchestrator.backends.registry import get_backend
 from cli_agent_orchestrator.clients.database import (
     create_inbox_message,
+    delete_old_handoff_results,
+    get_handoff_result,
     get_inbox_messages,
     get_terminal_metadata,
     init_db,
+    upsert_handoff_result,
 )
 from cli_agent_orchestrator.constants import (
     ALLOWED_HOSTS,
@@ -313,6 +316,28 @@ def validate_env_var_shape(self) -> "RunStepRequest":
             )
         return self
 
+    job_id: Optional[str] = Field(
+        default=None,
+        description=(
+            "Caller-generated opaque identifier for this job (issue #447). "
+            "When present, the server persists state and result under this key "
+            "so the caller can retrieve the result via GET /handoff-results/{job_id} "
+            "if the transport times out before the response arrives. "
+            "Must be a 32-character lowercase hex string (uuid4().hex)."
+        ),
+    )
+
+    @field_validator("job_id")
+    @classmethod
+    def _validate_job_id(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        if not re.fullmatch(r"[0-9a-f]{32}", v):
+            raise ValueError(
+                "job_id must be a 32-character lowercase hex string (e.g. uuid4().hex)"
+            )
+        return v
+
 
 class RunStepResponse(BaseModel):
     """Response wrapping an ``AgentStepResult`` from ``run_agent_step``."""
@@ -1805,7 +1830,20 @@ async def run_step(
 
     The plugin registry is threaded so teardown's ``post_kill_terminal`` hooks
     fire (parity with the DELETE endpoint).
+
+    Durability (issue #447): when the caller supplies a ``job_id``, the handler
+    records state="running" at request start. On success, ``run_agent_step``
+    itself persists state="completed" BETWEEN extraction and teardown (not
+    here, and not after it returns) — the terminal being torn down is the only
+    other place the result lives, so persistence must land before that
+    happens, not merely before the HTTP response. On failure, this handler
+    persists state="error". A caller that misses the response due to an
+    MCP-transport timeout can retrieve the result via
+    ``GET /handoff-results/{job_id}`` at any point thereafter. Requests without
+    a ``job_id`` behave exactly as before (no persistence overhead).
     """
+    job_id = body.job_id  # None when the caller omits it (backward-compatible)
+
     # BR-31: for a script-tier run-step call, record the created terminal into the
     # shared ScriptRunRecord's step_states AT creation time, so U4's orphan sweep
     # can tear it down if the subprocess dies mid-call. No-op for YAML/handoff
@@ -1857,6 +1895,16 @@ def _settle_step(terminal_id: Optional[str], error: Optional[str]) -> None:
         except KeyError as e:
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
+    # Mark the job as in-progress before the long-running substrate starts.
+    # This is best-effort; a failure here must not block execution. Placed
+    # AFTER the generation fence above so a fenced-out (stale-generation)
+    # call never leaves a job_id stuck at "running" with no terminal state.
+    if job_id:
+        try:
+            upsert_handoff_result(job_id, "running")
+        except Exception:
+            logger.warning("run_step: failed to record job_id=%s as running", job_id)
+
     try:
         result = await run_agent_step(
             provider=body.provider,
@@ -1872,11 +1920,19 @@ def _settle_step(terminal_id: Optional[str], error: Optional[str]) -> None:
             registry=get_plugin_registry(request),
             env_vars=body.env_vars,
             on_terminal_created=on_terminal_created,
+            job_id=job_id,
         )
         # Success -> transition the script step RUNNING->COMPLETED (no-op for
         # non-script callers). Before building the response so a settle failure
         # is logged, not raised.
         _settle_step(result.terminal_id, None)
+
+        # NOTE (issue #447 / PR #453 review): the "completed" persist happens
+        # INSIDE run_agent_step, between extraction and teardown — not here.
+        # Persisting only after this call returns would run after the terminal
+        # (the only other copy of the result) has already been torn down,
+        # contradicting the "persist result, then tear down" requirement.
+
         return RunStepResponse(
             terminal_id=result.terminal_id,
             last_message=result.last_message,
@@ -1891,22 +1947,47 @@ def _settle_step(terminal_id: Optional[str], error: Optional[str]) -> None:
         # rather than regex-scraping the message (the future engine reads it too).
         # Transition the script step RUNNING->FAILED (no-op for non-script callers).
         _settle_step(e.terminal_id, str(e))
+        if job_id:
+            try:
+                upsert_handoff_result(
+                    job_id,
+                    "error",
+                    terminal_id=e.terminal_id,
+                    error_message=str(e),
+                )
+            except Exception:
+                logger.warning("run_step: failed to persist error result for job_id=%s", job_id)
         code = status.HTTP_502_BAD_GATEWAY if e.kind == "error" else status.HTTP_504_GATEWAY_TIMEOUT
         raise HTTPException(
             status_code=code,
             detail={"message": str(e), "kind": e.kind, "terminal_id": e.terminal_id},
         )
     except TimeoutError as e:
         _settle_step(None, str(e))
+        if job_id:
+            try:
+                upsert_handoff_result(job_id, "error", error_message=str(e))
+            except Exception:
+                logger.warning("run_step: failed to persist TimeoutError for job_id=%s", job_id)
         raise HTTPException(
             status_code=status.HTTP_504_GATEWAY_TIMEOUT,
             detail={"message": str(e), "kind": "timeout", "terminal_id": None},
         )
     except ValueError as e:
         # Unknown terminal / bad input surfaced by the terminal layer.
+        if job_id:
+            try:
+                upsert_handoff_result(job_id, "error", error_message=str(e))
+            except Exception:
+                logger.warning("run_step: failed to persist ValueError for job_id=%s", job_id)
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
     except Exception as e:
         _settle_step(None, str(e))
+        if job_id:
+            try:
+                upsert_handoff_result(job_id, "error", error_message=str(e))
+            except Exception:
+                logger.warning("run_step: failed to persist Exception for job_id=%s", job_id)
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"Failed to run step: {str(e)}",
@@ -2426,6 +2507,42 @@ async def export_graph_endpoint(
     return {"written_files": written_files, "sink": body.sink, "dest": body.dest}
 
 
+@app.get(
+    "/handoff-results/{job_id}",
+    summary="Retrieve a durable handoff step result (issue #447)",
+    description=(
+        "Returns the persisted state and result for a handoff job identified by "
+        "``job_id`` (generated by the MCP client and passed to ``POST /terminals/run-step`` "
+        "in the ``job_id`` field). Use this to recover a result that was not delivered "
+        "over the original request because the MCP transport timed out. "
+        "Possible ``state`` values: ``running`` (step still in progress), "
+        "``completed`` (result in ``last_message``), ``error`` (failure in ``error_message``)."
+    ),
+)
+async def get_handoff_result_endpoint(
+    job_id: str,
+    _scopes: List[str] = Depends(require_any_scope(SCOPE_READ, SCOPE_WRITE, SCOPE_ADMIN)),
+) -> Dict:
+    """Retrieve a durable handoff step result by job_id (issue #447).
+
+    Returns the persisted record when found; 404 when the job_id is unknown
+    (either the caller never sent a job_id, or the record has been purged by
+    the retention sweep). Scope-gated (PR #453 review finding 4): ``last_message``
+    can carry worker output (prompts/secrets), so this follows the same
+    ``require_any_scope`` posture as other content-serving GETs (``/events``,
+    ``/memory/export``) rather than staying open. A no-op when auth is
+    disabled (the default) — ``require_any_scope`` only enforces when an IdP
+    is configured.
+    """
+    record = get_handoff_result(job_id)
+    if record is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Handoff result not found for job_id '{job_id}'",
+        )
+    return record
+
+
 @app.delete("/terminals/{terminal_id}")
 async def delete_terminal(
     request: Request,

src/cli_agent_orchestrator/clients/database.py

@@ -146,6 +146,31 @@ class FlowModel(Base):
     enabled = Column(Boolean, default=True)
 
 
+class HandoffResultModel(Base):
+    """Durable record of a handoff step result (issue #447).
+
+    Written by the run-step handler BEFORE returning the HTTP response, so the
+    result survives an MCP-transport timeout.  The caller supplies a ``job_id``
+    (generated client-side so retries use the same key); the server upserts on
+    that key.
+
+    ``state``:
+      - ``"running"`` — step in progress (written at request start)
+      - ``"completed"`` — step finished successfully; ``last_message`` populated
+      - ``"error"`` — step failed; ``error_message`` populated
+    """
+
+    __tablename__ = "handoff_results"
+
+    job_id = Column(String, primary_key=True)
+    state = Column(String, nullable=False)  # "running" | "completed" | "error"
+    terminal_id = Column(String, nullable=True)
+    last_message = Column(Text, nullable=True)
+    error_message = Column(Text, nullable=True)
+    created_at = Column(DateTime(timezone=True), default=_utcnow)
+    updated_at = Column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow)
+
+
 def _ensure_db_dir() -> None:
     """Create the DB dir owner-only (0o700).
 
@@ -181,6 +206,7 @@ def init_db() -> None:
     _migrate_workflow_index()
     _migrate_workflow_run()
     _migrate_workflow_run_step()
+    _migrate_add_handoff_results()
 
 
 def _restrict_db_file_permissions() -> None:
@@ -206,6 +232,36 @@ def _restrict_db_file_permissions() -> None:
             logger.warning(f"Could not restrict DB file permissions on {path}: {e}")
 
 
+def _migrate_add_handoff_results() -> None:
+    """Create the handoff_results table on existing databases (issue #447).
+
+    ``Base.metadata.create_all`` already handles fresh databases; this
+    idempotent migration handles existing ones where the table does not
+    exist yet.  SQLite supports ``CREATE TABLE IF NOT EXISTS``, so we
+    delegate to raw SQL rather than a full schema rebuild.
+    """
+    import sqlite3
+
+    from cli_agent_orchestrator.constants import DATABASE_FILE
+
+    try:
+        with sqlite3.connect(str(DATABASE_FILE)) as conn:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS handoff_results (
+                    job_id TEXT PRIMARY KEY,
+                    state TEXT NOT NULL,
+                    terminal_id TEXT,
+                    last_message TEXT,
+                    error_message TEXT,
+                    created_at DATETIME,
+                    updated_at DATETIME
+                )
+                """)
+            conn.commit()
+    except Exception as e:
+        logger.warning(f"Migration check for handoff_results failed: {e}")
+
+
 def _migrate_project_aliases_schema() -> None:
     """Rebuild project_aliases if it predates the alias-only primary key.
 
@@ -827,6 +883,86 @@ def list_aliases_for_project(project_id: str) -> List[Dict[str, Any]]:
         return []
 
 
+# ---------------------------------------------------------------------------
+# Handoff result durability helpers (issue #447)
+# ---------------------------------------------------------------------------
+
+
+def upsert_handoff_result(
+    job_id: str,
+    state: str,
+    *,
+    terminal_id: Optional[str] = None,
+    last_message: Optional[str] = None,
+    error_message: Optional[str] = None,
+) -> None:
+    """Create or update the durable record for a handoff step (issue #447).
+
+    Called at two points in the run-step handler:
+    1. Request start — ``state="running"`` (marks the job as in-progress so a
+       concurrent duplicate knows to wait rather than start a second worker).
+    2. Before sending the HTTP response — ``state="completed"`` or ``"error"``
+       (makes the result retrievable even if the transport closes before the
+       response arrives).
+
+    Idempotent: a second call for the same ``job_id`` updates the existing row.
+    """
+    now = _utcnow()
+    with SessionLocal() as db:
+        row = db.query(HandoffResultModel).filter(HandoffResultModel.job_id == job_id).first()
+        if row is None:
+            row = HandoffResultModel(
+                job_id=job_id,
+                state=state,
+                terminal_id=terminal_id,
+                last_message=last_message,
+                error_message=error_message,
+                created_at=now,
+                updated_at=now,
+            )
+            db.add(row)
+        else:
+            row.state = state
+            if terminal_id is not None:
+                row.terminal_id = terminal_id
+            if last_message is not None:
+                row.last_message = last_message
+            if error_message is not None:
+                row.error_message = error_message
+            row.updated_at = now
+        db.commit()
+
+
+def get_handoff_result(job_id: str) -> Optional[dict]:
+    """Return the handoff result record for ``job_id``, or None if not found."""
+    with SessionLocal() as db:
+        row = db.query(HandoffResultModel).filter(HandoffResultModel.job_id == job_id).first()
+        if row is None:
+            return None
+        return {
+            "job_id": row.job_id,
+            "state": row.state,
+            "terminal_id": row.terminal_id,
+            "last_message": row.last_message,
+            "error_message": row.error_message,
+            "created_at": row.created_at,
+            "updated_at": row.updated_at,
+        }
+
+
+def delete_old_handoff_results(cutoff: datetime) -> int:
+    """Delete handoff result rows older than ``cutoff`` (retention sweep).
+
+    Returns the number of rows deleted.
+    """
+    with SessionLocal() as db:
+        deleted = (
+            db.query(HandoffResultModel).filter(HandoffResultModel.created_at < cutoff).delete()
+        )
+        db.commit()
+        return deleted
+
+
 def update_message_status(message_id: int, status: MessageStatus) -> bool:
     """Update message status to MessageStatus.DELIVERED or MessageStatus.FAILED."""
     with SessionLocal() as db:

src/cli_agent_orchestrator/mcp_server/models.py

@@ -6,9 +6,31 @@
 
 
 class HandoffResult(BaseModel):
-    """Result of a handoff operation."""
+    """Result of a handoff operation.
+
+    When the MCP transport times out before the step completes (issue #447),
+    ``pending=True`` and ``job_id`` identify the in-flight job.  The caller
+    can retrieve the final result later via ``GET /handoff-results/{job_id}``.
+    """
 
     success: bool = Field(description="Whether the handoff was successful")
     message: str = Field(description="A message describing the result of the handoff")
     output: Optional[str] = Field(None, description="The output from the target agent")
     terminal_id: Optional[str] = Field(None, description="The terminal ID used for the handoff")
+    # Async-retrieval fields (issue #447). Present only when the transport timed
+    # out before the result arrived — absent (None) on the normal synchronous path.
+    job_id: Optional[str] = Field(
+        None,
+        description=(
+            "Opaque identifier for this handoff job. Present when pending=True. "
+            "Use with GET /handoff-results/{job_id} to retrieve the result later."
+        ),
+    )
+    pending: Optional[bool] = Field(
+        None,
+        description=(
+            "True when the transport timed out before the result was delivered. "
+            "The job is still running (or completed) server-side; poll "
+            "GET /handoff-results/{job_id} until state='completed'."
+        ),
+    )