diff --git a/backend/druks/api/app.py b/backend/druks/api/app.py
index 39c969d..0795815 100644
--- a/backend/druks/api/app.py
+++ b/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)
 
diff --git a/backend/druks/build/contracts.py b/backend/druks/build/contracts.py
index 5bcaeda..173f584 100644
--- a/backend/druks/build/contracts.py
+++ b/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}
diff --git a/backend/druks/build/schemas.py b/backend/druks/build/schemas.py
index ed96fdd..95c3eae 100644
--- a/backend/druks/build/schemas.py
+++ b/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),
         )
 
 
diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py
index 08fd80f..4529472 100644
--- a/backend/druks/durable/models.py
+++ b/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:
diff --git a/backend/druks/durable/reads.py b/backend/druks/durable/reads.py
index c5624c7..aa45309 100644
--- a/backend/druks/durable/reads.py
+++ b/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,
     )
 
 
diff --git a/backend/druks/durable/schemas.py b/backend/druks/durable/schemas.py
index 2df3e80..917dd5f 100644
--- a/backend/druks/durable/schemas.py
+++ b/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
diff --git a/backend/druks/mcp/gateway/__init__.py b/backend/druks/mcp/gateway/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/druks/mcp/gateway/exceptions.py b/backend/druks/mcp/gateway/exceptions.py
new file mode 100644
index 0000000..4477bd6
--- /dev/null
+++ b/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.")
diff --git a/backend/druks/mcp/gateway/routes.py b/backend/druks/mcp/gateway/routes.py
new file mode 100644
index 0000000..1a46a95
--- /dev/null
+++ b/backend/druks/mcp/gateway/routes.py
@@ -0,0 +1,67 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Body, Depends
+
+from druks.accounts.dependencies import current_account
+from druks.accounts.models import Account
+from druks.mcp.gateway import schemas, services
+
+router = APIRouter(prefix="/mcp", tags=["mcp"])
+
+
+@router.get(
+    "/gates/{run_id}",
+    operation_id="get_gate",
+    response_model=schemas.GateResponse,
+    response_model_by_alias=True,
+)
+async def get_gate(run_id: str) -> schemas.GateResponse:
+    return services.get_gate(run_id)
+
+
+@router.post(
+    "/gates/{run_id}/answer",
+    operation_id="answer_gate",
+    response_model=schemas.GateAnswerResponse,
+    response_model_by_alias=True,
+)
+async def answer_gate(run_id: str, body: schemas.AnswerGateRequest) -> schemas.GateAnswerResponse:
+    return await services.answer_gate(
+        run_id,
+        parked_at=body.parked_at,
+        control=body.control,
+        answers=body.answers,
+        note=body.note,
+    )
+
+
+@router.get(
+    "/agent-calls/{call_id}",
+    operation_id="get_agent_call",
+    response_model=schemas.AgentCallDetailResponse,
+    response_model_by_alias=True,
+)
+async def get_agent_call(call_id: str) -> schemas.AgentCallDetailResponse:
+    return services.get_agent_call(call_id)
+
+
+@router.post(
+    "/runs/{run_id}/cancel",
+    operation_id="cancel_run",
+    response_model=schemas.CancelRunResponse,
+    response_model_by_alias=True,
+)
+async def cancel_run(
+    run_id: str, reason: Annotated[str, Body(embed=True, min_length=1, max_length=500)]
+) -> schemas.CancelRunResponse:
+    return await services.cancel_run(run_id, reason=reason)
+
+
+@router.get(
+    "/usage",
+    operation_id="get_usage",
+    response_model=schemas.AgentUsageResponse,
+    response_model_by_alias=True,
+)
+async def get_usage(account: Account = Depends(current_account)) -> schemas.AgentUsageResponse:
+    return services.get_usage(account)
diff --git a/backend/druks/mcp/gateway/schemas.py b/backend/druks/mcp/gateway/schemas.py
new file mode 100644
index 0000000..95e6ffe
--- /dev/null
+++ b/backend/druks/mcp/gateway/schemas.py
@@ -0,0 +1,77 @@
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
+
+from druks.durable.schemas import AgentCallResponse
+from druks.schemas import BaseResponse
+from druks.usage.schemas import UsageHistoryPoint
+
+
+class ArtifactContent(BaseResponse):
+    call_id: str
+    kind: str
+    title: str
+    content: str
+
+
+class GateResponse(BaseResponse):
+    # parked_at is the park identity answer_gate must echo back.
+    run_id: str
+    gate: str
+    parked_at: datetime
+    ask: dict[str, Any]
+    artifact: ArtifactContent | None = None
+
+
+class GateAnswerResponse(BaseResponse):
+    run_id: str
+    parked_at: datetime
+    result: Literal["answered", "already_answered"]
+
+
+class AnswerGateRequest(BaseModel):
+    # parked_at echoes get_gate's value unchanged.
+    model_config = ConfigDict(str_strip_whitespace=True, alias_generator=to_camel)
+    parked_at: AwareDatetime
+    control: str
+    answers: dict[str, str] = Field(default_factory=dict)
+    note: str = ""
+
+
+class AgentCallDetailResponse(BaseResponse):
+    run_id: str
+    call: AgentCallResponse
+    transcript: str
+    stderr: str
+    artifact: ArtifactContent | None = None
+
+
+class CancelRunResponse(BaseResponse):
+    run_id: str
+    result: Literal["cancelled", "already_cancelled"]
+
+
+class AgentHarnessUsage(BaseResponse):
+    # *_history: percent-left trend samples, oldest first.
+    name: str
+    is_connected: bool = False
+    plan_tier: str | None = None
+    five_hour_percent_left: int | None = None
+    five_hour_resets_at: datetime | None = None
+    week_percent_left: int | None = None
+    week_resets_at: datetime | None = None
+    is_unlimited: bool = False
+    scraped_at: datetime | None = None
+    five_hour_history: list[UsageHistoryPoint] = Field(default_factory=list)
+    week_history: list[UsageHistoryPoint] = Field(default_factory=list)
+
+
+class AgentUsageResponse(BaseResponse):
+    day: str
+    timezone: str
+    spend_today_usd: float
+    tokens_today: int
+    runs_today: int
+    harnesses: list[AgentHarnessUsage] = Field(default_factory=list)
diff --git a/backend/druks/mcp/gateway/services.py b/backend/druks/mcp/gateway/services.py
new file mode 100644
index 0000000..6ab2302
--- /dev/null
+++ b/backend/druks/mcp/gateway/services.py
@@ -0,0 +1,166 @@
+from datetime import UTC, datetime, timedelta
+
+from druks.accounts.models import Account
+from druks.core.utils.time import operator_local_day
+from druks.database import db_session
+from druks.durable.enums import RunState
+from druks.durable.models import AgentCall, Artifact, Run
+from druks.durable.reads import read_slice
+from druks.durable.schemas import AgentCallResponse
+from druks.harnesses.artifacts import normalize_token_usage
+from druks.harnesses.models import HarnessConnection
+from druks.harnesses.registry import get_harnesses
+from druks.mcp.gateway import exceptions, schemas
+from druks.notifications.exceptions import InvalidChoiceError
+from druks.notifications.services import validate_in_app_answer
+from druks.usage.models import UsageScrape
+from druks.usage.reads import list_finished_calls
+from druks.usage.schemas import UsageHistoryPoint
+from druks.usage.trends import FIVE_HOUR_RANGE, WEEK_RANGE, downsample
+from druks.user_settings.models import UserSettings
+
+_TRANSCRIPT_TAIL_BYTES = 8 * 1024
+_STDERR_TAIL_BYTES = 4 * 1024
+_ARTIFACT_CHUNK_BYTES = 4 * 1024
+_HISTORY_POINTS = 8
+
+
+def get_gate(run_id: str) -> schemas.GateResponse:
+    run = Run.get(run_id)
+    if not run:
+        raise exceptions.RunNotFound(run_id)
+    if run.state != RunState.PENDING_INPUT.value:
+        raise exceptions.GateNotOpen(run_id)
+    ask = run.input_request
+    if not ask or ask.get("presentation") != "in_app":
+        raise exceptions.GateNotAnswerable(run_id)
+    return schemas.GateResponse(
+        run_id=run.id,
+        gate=run.input_gate,  # type: ignore[arg-type]
+        parked_at=run.input_requested_at,  # type: ignore[arg-type]
+        ask=run.get_ask(),
+        artifact=_artifact_content(Artifact.get_latest_for_run(run.id)),
+    )
+
+
+async def answer_gate(
+    run_id: str, *, parked_at: datetime, control: str, answers: dict[str, str], note: str
+) -> schemas.GateAnswerResponse:
+    run = Run.get(run_id)
+    if not run:
+        raise exceptions.RunNotFound(run_id)
+    db_session().expire(run)  # the receipt/park comparison must read fresh
+    if run.answer_parked_at == parked_at:
+        return schemas.GateAnswerResponse(
+            run_id=run.id, parked_at=parked_at, result="already_answered"
+        )
+    if run.state != RunState.PENDING_INPUT.value:
+        raise exceptions.GateNotOpen(run_id)
+    if run.input_requested_at != parked_at:
+        raise exceptions.GateRoundStale(run_id)
+    ask = run.input_request
+    if not ask or ask.get("presentation") != "in_app":
+        raise exceptions.GateNotAnswerable(run_id)
+    try:
+        payload = validate_in_app_answer(run.get_ask(), control, answers, note)
+    except InvalidChoiceError as error:
+        raise exceptions.InvalidGateAnswer(str(error)) from error
+    await run.resume(**payload)
+    return schemas.GateAnswerResponse(run_id=run.id, parked_at=parked_at, result="answered")
+
+
+def get_agent_call(call_id: str) -> schemas.AgentCallDetailResponse:
+    call = AgentCall.get(call_id)
+    if not call:
+        raise exceptions.AgentCallNotFound(call_id)
+    layout = call.artifact_layout
+    return schemas.AgentCallDetailResponse(
+        run_id=call.run_id,
+        call=AgentCallResponse.from_call(call),
+        transcript=read_slice(
+            layout.transcript, offset=-_TRANSCRIPT_TAIL_BYTES, limit=_TRANSCRIPT_TAIL_BYTES
+        ).text,
+        stderr=read_slice(layout.stderr, offset=-_STDERR_TAIL_BYTES, limit=_STDERR_TAIL_BYTES).text,
+        artifact=_artifact_content(Artifact.get_for_call(call.id)),
+    )
+
+
+async def cancel_run(run_id: str, *, reason: str) -> schemas.CancelRunResponse:
+    run = Run.get(run_id)
+    if not run:
+        raise exceptions.RunNotFound(run_id)
+    if run.state == RunState.CANCELLED.value:
+        return schemas.CancelRunResponse(run_id=run.id, result="already_cancelled")
+    if not run.is_active:
+        raise exceptions.RunNotActive(run_id)
+    await run.cancel(failure=reason)
+    return schemas.CancelRunResponse(run_id=run.id, result="cancelled")
+
+
+def _artifact_content(artifact: Artifact | None) -> schemas.ArtifactContent | None:
+    if not artifact:
+        return
+    call = AgentCall.get(artifact.agent_call_id)
+    path = call.get_file_path(artifact.path) if call else None
+    if not path:
+        return
+    return schemas.ArtifactContent(
+        call_id=artifact.agent_call_id,
+        kind=artifact.kind,
+        title=artifact.title,
+        content=read_slice(path, offset=0, limit=_ARTIFACT_CHUNK_BYTES).text,
+    )
+
+
+def get_usage(account: Account) -> schemas.AgentUsageResponse:
+    now = datetime.now(UTC)
+    timezone, local_start = operator_local_day(UserSettings.get().timezone, now)
+    rows = list_finished_calls(account.id, since=local_start, until=local_start + timedelta(days=1))
+    spend = 0.0
+    tokens = 0
+    for _, cost_usd, cost_metadata, _ in rows:
+        if cost_usd is not None:
+            spend += float(cost_usd)
+        usage = normalize_token_usage(cost_metadata)
+        if usage:
+            tokens += usage["total_tokens"]
+    return schemas.AgentUsageResponse(
+        day=local_start.date().isoformat(),
+        timezone=str(timezone),
+        spend_today_usd=round(spend, 4),
+        tokens_today=tokens,
+        runs_today=len(rows),
+        harnesses=[_harness_usage(h.name, account.id, now=now) for h in get_harnesses()],
+    )
+
+
+def _harness_usage(name: str, account_id: str, *, now: datetime) -> schemas.AgentHarnessUsage:
+    is_connected = bool(HarnessConnection.get_for_account(name, account_id))
+    row = UsageScrape.latest_for(name, account_id)
+    if not row:
+        return schemas.AgentHarnessUsage(name=name, is_connected=is_connected)
+    history = UsageScrape.history_for(name, account_id, since=now - WEEK_RANGE)
+    five_hour_cutoff = now - FIVE_HOUR_RANGE
+    five_hour = [
+        UsageHistoryPoint(t=point.scraped_at, pct=point.five_hour_percent_left)
+        for point in history
+        if point.five_hour_percent_left is not None and point.scraped_at >= five_hour_cutoff
+    ]
+    week = [
+        UsageHistoryPoint(t=point.scraped_at, pct=point.week_percent_left)
+        for point in history
+        if point.week_percent_left is not None
+    ]
+    return schemas.AgentHarnessUsage(
+        name=name,
+        is_connected=is_connected,
+        plan_tier=row.plan_tier,
+        five_hour_percent_left=row.five_hour_percent_left,
+        five_hour_resets_at=row.five_hour_resets_at,
+        week_percent_left=row.week_percent_left,
+        week_resets_at=row.week_resets_at,
+        is_unlimited=row.unlimited,
+        scraped_at=row.scraped_at,
+        five_hour_history=downsample(five_hour, cap=_HISTORY_POINTS),
+        week_history=downsample(week, cap=_HISTORY_POINTS),
+    )
diff --git a/backend/druks/usage/reads.py b/backend/druks/usage/reads.py
new file mode 100644
index 0000000..9f56e5a
--- /dev/null
+++ b/backend/druks/usage/reads.py
@@ -0,0 +1,25 @@
+from datetime import datetime
+
+from sqlalchemy import Row, select
+
+from druks.database import db_session
+from druks.durable.models import AgentCall
+
+
+def list_finished_calls(account_id: str, *, since: datetime, until: datetime) -> list[Row]:
+    return list(
+        db_session()
+        .execute(
+            select(
+                AgentCall.model,
+                AgentCall.cost_usd,
+                AgentCall.cost_metadata,
+                AgentCall.finished_at,
+            )
+            .where(AgentCall.account_id == account_id)
+            .where(AgentCall.finished_at.is_not(None))
+            .where(AgentCall.finished_at >= since)
+            .where(AgentCall.finished_at < until)
+        )
+        .all()
+    )
diff --git a/backend/druks/usage/routes.py b/backend/druks/usage/routes.py
index 05761c4..1ae11a5 100644
--- a/backend/druks/usage/routes.py
+++ b/backend/druks/usage/routes.py
@@ -1,16 +1,15 @@
 from datetime import UTC, datetime, timedelta
 
 from fastapi import APIRouter, Depends
