Agent services, agent-tagged routes, and the gate answer receipt

backend/druks/api/app.py

@@ -18,6 +18,8 @@
 from druks.events.routes import router as events_router
 from druks.extensions.loader import iter_extensions, load
 from druks.mcp.catalog import load_mcp_catalog
+from druks.mcp.gateway.exceptions import AgentApiError
+from druks.mcp.gateway.routes import router as gateway_router
 from druks.mcp.routes import router as mcp_router
 from druks.notifications.routes import external_router as notifications_external_router
 from druks.notifications.routes import router as notifications_router
@@ -128,6 +130,16 @@ async def _http_exception_handler(request: Request, exc: HTTPException) -> JSONR
     )
 
 
+# The agent surface's one error shape. Messages are authored for the caller —
+# the handler never serializes tracebacks or internals.
+@app.exception_handler(AgentApiError)
+async def _agent_api_error_handler(request: Request, exc: AgentApiError) -> JSONResponse:
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"code": exc.code, "message": str(exc), "retryable": exc.retryable},
+    )
+
+
 @app.exception_handler(RequestValidationError)
 async def _validation_exception_handler(
     request: Request,
@@ -185,6 +197,7 @@ async def _unhandled_exception_handler(
 app.include_router(notifications_router, dependencies=_session_gate)
 app.include_router(events_router, dependencies=_session_gate)
 app.include_router(runs_router, dependencies=_session_gate)
+app.include_router(gateway_router, dependencies=_session_gate)
 app.include_router(artifacts_router, dependencies=_session_gate)
 load(app)
 

backend/druks/build/contracts.py

@@ -50,14 +50,17 @@ def to_result(self) -> dict[str, Any]:
 
 
 class QuestionOptionOutput(AgentOutput):
-    id: str
-    label: str
+    # Identity and cardinality are contract-capped at the agent boundary so a
+    # parked ask (and the gate view built from it) is bounded by construction —
+    # an oversized answer fails parse loudly instead of being clipped later.
+    id: str = Field(max_length=64)
+    label: str = Field(max_length=256)
 
 
 class QuestionOutput(AgentOutput):
-    id: str
-    prompt: str
-    options: list[QuestionOptionOutput]
+    id: str = Field(max_length=64)
+    prompt: str = Field(max_length=2048)
+    options: list[QuestionOptionOutput] = Field(max_length=16)
 
 
 class AcceptanceCriterionOutput(AgentOutput):
@@ -95,7 +98,7 @@ def get_answered(self, picks: dict[str, str]) -> list[dict[str, str]]:
 class PlanOutput(AgentOutput):
     plan_markdown: str
     acceptance_criteria: list[AcceptanceCriterionOutput]
-    questions: list[QuestionOutput]
+    questions: list[QuestionOutput] = Field(max_length=8)
 
     def get_artifact(self) -> dict[str, str]:
         return {"kind": "markdown", "title": "Implementation plan", "content": self.plan_markdown}

backend/druks/build/schemas.py

@@ -102,6 +102,11 @@ class Links(BaseResponse):
     pr: str | None = None
     ticket: str | None = None
 
+    @classmethod
+    def from_work_item(cls, item: WorkItem) -> "Links":
+        pr = f"https://github.com/{item.repo}/pull/{item.pr_number}" if item.pr_number else None
+        return cls(repo=f"https://github.com/{item.repo}", pr=pr, ticket=item.remote_url)
+
 
 class WorkItemSummary(SubjectSummary):
     # The work item's domain header — what only build knows. Status (where it is
@@ -124,8 +129,6 @@ class WorkItemSummary(SubjectSummary):
 
     @classmethod
     def from_work_item(cls, item: WorkItem) -> "WorkItemSummary":
-        repo_url = f"https://github.com/{item.repo}"
-        pr_url = f"https://github.com/{item.repo}/pull/{item.pr_number}" if item.pr_number else None
         return cls(
             id=str(item.id),
             source=item.source,  # type: ignore[arg-type]
@@ -138,7 +141,7 @@ def from_work_item(cls, item: WorkItem) -> "WorkItemSummary":
             branch=item.branch,
             created_at=item.created_at,
             updated_at=item.updated_at,
-            links=Links(repo=repo_url, pr=pr_url, ticket=item.remote_url),
+            links=Links.from_work_item(item),
         )
 
 
@@ -161,8 +164,6 @@ class DashboardItem(BaseResponse):
     @classmethod
     def from_work_item(cls, item: WorkItem) -> "DashboardItem":
         assert item.status is not None  # History is terminal-only: status is always set.
-        repo_url = f"https://github.com/{item.repo}"
-        pr_url = f"https://github.com/{item.repo}/pull/{item.pr_number}" if item.pr_number else None
         label, outcome = _outcome_from_status(item.status)
         return cls(
             key=f"code:{item.id}",
@@ -178,7 +179,7 @@ def from_work_item(cls, item: WorkItem) -> "DashboardItem":
             outcome=outcome,
             created_at=item.created_at,
             updated_at=item.updated_at,
-            links=Links(repo=repo_url, pr=pr_url, ticket=item.remote_url),
+            links=Links.from_work_item(item),
         )
 
 

backend/druks/durable/models.py

@@ -41,6 +41,8 @@ class Run(Base):
     # When the run last asked for input — a historical fact (input_gate says
     # whether it's still waiting), and the one transition DBOS doesn't stamp.
     input_requested_at: Mapped[datetime | None] = mapped_column(default=None)
+    # The receipt: the input_requested_at stamp an answer last resumed.
+    answer_parked_at: Mapped[datetime | None] = mapped_column(default=None)
     failure: Mapped[str | None] = mapped_column(default=None)
     # The FatalError subtype's code when the run stopped on a deliberate domain
     # error, so read-sides tell e.g. a gate timeout from a crash without parsing
@@ -250,6 +252,14 @@ class AgentCall(Base, Uuid7Pk):
     # and treat 404 as the host being gone.
     sandbox_host_id: Mapped[str]
 
+    def get_live_status(self) -> str:
+        # Unfinished: "running" while the run is live, "abandoned" once it's terminal.
+        if self.finished_at:
+            return AgentCallStatus(self.status).value
+        if self.run.is_active:
+            return "running"
+        return "abandoned"
+
     @property
     def artifact_dir(self) -> str:
         return str(load_settings().artifacts_dir / f"run-{self.run_id}")
@@ -279,7 +289,7 @@ def get_file_path(self, name: str) -> Path | None:
         try:
             candidate.relative_to(self.call_dir.resolve())
         except ValueError:
-            return None
+            return
         return candidate if candidate.is_file() else None
 
     def get_stream_path(self, stream: Literal["stdout", "stderr"]) -> Path | None:

backend/druks/durable/reads.py

@@ -2,6 +2,7 @@
 # Schemas stay pure projections; routes call in here.
 import asyncio
 from collections.abc import AsyncIterator
+from pathlib import Path
 from typing import Literal
 
 from sqlalchemy import Engine
@@ -19,6 +20,7 @@
     SubjectResponse,
     SubjectStatus,
     SubjectSummary,
+    TextSlice,
     TranscriptChunk,
 )
 
@@ -30,7 +32,7 @@
 def get_agent_call_files(call_id: str) -> AgentCallFiles | None:
     call = AgentCall.get(call_id)
     if not call:
-        return None
+        return
     return AgentCallFiles.from_call(call, Artifact.get_for_call(call.id))
 
 
@@ -113,6 +115,28 @@ def _status(
     )
 
 
+def read_slice(path: Path, *, offset: int, limit: int) -> TextSlice:
+    # A bounded byte window of a text file. A multibyte character split at a
+    # window seam shows one � — the live tail below does the same. Negative
+    # offset reads the tail; missing file is an empty eof slice.
+    if not path.exists():
+        return TextSlice(offset=0, next_offset=0, eof=True, has_earlier=False, text="")
+    size = path.stat().st_size
+    if offset < 0:
+        offset = max(size + offset, 0)
+    with path.open("rb") as handle:
+        handle.seek(offset)
+        payload = handle.read(limit)
+    next_offset = offset + len(payload)
+    return TextSlice(
+        offset=offset,
+        next_offset=next_offset,
+        eof=next_offset >= size,
+        has_earlier=offset > 0,
+        text=payload.decode("utf-8", errors="replace"),
+    )
+
+
 def read_transcript_chunk(
     engine: Engine,
     call_id: str,
@@ -127,23 +151,20 @@ def read_transcript_chunk(
     with session_scope(engine):
         call = AgentCall.get(call_id)
         if not call:
-            return None
+            return
         path = call.get_stream_path(stream)
-    if not path or not path.exists():
+    if not path:
         return TranscriptChunk(
             call_id=call_id, stream=stream, offset=offset, next_offset=offset, eof=True, text=""
         )
-    with path.open("rb") as handle:
-        handle.seek(offset)
-        payload = handle.read(limit)
-    next_offset = offset + len(payload)
+    piece = read_slice(path, offset=offset, limit=limit)
     return TranscriptChunk(
         call_id=call_id,
         stream=stream,
-        offset=offset,
-        next_offset=next_offset,
-        eof=next_offset >= path.stat().st_size,
-        text=payload.decode("utf-8", errors="replace"),
+        offset=piece.offset,
+        next_offset=piece.next_offset,
+        eof=piece.eof,
+        text=piece.text,
     )
 
 

backend/druks/durable/schemas.py

@@ -7,7 +7,7 @@
 from druks.harnesses.artifacts import normalize_token_usage
 from druks.schemas import BaseResponse
 
-from .enums import AgentCallStatus, RunState
+from .enums import RunState
 from .models import AgentCall, Artifact, Run
 
 
@@ -48,21 +48,12 @@ class AgentCallResponse(BaseResponse):
 
     @classmethod
     def from_call(cls, call: AgentCall) -> "AgentCallResponse":
-        # Liveness is derived, not stored: an unfinished call reads "running" while its
-        # run is active and "abandoned" once the run is terminal (the run ended without
-        # the call closing). A finished call keeps its recorded outcome.
-        if call.finished_at:
-            status = AgentCallStatus(call.status).value
-        elif call.run.is_active:
-            status = "running"
-        else:
-            status = "abandoned"
         return cls(
             id=call.id,
             agent=call.agent,
             label=get_display_label(call.agent) if call.agent else "Agent",
             account_username=call.account.username,
-            status=status,  # type: ignore[arg-type]
+            status=call.get_live_status(),  # type: ignore[arg-type]
             started_at=call.started_at,
             finished_at=call.finished_at,
             last_error=call.last_error,
@@ -107,7 +98,7 @@ def from_call(cls, call: AgentCall, artifact: Artifact | None) -> "AgentCallFile
 
         def named(path: Path) -> ArtifactFile | None:
             if not path.is_file():
-                return None
+                return
             stat = path.stat()
             return ArtifactFile(
                 name=path.name,
@@ -221,3 +212,13 @@ class TranscriptChunk(BaseResponse):
     next_offset: int
     eof: bool
     text: str
+
+
+class TextSlice(BaseResponse):
+    # One bounded UTF-8-safe cut of an on-disk text file; offsets are byte
+    # positions, has_earlier marks content before this slice.
+    offset: int
+    next_offset: int
+    eof: bool
+    has_earlier: bool
+    text: str

backend/druks/mcp/gateway/exceptions.py

@@ -0,0 +1,67 @@
+from typing import ClassVar
+
+
+class AgentApiError(Exception):
+    # Base for the agent surface's wire errors: each subclass names its HTTP
+    # status and stable code, serialized as the one {code, message, retryable}
+    # response shape. Messages are authored for the caller — never tracebacks,
+    # paths, or engine internals. retryable=True marks a failure the caller can
+    # fix by re-reading state and retrying.
+    status_code: ClassVar[int] = 400
+    code: ClassVar[str] = ""
+    retryable: ClassVar[bool] = False
+
+
+class RunNotFound(AgentApiError):
+    status_code = 404
+    code = "RUN_NOT_FOUND"
+
+    def __init__(self, run_id: str) -> None:
+        super().__init__(f"No run {run_id}.")
+
+
+class GateNotOpen(AgentApiError):
+    status_code = 409
+    code = "GATE_NOT_OPEN"
+
+    def __init__(self, run_id: str) -> None:
+        super().__init__(f"Run {run_id} is not parked on a gate.")
+
+
+class GateRoundStale(AgentApiError):
+    status_code = 409
+    code = "GATE_ROUND_STALE"
+    retryable = True
+
+    def __init__(self, run_id: str) -> None:
+        super().__init__(
+            f"Run {run_id} has re-parked since the parked_at you read; fetch the gate again."
+        )
+
+
+class GateNotAnswerable(AgentApiError):
+    status_code = 409
+    code = "GATE_NOT_ANSWERABLE"
+
+    def __init__(self, run_id: str) -> None:
+        super().__init__(f"The gate on run {run_id} is answered on its source, not through here.")
+
+
+class InvalidGateAnswer(AgentApiError):
+    code = "INVALID_GATE_ANSWER"
+
+
+class AgentCallNotFound(AgentApiError):
+    status_code = 404
+    code = "AGENT_CALL_NOT_FOUND"
+
+    def __init__(self, call_id: str) -> None:
+        super().__init__(f"No agent call {call_id}.")
+
+
+class RunNotActive(AgentApiError):
+    status_code = 409
+    code = "RUN_NOT_ACTIVE"
+
+    def __init__(self, run_id: str) -> None:
+        super().__init__(f"Run {run_id} already ended.")