-from sqlalchemy import select
 
 from druks.accounts.dependencies import current_account
 from druks.accounts.models import Account
 from druks.core.utils.time import operator_local_day
-from druks.db import db_session
 from druks.harnesses.artifacts import normalize_token_usage
 from druks.harnesses.models import HarnessConnection
 from druks.harnesses.registry import get_harnesses
 from druks.usage.models import UsageScrape
+from druks.usage.reads import list_finished_calls
 from druks.usage.schemas import (
     UsageHarnessHistory,
     UsageHarnessSummary,
@@ -21,8 +20,8 @@
     UsageResponse,
     UsageTodayResponse,
 )
+from druks.usage.trends import FIVE_HOUR_RANGE, WEEK_RANGE, downsample
 from druks.user_settings.models import HarnessSettings, UserSettings
-from druks.workflows import AgentCall
 
 router = APIRouter(tags=["usage"])
 
@@ -38,12 +37,7 @@
 # default 5-min poll cadence.
 _STALE_AFTER_SECONDS = 24 * 60 * 60
 
-# Trend ranges for the usage page sparklines. The 5h window gets one
-# full window plus headroom so an exhaustion arc is visible end to end;
-# weekly gets the whole week. Both are downsampled to keep the payload
-# flat regardless of poll cadence.
-_FIVE_HOUR_RANGE = timedelta(hours=6)
-_WEEK_RANGE = timedelta(days=7)
+# The dashboard sparklines keep this many points regardless of poll cadence.
 _MAX_SPARK_POINTS = 72
 
 
@@ -96,28 +90,11 @@ async def get_usage_history(account: Account = Depends(current_account)) -> Usag
     response_model_by_alias=True,
 )
 async def get_usage_today(account: Account = Depends(current_account)) -> UsageTodayResponse:
-    # The shared operator-local-day boundary keeps this total identical to
-    # the sys-strip's spend-today figure.
-    timezone_name = UserSettings.get().timezone
-    now = datetime.now(UTC)
-    timezone, local_start = operator_local_day(timezone_name, now)
+    # Deriving the operator-local-day window here (the query just takes it) keeps
+    # this total identical to the sys-strip's and the agent surface's figures.
+    timezone, local_start = operator_local_day(UserSettings.get().timezone, datetime.now(UTC))
+    rows = list_finished_calls(account.id, since=local_start, until=local_start + timedelta(days=1))
     timezone_name = str(timezone)
-    rows = (
-        db_session()
-        .execute(
-            select(
-                AgentCall.model,
-                AgentCall.cost_usd,
-                AgentCall.cost_metadata,
-                AgentCall.finished_at,
-            )
-            .where(AgentCall.account_id == account.id)
-            .where(AgentCall.finished_at.is_not(None))
-            .where(AgentCall.finished_at >= local_start.astimezone(UTC))
-            .where(AgentCall.finished_at < (local_start + timedelta(days=1)).astimezone(UTC)),
-        )
-        .all()
-    )
 
     # Every call counts, even one whose model no picker list claims (pinned
     # outside the list, or never resolved): money spent must not vanish from
@@ -162,8 +139,8 @@ async def get_usage_today(account: Account = Depends(current_account)) -> UsageT
 
 
 def _harness_history(name: str, account_id: str, *, now: datetime) -> UsageHarnessHistory:
-    rows = UsageScrape.history_for(name, account_id, since=now - _WEEK_RANGE)
-    five_hour_cutoff = now - _FIVE_HOUR_RANGE
+    rows = UsageScrape.history_for(name, account_id, since=now - WEEK_RANGE)
+    five_hour_cutoff = now - FIVE_HOUR_RANGE
     five_hour = [
         UsageHistoryPoint(t=row.scraped_at, pct=row.five_hour_percent_left)
         for row in rows
@@ -176,23 +153,11 @@ def _harness_history(name: str, account_id: str, *, now: datetime) -> UsageHarne
     ]
     return UsageHarnessHistory(
         name=name,
-        five_hour=_downsample(five_hour),
-        week=_downsample(week),
+        five_hour=downsample(five_hour, cap=_MAX_SPARK_POINTS),
+        week=downsample(week, cap=_MAX_SPARK_POINTS),
     )
 
 
-def _downsample(points: list[UsageHistoryPoint]) -> list[UsageHistoryPoint]:
-    """Thin a series to ≤ _MAX_SPARK_POINTS, always keeping the newest
-    sample (the page's "now" anchor)."""
-    if len(points) <= _MAX_SPARK_POINTS:
-        return points
-    stride = -(-len(points) // _MAX_SPARK_POINTS)  # ceil division
-    thinned = points[::stride]
-    if thinned[-1] is not points[-1]:
-        thinned.append(points[-1])
-    return thinned
-
-
 def _summarize(
     row: UsageScrape | None,
     *,
@@ -221,11 +186,11 @@ def _summarize(
 
 def _metric(percent_left: int | None, resets_at: datetime | None) -> UsageMetricSummary | None:
     if percent_left is None and resets_at is None:
-        return None
+        return
     return UsageMetricSummary(percent_left=percent_left, resets_at=resets_at)
 
 
 def _age_seconds(scraped_at: datetime | None, *, now: datetime) -> int | None:
     if not scraped_at:
-        return None
+        return
     return max(0, int((now - scraped_at).total_seconds()))
diff --git a/backend/druks/usage/trends.py b/backend/druks/usage/trends.py
new file mode 100644
index 0000000..6b12b23
--- /dev/null
+++ b/backend/druks/usage/trends.py
@@ -0,0 +1,20 @@
+from datetime import timedelta
+
+from druks.usage.schemas import UsageHistoryPoint
+
+# Trend ranges for the percent-left sparklines. The 5h window gets one full
+# window plus headroom so an exhaustion arc is visible end to end; weekly gets
+# the whole week.
+FIVE_HOUR_RANGE = timedelta(hours=6)
+WEEK_RANGE = timedelta(days=7)
+
+
+def downsample(points: list[UsageHistoryPoint], *, cap: int) -> list[UsageHistoryPoint]:
+    # Thin a series to ≤ cap points, always keeping the newest sample (the
+    # "now" anchor) — it replaces the last strided sample so the cap holds.
+    if len(points) <= cap:
+        return points
+    stride = -(-len(points) // cap)  # ceil division
+    thinned = points[::stride]
+    thinned[-1] = points[-1]
+    return thinned
diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py
index 2cf3ff3..8c991a8 100644
--- a/backend/druks/workflows.py
+++ b/backend/druks/workflows.py
@@ -143,7 +143,7 @@ def _input_model_from_signature(cls: type["Workflow"]) -> type[BaseModel] | None
     method = getattr(cls, method_name)
     parameters = [p for name, p in inspect.signature(method).parameters.items() if name != "self"]
     if not parameters:
-        return None
+        return
     hints = get_type_hints(method)
     fields: dict[str, Any] = {}
     for p in parameters:
@@ -239,7 +239,7 @@ async def _park(
         workflow.workflow_id,
         RunState.RUNNING,
         subject=workflow.subject,
-        facts=_GATE_CLEARED,
+        facts={**_GATE_CLEARED, "answer_parked_at": Run.input_requested_at},
     )
     return payload
 
@@ -251,9 +251,8 @@ async def _notify_designated_destination(workflow_id: str, subject: dict[str, An
     async def _create() -> str | None:
         async with step_session():
             destination_id = UserSettings.get().gate_park_destination_id
-            if not destination_id:
-                return None
-            return Run.get(workflow_id).create_park_notification(destination_id, subject)
+            if destination_id:
+                return Run.get(workflow_id).create_park_notification(destination_id, subject)
 
     notification_id = await DBOS.run_step_async(
         StepOptions(name="notifications.gate_park", **_IO_RETRIES), _create
@@ -306,14 +305,13 @@ async def _transition() -> dict[str, Any] | None:
                 for field, value in facts.items():
                     setattr(run, field, value)
                 session.flush()
-            if not subject:
-                # Subjectless framework crons are plumbing: no feed entry.
-                return None
-            return {
-                "kind": run.kind,
-                "subject": subject,
-                "payload": _log_run_event(run, event, subject, result),
-            }
+            # Subjectless framework crons are plumbing: no feed entry.
+            if subject:
+                return {
+                    "kind": run.kind,
+                    "subject": subject,
+                    "payload": _log_run_event(run, event, subject, result),
+                }
 
     transition = await DBOS.run_step_async(
         StepOptions(name=f"run.{event.value}", **_IO_RETRIES), _transition
@@ -554,7 +552,7 @@ async def _ensure_host(self) -> str | None:
         # The warm VM, provisioned once per segment; state is carried in git, so
         # only the host-id matters across steps — held-across-steps never fights replay.
         if not self.steps_reuse_sandbox:
-            return None
+            return
         if self._host and self._host.expires_at:
             remaining = (self._host.expires_at - datetime.now(UTC)).total_seconds()
             if remaining < SANDBOX_HOST_ROTATE_BEFORE_SECONDS:
@@ -628,13 +626,13 @@ async def start(
         **input: Any,
     ) -> str:
         # Mint the id, write the projection row, enqueue the body. Returns the
-        # workflow id; an extension that wants one-active-run-per-subject enforces
-        # that on its own side before calling this. Enqueuing (not start_workflow)
-        # routes execution onto the shared queue, so the process that kicks a run
-        # off — often the web process — doesn't have to be the one that runs it;
-        # any launched executor picks it up. The input kwargs mirror the body's own
-        # signature and validate against the model synthesized from it, so a bad
-        # shape fails at start, not inside the run.
+        # run to follow — a freshly enqueued run, or the subject's already-live
+        # one handed back. Enqueuing (not start_workflow) routes execution
+        # onto the shared queue, so the process that kicks a run off — often the
+        # web process — doesn't have to be the one that runs it; any launched
+        # executor picks it up. The input kwargs mirror the body's own signature
+        # and validate against the model synthesized from it, so a bad shape
+        # fails at start, not inside the run.
         # subject is required (no default) so a run can't silently lose its
         # timeline by omission — pass subject=None explicitly for a background run.
         if account_id is None:
@@ -643,12 +641,11 @@ async def start(
             account_id = current_account_id.get()
         if subject is not None:
             _Subject.model_validate(subject)  # raises on a bad shape (wrong/extra keys, types)
-        if cls._run_input_model is None:
-            if input:
-                raise WorkflowError(f"{cls.__name__}.{cls._body_method}() takes no input")
-            wire: dict[str, Any] = {}
-        else:
+        wire: dict[str, Any] = {}
+        if cls._run_input_model:
             wire = cls._run_input_model.model_validate(input).model_dump(mode="json")
+        elif input:
+            raise WorkflowError(f"{cls.__name__}.{cls._body_method}() takes no input")
         if account_id:
             wire[_ACCOUNT_INPUT_KEY] = account_id
         workflow_id = str(uuid7())
diff --git a/backend/migrations/versions/ff43df27a2e0_gate_answer_receipt.py b/backend/migrations/versions/ff43df27a2e0_gate_answer_receipt.py
new file mode 100644
index 0000000..366ac5a
--- /dev/null
+++ b/backend/migrations/versions/ff43df27a2e0_gate_answer_receipt.py
@@ -0,0 +1,29 @@
+"""gate answer receipt
+
+Revision ID: ff43df27a2e0
+Revises: 0ec9db44973e
+Create Date: 2026-07-19 12:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "ff43df27a2e0"
+down_revision: str | Sequence[str] | None = "0ec9db44973e"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "durable_runs",
+        sa.Column("answer_parked_at", sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("durable_runs", "answer_parked_at")
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index c71d0f9..56eedef 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -479,6 +479,7 @@ def seed_build_run(
     input_gate: str | None = None,
     input_request: dict | None = None,
     failure: str | None = None,
+    account_id: str | None = None,
 ):
     """Seed a build Run row for a work item and bind it via
     ``item.build_run_id``. Returns the Run. Attach agent calls with
@@ -496,6 +497,7 @@ def seed_build_run(
         input_gate=input_gate,
         input_request=input_request,
         failure=failure,
+        account_id=account_id,
     )
     session.add(run)
     session.flush()
diff --git a/backend/tests/test_agent_routes.py b/backend/tests/test_agent_routes.py
new file mode 100644
index 0000000..175c5e0
--- /dev/null
+++ b/backend/tests/test_agent_routes.py
@@ -0,0 +1,270 @@
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+from conftest import (
+    configure_app_for_test,
+    make_settings,
+    make_test_work_item,
+    seed_build_run,
+    seed_run,
+)
+from druks.accounts.models import Account
+from druks.durable.models import Run
+from druks.durable.reads import read_transcript_chunk
+from druks.mcp.gateway import services
+from fastapi.testclient import TestClient
+
+_IN_APP_ASK = {
+    "presentation": "in_app",
+    "controls": ["approve", "request_changes", "cancel"],
+    "questions": [],
+}
+
+_MCP_ROUTES = {
+    ("get", "/mcp/gates/{run_id}"): "get_gate",
+    ("post", "/mcp/gates/{run_id}/answer"): "answer_gate",
+    ("get", "/mcp/agent-calls/{call_id}"): "get_agent_call",
+    ("post", "/mcp/runs/{run_id}/cancel"): "cancel_run",
+    ("get", "/mcp/usage"): "get_usage",
+}
+
+
+@pytest.fixture
+def client(tmp_path: Path, db_session, monkeypatch):
+    monkeypatch.setenv("DRUKS_DATA_DIR", str(tmp_path))
+    app = configure_app_for_test(settings=make_settings(tmp_path))
+    with TestClient(app) as client:
+        yield client
+
+
+@pytest.fixture
+def account(db_session):
+    # The account configure_app_for_test signs requests in as.
+    return Account.get_or_create("op@example.com")
+
+
+@pytest.fixture
+def resume_spy(monkeypatch):
+    calls = []
+
+    async def _spy(self, **fields):
+        calls.append({"id": self.id, **fields})
+
+    monkeypatch.setattr(Run, "resume", _spy)
+    return calls
+
+
+def _park(db_session, item_id):
+    run = seed_build_run(
+        db_session,
+        work_item_id=item_id,
+        state="pending_input",
+        input_gate="review",
+        input_request=dict(_IN_APP_ASK),
+    )
+    run.input_requested_at = datetime.now(UTC)
+    db_session.flush()
+    return run
+
+
+def test_openapi_pins_the_five_agent_routes(client: TestClient):
+    from druks.api.app import app
+
+    schema = app.openapi()
+    found = {
+        (method, path): operation
+        for path, operations in schema["paths"].items()
+        for method, operation in operations.items()
+        if operation.get("tags") == ["mcp"]
+    }
+    assert {key: op["operationId"] for key, op in found.items()} == _MCP_ROUTES
+
+
+def test_agent_routes_sit_behind_the_gate(tmp_path, db_session):
+    app = configure_app_for_test(settings=make_settings(tmp_path), authenticated=False)
+    with TestClient(app) as anonymous:
+        assert anonymous.get("/mcp/gates/x").status_code == 401
+        assert anonymous.get("/mcp/usage").status_code == 401
+
+
+def test_agent_errors_share_one_shape(client: TestClient, db_session):
+    missing = client.get("/mcp/gates/no-such-run")
+    assert missing.status_code == 404
+    assert missing.json() == {
+        "code": "RUN_NOT_FOUND",
+        "message": "No run no-such-run.",
+        "retryable": False,
+    }
+
+    item = make_test_work_item(repo="o/r", title="t")
+    run = _park(db_session, item.id)
+    stale = client.post(
+        f"/mcp/gates/{run.id}/answer",
+        json={"parkedAt": "2020-01-01T00:00:00+00:00", "control": "approve"},
+    )
+    assert stale.status_code == 409
+    body = stale.json()
+    assert body["code"] == "GATE_ROUND_STALE"
+    assert body["retryable"] is True
+
+
+def test_get_gate_then_answer_roundtrip(client: TestClient, db_session, resume_spy):
+    item = make_test_work_item(repo="o/r", title="t")
+    run = _park(db_session, item.id)
+
+    view = client.get(f"/mcp/gates/{run.id}")
+    assert view.status_code == 200
+    data = view.json()
+    assert data == services.get_gate(run.id).model_dump(mode="json", by_alias=True)
+
+    answered = client.post(
+        f"/mcp/gates/{run.id}/answer",
+        json={"parkedAt": data["parkedAt"], "control": "approve", "note": "ship it"},
+    )
+    assert answered.status_code == 200
+    assert answered.json()["result"] == "answered"
+    assert resume_spy == [{"id": run.id, "action": "approve", "answers": {}, "note": "ship it"}]
+
+
+def test_answer_gate_reads_already_answered_off_the_receipt(
+    client: TestClient, db_session, resume_spy
+):
+    item = make_test_work_item(repo="o/r", title="t")
+    parked_at = datetime.now(UTC)
+    run = seed_build_run(db_session, work_item_id=item.id, state="running")
+    run.input_requested_at = parked_at
+    run.answer_parked_at = parked_at
+    db_session.flush()
+
+    response = client.post(
+        f"/mcp/gates/{run.id}/answer",
+        json={"parkedAt": parked_at.isoformat(), "control": "approve"},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["result"] == "already_answered"
+    assert resume_spy == []
+
+
+def test_answer_gate_requires_an_aware_parked_at(client: TestClient, db_session):
+    item = make_test_work_item(repo="o/r", title="t")
+    run = _park(db_session, item.id)
+
+    naive = client.post(
+        f"/mcp/gates/{run.id}/answer",
+        json={"parkedAt": "2026-07-19T10:00:00", "control": "approve"},
+    )
+
+    assert naive.status_code == 422  # Pydantic's, not the agent taxonomy
+
+
+def test_cancel_run_route(client: TestClient, db_session):
+    item = make_test_work_item(repo="o/r", title="t")
+    run = seed_build_run(db_session, work_item_id=item.id, state="running")
+    # Parked, so the cancel must clear the gate — and never write the receipt.
+    run.input_gate = "review"
+    run.input_request = {"presentation": "in_app", "questions": []}
+    run.input_requested_at = run.utc_now()
+    db_session.flush()
+
+    unbounded = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": "r" * 501})
+    assert unbounded.status_code == 422
+    blank = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": ""})
+    assert blank.status_code == 422
+
+    cancelled = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": "wrong branch"})
+    assert cancelled.status_code == 200
+    assert cancelled.json() == {"runId": run.id, "result": "cancelled"}
+
+    db_session.expire_all()
+    run = db_session.get(type(run), run.id)
+    assert not run.answer_parked_at
+    assert not run.input_gate
+    assert run.failure == "wrong branch"
+
+    again = client.post(f"/mcp/runs/{run.id}/cancel", json={"reason": "wrong branch"})
+    assert again.status_code == 200
+    assert again.json()["result"] == "already_cancelled"
+
+
+def test_transcript_route_matches_the_read_machinery(client: TestClient, db_session, db_engine):
+    from conftest import seed_agent_run
+
+    call = seed_agent_run()
+    call_dir = call.call_dir
+    call_dir.mkdir(parents=True, exist_ok=True)
+    (call_dir / "stdout.jsonl").write_bytes(b"hello " + "é".encode() + b" transcript")
+
+    response = client.get(
+        f"/api/build/transcripts/{call.id}", params={"stream": "stdout", "limit": 7}
+    )
+    assert response.status_code == 200
+    chunk = read_transcript_chunk(db_engine, call.id, "stdout", offset=0, limit=7)
+    assert response.json() == chunk.model_dump(mode="json", by_alias=True)
+    # The 7-byte cut lands mid-é; the served chunk ends on a character boundary.
+    assert response.json()["text"] == "hello "
+
+
+def test_resume_route_contract_is_preserved(client: TestClient, db_session, resume_spy):
+    unknown = client.post("/api/runs/no-such-run/resume", json={"control": "approve"})
+    assert unknown.status_code == 404
+
+    item = make_test_work_item(repo="o/r", title="t")
+    idle = seed_build_run(db_session, work_item_id=item.id, state="running")
+    not_waiting = client.post(f"/api/runs/{idle.id}/resume", json={"control": "approve"})
+    assert not_waiting.status_code == 409
+
+    parked_item = make_test_work_item(repo="o/r2", title="t")
+    run = _park(db_session, parked_item.id)
+    bad_control = client.post(f"/api/runs/{run.id}/resume", json={"control": "merge"})
+    assert bad_control.status_code == 422
+    assert resume_spy == []
+
+    ok = client.post(
+        f"/api/runs/{run.id}/resume",
+        json={"control": "approve", "answers": {}, "note": "go"},
+    )
+    assert ok.status_code == 204
+    assert resume_spy == [{"id": run.id, "action": "approve", "answers": {}, "note": "go"}]
+
+    # Once the answer has landed (receipt written, gate cleared), the
+    # dashboard's double-submit stays the conflict it has always been.
+    run.answer_parked_at = run.input_requested_at
+    run.input_gate = None
+    run.input_request = None
+    db_session.flush()
+    late = client.post(f"/api/runs/{run.id}/resume", json={"control": "approve"})
+    assert late.status_code == 409
+    assert len(resume_spy) == 1
+
+
+def test_usage_agent_route_matches_the_service(client: TestClient, db_session, account):
+    from druks.durable.models import AgentCall
+
+    run = seed_run(db_session, "run-usage-route")
+    db_session.add(
+        AgentCall(
+            run_id=run.id,
+            account_id=account.id,
+            sandbox_host_id="host",
+            model="gpt-5.5",
+            status="succeeded",
+            finished_at=datetime.now(UTC),
+            cost_usd=1.25,
+            cost_metadata={"total_tokens": 500},
+        )
+    )
+    db_session.flush()
+
+    response = client.get("/mcp/usage")
+    assert response.status_code == 200
+    body = response.json()
+    assert body == services.get_usage(account).model_dump(mode="json", by_alias=True)
+    assert len(response.content) <= 4 * 1024
+
+    today = client.get("/api/usage/today").json()
+    assert sum(h["spendUsd"] for h in today["harnesses"]) == pytest.approx(body["spendTodayUsd"])
+    assert sum(h["runs"] for h in today["harnesses"]) == body["runsToday"]
+    assert sum(h["tokens"] for h in today["harnesses"]) == body["tokensToday"]
+    assert today["day"] == body["day"]
diff --git a/backend/tests/test_agent_services.py b/backend/tests/test_agent_services.py
new file mode 100644
index 0000000..e8581bf
--- /dev/null
+++ b/backend/tests/test_agent_services.py
@@ -0,0 +1,362 @@
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from conftest import make_test_work_item, seed_build_run, seed_call
+from druks.accounts.models import Account
+from druks.durable.models import Artifact, Run
+from druks.durable.reads import read_slice
+from druks.mcp.gateway import exceptions, services
+from druks.usage.models import UsageScrape
+
+pytestmark = pytest.mark.usefixtures("_data_dir")
+
+
+@pytest.fixture
+def _data_dir(tmp_path, monkeypatch):
+    # AgentCall.call_dir derives from load_settings().artifacts_dir.
+    monkeypatch.setenv("DRUKS_DATA_DIR", str(tmp_path))
+
+
+@pytest.fixture
+def account(db_session):
+    return Account.get_or_create("op@example.com")
+
+
+@pytest.fixture
+def resume_spy(monkeypatch):
+    calls = []
+
+    async def _spy(self, **fields):
+        calls.append({"id": self.id, **fields})
+
+    monkeypatch.setattr(Run, "resume", _spy)
+    return calls
+
+
+def _in_app_ask(questions=()):
+    return {
+        "presentation": "in_app",
+        "controls": ["approve", "request_changes", "cancel"],
+        "questions": list(questions),
+    }
+
+
+def _park(db_session, item_id, *, ask=None):
+    run = seed_build_run(
+        db_session,
+        work_item_id=item_id,
+        state="pending_input",
+        input_gate="review",
+        input_request=ask if ask is not None else _in_app_ask(),
+    )
+    run.input_requested_at = datetime.now(UTC)
+    db_session.flush()
+    return run
+
+
+# ---- read_slice -----------------------------------------------------------
+
+
+def test_read_slice_paginates_a_window(tmp_path: Path):
+    path = tmp_path / "log.txt"
+    path.write_bytes(b"hello world")
+
+    head = read_slice(path, offset=0, limit=5)
+    assert head.text == "hello"
+    assert head.next_offset == 5
+    assert head.eof is False
+    assert head.has_earlier is False
+
+    rest = read_slice(path, offset=head.next_offset, limit=100)
+    assert rest.text == " world"
+    assert rest.eof is True
+
+
+def test_read_slice_reads_the_tail(tmp_path: Path):
+    path = tmp_path / "log.txt"
+    path.write_bytes(b"hello world")
+
+    tail = read_slice(path, offset=-5, limit=5)
+    assert tail.offset == 6
+    assert tail.text == "world"
+    assert tail.has_earlier is True
+    assert tail.eof is True
+
+
+def test_read_slice_missing_file_is_an_empty_eof(tmp_path: Path):
+    piece = read_slice(tmp_path / "absent.txt", offset=-1024, limit=1024)
+    assert piece.text == ""
+    assert piece.eof is True
+    assert piece.has_earlier is False
+
+
+# ---- gates ----------------------------------------------------------------
+
+
+def test_get_gate_returns_the_ask_and_parked_at(db_session):
+    item = make_test_work_item(repo="o/r", title="t")
+    question = {"id": "q1", "prompt": "Which db?", "options": [{"id": "pg", "label": "Postgres"}]}
+    run = _park(db_session, item.id, ask=_in_app_ask([question]))
+
+    view = services.get_gate(run.id)
+
+    assert view.run_id == run.id
+    assert view.gate == "review"
+    assert view.parked_at == run.input_requested_at
+    assert view.ask["controls"] == ["approve", "request_changes", "cancel"]
+    assert view.ask["questions"][0]["prompt"] == "Which db?"
+
+
+def test_get_gate_serves_the_artifact(db_session):
+    item = make_test_work_item(repo="o/r", title="t")
+    run = _park(db_session, item.id)
+    call = seed_call(db_session, run, "generate_plan")
+    Artifact.record(
+        call_dir=call.call_dir, call_id=call.id, kind="markdown", title="Plan", content="x" * 10240
+    )
+
+    view = services.get_gate(run.id)
+
+    assert view.artifact is not None
+    assert view.artifact.call_id == call.id
+    assert view.artifact.title == "Plan"
+    assert len(view.artifact.content.encode()) <= 4096
+
+
+def test_get_gate_refuses_when_not_parked_or_external(db_session):
+    with pytest.raises(exceptions.RunNotFound):
+        services.get_gate("no-such-run")
+
+    item = make_test_work_item(repo="o/r", title="t")
+    running = seed_build_run(db_session, work_item_id=item.id, state="running")
+    with pytest.raises(exceptions.GateNotOpen):
+        services.get_gate(running.id)
+
+    external_item = make_test_work_item(repo="o/r2", title="t")
+    external = _park(
+        db_session,
+        external_item.id,
+        ask={"presentation": "external", "label": "Answer on the ticket"},
+    )
+    with pytest.raises(exceptions.GateNotAnswerable):
+        services.get_gate(external.id)
+
+
+async def test_answer_gate_resumes_through_run_resume(db_session, resume_spy):
+    item = make_test_work_item(repo="o/r", title="t")
+    run = _park(db_session, item.id)
+
+    result = await services.answer_gate(
+        run.id,
+        parked_at=run.input_requested_at,
+        control="approve",
+        answers={},
+        note="ship it",
+    )
+
+    assert result.result == "answered"
+    assert result.parked_at == run.input_requested_at
+    assert resume_spy == [{"id": run.id, "action": "approve", "answers": {}, "note": "ship it"}]
+
+
+async def test_answer_gate_uses_the_receipt_for_already_answered(db_session, resume_spy):
+    item = make_test_work_item(repo="o/r", title="t")
+    parked_at = datetime.now(UTC)
+    run = seed_build_run(db_session, work_item_id=item.id, state="running")
+    run.input_requested_at = parked_at
+    run.answer_parked_at = parked_at
+    db_session.flush()
+
+    result = await services.answer_gate(
+        run.id, parked_at=parked_at, control="approve", answers={}, note=""
+    )
+
+    assert result.result == "already_answered"
+    assert resume_spy == []
+
+
+async def test_answer_gate_error_taxonomy(db_session, resume_spy):
+    with pytest.raises(exceptions.RunNotFound):
+        await services.answer_gate(
+            "no-such-run", parked_at=datetime.now(UTC), control="approve", answers={}, note=""
+        )
+
+    item = make_test_work_item(repo="o/r", title="t")
+    finished = seed_build_run(db_session, work_item_id=item.id, state="finished")
+    with pytest.raises(exceptions.GateNotOpen):
+        await services.answer_gate(
+            finished.id, parked_at=datetime.now(UTC), control="approve", answers={}, note=""
+        )
+
+    parked_item = make_test_work_item(repo="o/r2", title="t")
+    run = _park(db_session, parked_item.id)
+    with pytest.raises(exceptions.GateRoundStale):
+        await services.answer_gate(
+            run.id,
+            parked_at=run.input_requested_at - timedelta(seconds=5),
+            control="approve",
+            answers={},
+            note="",
+        )
+    with pytest.raises(exceptions.InvalidGateAnswer):
+        await services.answer_gate(
+            run.id, parked_at=run.input_requested_at, control="merge", answers={}, note=""
+        )
+
+    external_item = make_test_work_item(repo="o/r3", title="t")
+    external = _park(
+        db_session,
+        external_item.id,
+        ask={"presentation": "external", "label": "Answer on the ticket"},
+    )
+    with pytest.raises(exceptions.GateNotAnswerable):
+        await services.answer_gate(
+            external.id,
+            parked_at=external.input_requested_at,
+            control="approve",
+            answers={},
+            note="",
+        )
+    assert resume_spy == []
+
+
+# ---- agent calls ----------------------------------------------------------
+
+
+def test_get_agent_call_serves_bounded_tails(db_session):
+    from conftest import finish_agent_run, seed_agent_run
+
+    call = seed_agent_run()
+    call_dir = call.call_dir
+    call_dir.mkdir(parents=True, exist_ok=True)
+    (call_dir / "stdout.jsonl").write_bytes(b"s" * 20480)
+    (call_dir / "stderr.log").write_bytes(b"e" * 10240)
+    finish_agent_run(call, last_error="boom " * 100)
+    Artifact.record(
+        call_dir=call_dir, call_id=call.id, kind="markdown", title="Out", content="a" * 10240
+    )
+
+    detail = services.get_agent_call(call.id)
+
+    assert detail.run_id == call.run_id
+    assert detail.call.id == call.id
+    assert detail.call.last_error == "boom " * 100
+    assert detail.transcript == "s" * 8192
+    assert detail.stderr == "e" * 4096
+    assert detail.artifact is not None
+    assert detail.artifact.content == "a" * 4096
+
+    with pytest.raises(exceptions.AgentCallNotFound):
+        services.get_agent_call("no-such-call")
+
+
+def test_get_agent_call_without_files_reads_empty(db_session):
+    from conftest import seed_agent_run
+
+    call = seed_agent_run()
+
+    detail = services.get_agent_call(call.id)
+
+    assert detail.transcript == ""
+    assert detail.stderr == ""
+    assert detail.artifact is None
+
+
+# ---- cancel ---------------------------------------------------------------
+
+
+async def test_cancel_run_paths(db_session):
+    item = make_test_work_item(repo="o/r", title="t")
+    run = seed_build_run(db_session, work_item_id=item.id, state="running")
+
+    result = await services.cancel_run(run.id, reason="stuck")
+    assert result.result == "cancelled"
+    db_session.expire_all()
+    assert Run.get(run.id).state == "cancelled"
+    assert Run.get(run.id).failure == "stuck"
+
+    again = await services.cancel_run(run.id, reason="stuck")
+    assert again.result == "already_cancelled"
+
+    finished_item = make_test_work_item(repo="o/r2", title="t")
+    finished = seed_build_run(db_session, work_item_id=finished_item.id, state="finished")
+    with pytest.raises(exceptions.RunNotActive):
+        await services.cancel_run(finished.id, reason="late")
+
+    with pytest.raises(exceptions.RunNotFound):
+        await services.cancel_run("no-such-run", reason="x")
+
+
+# ---- usage ----------------------------------------------------------------
+
+
+def test_get_usage_is_a_bounded_pure_read(db_session, account):
+    from conftest import seed_run
+    from druks.durable.models import AgentCall
+
+    now = datetime.now(UTC)
+    run = seed_run(db_session, "run-usage")
+    for index in range(30):
+        db_session.add(
+            AgentCall(
+                run_id=run.id,
+                account_id=account.id,
+                sandbox_host_id="host",
+                model="gpt-5.5",
+                status="succeeded",
+                finished_at=now,
+                cost_usd=0.5,
+                cost_metadata={"input_tokens": 100 + index},
+            )
+        )
+    for tick in range(40):
+        UsageScrape(
+            harness="claude",
+            account_id=account.id,
+            scraped_at=now - timedelta(minutes=5 * tick),
+            plan_tier="max",
+            five_hour_percent_left=90 - tick,
+            five_hour_resets_at=now + timedelta(hours=2),
+            week_percent_left=80 - tick,
+            week_resets_at=now + timedelta(days=3),
+        ).save()
+
+    usage = services.get_usage(account)
+
+    assert usage.runs_today == 30
+    assert usage.spend_today_usd == pytest.approx(15.0)
+    assert usage.tokens_today == sum(100 + i for i in range(30))
+    claude = next(h for h in usage.harnesses if h.name == "claude")
+    assert claude.plan_tier == "max"
+    assert claude.five_hour_percent_left == 90
+    assert len(claude.five_hour_history) <= 8
+    assert len(claude.week_history) <= 8
+    # The newest sample anchors each trend.
+    assert claude.week_history[-1].pct == 80
+    assert len(usage.model_dump_json(by_alias=True).encode()) <= 4 * 1024
+
+
+def test_get_usage_only_counts_the_callers_spend(db_session, account):
+    from conftest import seed_run
+    from druks.durable.models import AgentCall
+
+    other = Account.get_or_create("other@example.com")
+    run = seed_run(db_session, "run-usage-other")
+    db_session.add(
+        AgentCall(
+            run_id=run.id,
+            account_id=other.id,
+            sandbox_host_id="host",
+            model="gpt-5.5",
+            status="succeeded",
+            finished_at=datetime.now(UTC),
+            cost_usd=9.0,
+        )
+    )
+    db_session.flush()
+
+    usage = services.get_usage(account)
+
+    assert usage.runs_today == 0
+    assert usage.spend_today_usd == 0.0
diff --git a/backend/tests/test_contracts.py b/backend/tests/test_contracts.py
index 458b2a2..53a80de 100644
--- a/backend/tests/test_contracts.py
+++ b/backend/tests/test_contracts.py
@@ -4,6 +4,7 @@
 # real schema, so this guards it directly.
 import pytest
 from druks.build import contracts as O
+from pydantic import ValidationError
 
 MODELS = [
     O.PlanOutput,
@@ -111,3 +112,21 @@ def test_get_answered_maps_picks_to_labels_and_keeps_free_text_verbatim():
         {"question": "Which cache?", "answer": "Redis"},
         {"question": "Which queue?", "answer": "kafka — we already run it"},
     ]
+
+
+def test_ask_contracts_cap_identity_and_cardinality():
+    # The gate view is bounded by construction: identity and list sizes are
+    # hard caps at the agent boundary, never clipped downstream.
+    option = O.QuestionOptionOutput(id="a", label="Redis")
+    with pytest.raises(ValidationError):
+        O.QuestionOptionOutput(id="a" * 65, label="Redis")
+    with pytest.raises(ValidationError):
+        O.QuestionOutput(id="q", prompt="p" * 2049, options=[])
+    with pytest.raises(ValidationError):
+        O.QuestionOutput(id="q", prompt="p", options=[option] * 17)
+    with pytest.raises(ValidationError):
+        O.PlanOutput(
+            plan_markdown="m",
+            acceptance_criteria=[],
+            questions=[O.QuestionOutput(id=f"q{i}", prompt="p", options=[]) for i in range(9)],
+        )
diff --git a/backend/tests/test_gate_receipt.py b/backend/tests/test_gate_receipt.py
new file mode 100644
index 0000000..c9f26d2
--- /dev/null
+++ b/backend/tests/test_gate_receipt.py
@@ -0,0 +1,86 @@
+import pytest
+from conftest import seed_run
+from dbos import DBOS
+from dbos._error import DBOSWorkflowCancelledError
+from druks.durable.exceptions import GateTimeout
+from druks.durable.models import Run
+from druks.workflows import _park
+
+_ASK = {"presentation": "in_app", "controls": ["approve"], "questions": []}
+
+
+class _ParkedWorkflow:
+    # The slice of Workflow that _park touches. subject=None keeps the emit to
+    # its facts write (no feed event, no notification) — the receipt path under
+    # test is exactly that write.
+    def __init__(self, workflow_id: str) -> None:
+        self.workflow_id = workflow_id
+        self.subject = None
+
+    async def _reap_run(self) -> None:
+        return
+
+
+@pytest.fixture
+def _direct_steps(monkeypatch):
+    # Run each durable step body inline — the test exercises _park's own logic,
+    # not DBOS checkpointing.
+    async def _call_through(options, func, *args, **kwargs):
+        return await func(*args, **kwargs)
+
+    monkeypatch.setattr(DBOS, "run_step_async", _call_through)
+
+
+def _reload(db_session, run_id: str) -> Run:
+    db_session.expire_all()
+    return db_session.get(Run, run_id)
+
+
+async def test_answer_stamps_the_receipt_beside_the_gate_clear(
+    db_session, _direct_steps, monkeypatch
+):
+    run = seed_run(db_session, "run-receipt-answer")
+
+    async def _answer(topic, timeout_seconds):
+        return {"action": "approve"}
+
+    monkeypatch.setattr(DBOS, "recv_async", _answer)
+    payload = await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0)
+
+    assert payload == {"action": "approve"}
+    run = _reload(db_session, run.id)
+    # The receipt is the round the answer cleared: the same stamp the park
+    # wrote, which _GATE_CLEARED preserves on the row.
+    assert run.input_requested_at
+    assert run.answer_parked_at == run.input_requested_at
+    assert not run.input_gate
+    assert not run.input_request
+
+
+async def test_timeout_never_writes_the_receipt(db_session, _direct_steps, monkeypatch):
+    run = seed_run(db_session, "run-receipt-timeout")
+
+    async def _lapse(topic, timeout_seconds):
+        return None
+
+    monkeypatch.setattr(DBOS, "recv_async", _lapse)
+    with pytest.raises(GateTimeout):
+        await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0)
+
+    run = _reload(db_session, run.id)
+    assert not run.answer_parked_at
+    assert run.input_requested_at
+
+
+async def test_cancel_never_writes_the_receipt(db_session, _direct_steps, monkeypatch):
+    run = seed_run(db_session, "run-receipt-cancel")
+
+    async def _cancelled(topic, timeout_seconds):
+        raise DBOSWorkflowCancelledError(run.id)
+
+    monkeypatch.setattr(DBOS, "recv_async", _cancelled)
+    with pytest.raises(DBOSWorkflowCancelledError):
+        await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0)
+
+    run = _reload(db_session, run.id)
+    assert not run.answer_parked_at
diff --git a/backend/tests/test_resume.py b/backend/tests/test_resume.py
index 2da9384..4ac7acb 100644
--- a/backend/tests/test_resume.py
+++ b/backend/tests/test_resume.py
@@ -5,14 +5,25 @@
 from druks.durable import Run
 from fastapi import HTTPException
 
+# The shape review() parks with: an in-app ask plus the parked_at stamp the
+# answer service echoes.
 _ASK = {
+    "presentation": "in_app",
     "controls": ["approve", "request_changes", "cancel"],
     "questions": [{"id": "q1", "prompt": "which?", "options": [{"id": "a", "label": "A"}]}],
 }
 
 
 def _park(db_session) -> None:
-    db_session.add(Run(id="r1", kind="build", input_gate="review_plan", input_request=_ASK))
+    db_session.add(
+        Run(
+            id="r1",
+            kind="build",
+            input_gate="review_plan",
+            input_request=_ASK,
+            input_requested_at=Run.utc_now(),
+        )
+    )
     db_session.flush()
     seed_dbos_status(db_session, "r1", "pending_input")