From 4e431150148288dd2a77957ef46224be1f27243e Mon Sep 17 00:00:00 2001 From: kina0630 Date: Sun, 31 May 2026 22:03:59 +0800 Subject: [PATCH] feat: add advanced worker control plane foundations --- ...ternal-worker-control-plane-task-packet.md | 45 ++ ...ternal-worker-control-plane-foundations.md | 73 +++ .../src/coding_agent/cli/remote_commands.py | 95 ++++ .../src/coding_agent/remote/client.py | 78 ++++ .../src/coding_agent/remote/worker.py | 45 ++ .../src/coding_agent/server/http_server.py | 182 ++++++++ .../src/coding_agent/server/schemas.py | 34 ++ .../coding_agent/server/session_manager.py | 39 ++ coding-agent/tests/cli/test_remote_client.py | 89 ++++ .../tests/dogfood/test_local_dogfood_run.py | 6 + .../dogfood/test_workspace_provider_demo.py | 6 + .../integration/test_durable_runtime_smoke.py | 6 + .../tests/ui/test_developer_console.py | 426 ++++++++++-------- coding-agent/tests/ui/test_http_server.py | 220 +++++++++ .../tests/ui/test_session_manager_runtime.py | 13 +- 15 files changed, 1157 insertions(+), 200 deletions(-) create mode 100644 coding-agent/.opencode/prompts/tasks/advanced-external-worker-control-plane-task-packet.md create mode 100644 coding-agent/docs/adr/0053-advanced-external-worker-control-plane-foundations.md diff --git a/coding-agent/.opencode/prompts/tasks/advanced-external-worker-control-plane-task-packet.md b/coding-agent/.opencode/prompts/tasks/advanced-external-worker-control-plane-task-packet.md new file mode 100644 index 000000000..53fc6ccc6 --- /dev/null +++ b/coding-agent/.opencode/prompts/tasks/advanced-external-worker-control-plane-task-packet.md @@ -0,0 +1,45 @@ +# Advanced External Worker Control Plane Foundations Task Packet + +## Goal + +Implement the first advanced external-worker control-plane slice for worker +process metadata, durable approval interaction management, and workspace-sync +extension metadata. + +## Scope + +- Persist optional worker instance/process/capability/workspace-sync metadata + from external-worker claim and heartbeat requests. +- Expose that metadata through worker status APIs and CLI output. +- Add durable runtime interaction list/inspect/resolve HTTP APIs. +- Add remote client and CLI commands for interaction list/inspect/resolve. +- Reuse the existing session approval path for interaction resolution. + +## Out of Scope + +- Restoring an already-running local process after reconnect. +- Cross-machine workspace file synchronization. +- A dedicated worker registry table. +- Worker pool management UI. + +## ADRs + +- `docs/adr/0051-external-worker-execution-control-plane.md` +- `docs/adr/0052-external-worker-usable-control-plane.md` +- `docs/adr/0053-advanced-external-worker-control-plane-foundations.md` + +## Target Tests + +```bash +uv run pytest tests/ui/test_http_server.py -k "external_worker or interactions" -v +uv run pytest tests/cli/test_remote_client.py -k "remote_interaction or remote_worker or external_worker" -v +uv run pytest tests/coding_agent/test_pg_runtime_store.py -v +uv run ruff check docs/adr/0053-advanced-external-worker-control-plane-foundations.md src/coding_agent/server/http_server.py src/coding_agent/server/session_manager.py src/coding_agent/server/schemas.py src/coding_agent/remote/client.py src/coding_agent/remote/worker.py src/coding_agent/cli/remote_commands.py tests/ui/test_http_server.py tests/cli/test_remote_client.py tests/coding_agent/test_pg_runtime_store.py +``` + +## Stop Conditions + +- Stop if process restoration requires supervising or attaching to an existing + local process. +- Stop if workspace sync requires a new file transport protocol. +- Stop if approval resolution cannot reuse the existing session approval flow. diff --git a/coding-agent/docs/adr/0053-advanced-external-worker-control-plane-foundations.md b/coding-agent/docs/adr/0053-advanced-external-worker-control-plane-foundations.md new file mode 100644 index 000000000..133385854 --- /dev/null +++ b/coding-agent/docs/adr/0053-advanced-external-worker-control-plane-foundations.md @@ -0,0 +1,73 @@ +# ADR-0053: Advanced external worker control-plane foundations + +**Status**: Accepted +**Date**: 2026-05-31 + +## Context + +ADR-0051 and ADR-0052 made `external_worker` sessions usable: the server can +record run requests, local workers can claim and execute them, and users can +inspect sessions, runs, events, and derived worker health. The next step is to +leave concrete extension points for reconnect, durable approval interaction +management, and future workspace sync without prematurely implementing process +checkpointing or bidirectional workspace replication. + +The control plane already persists runtime runs and approval interactions. A +first advanced slice should reuse those records so API and CLI users can see and +resolve pending approvals consistently, while workers report enough metadata for +operators to distinguish a logical `worker_id` from a particular local process. + +## Decision + +Extend external-worker claim and heartbeat requests with optional +`worker_instance_id`, `process_id`, `capabilities`, and `workspace_sync` +metadata. The server stores these fields in run metadata and exposes them in +worker status responses. This is metadata-only reconnect support: it allows +operators to see which process owns a lease and what the worker advertises, but +it does not restore an already-running local process after a disconnect. + +Expose durable runtime interaction APIs for active approvals: +`GET /runs/{run_id}/interactions`, `GET /interactions`, `GET +/interactions/{interaction_id}`, and `POST /interactions/{interaction_id}/resolve`. +Resolution delegates to the existing session approval path so the runtime +message bus, in-memory approval coordinator, and persisted interaction record +remain one state machine. If the matching approval is no longer pending, the API +returns a conflict instead of directly mutating the database. + +Add remote client and CLI commands for listing, inspecting, and resolving +interactions. Workspace sync remains a declared metadata envelope (`mode`, +workspace reference, and capability data) until a later ADR defines actual sync +protocol semantics. + +## Alternatives Rejected + +- Add a separate approval queue table - rejected because `agent_interactions` + already persists approval request/response state and splitting it would create + reconciliation problems. +- Resolve approvals by updating `agent_interactions` directly - rejected because + it would not wake the waiting worker approval request or publish a runtime + approval decision. +- Implement process restoration now - rejected because reconnecting to an + already-running local process needs a local process registry, durable worker + supervisor semantics, and recovery rules beyond this slice. +- Implement cross-machine workspace sync now - rejected because live sync needs a + separate conflict model and transport protocol. + +## Acceptance Criteria + +- [ ] `test_external_worker_worker_metadata_surfaces_in_status` +- [ ] `test_runtime_run_interactions_endpoint_lists_interactions` +- [ ] `test_runtime_interaction_resolve_uses_session_approval_flow` +- [ ] `test_remote_interactions_list_and_resolve` +- [ ] `uv run pytest tests/ui/test_http_server.py -k "external_worker or interactions" -v` +- [ ] `uv run pytest tests/cli/test_remote_client.py -k "remote_interaction or remote_worker or external_worker" -v` + +## References + +- `docs/adr/0051-external-worker-execution-control-plane.md` +- `docs/adr/0052-external-worker-usable-control-plane.md` +- `src/coding_agent/server/http_server.py` +- `src/coding_agent/server/session_manager.py` +- `src/coding_agent/runtime_store.py` +- `src/coding_agent/remote/client.py` +- `src/coding_agent/cli/remote_commands.py` diff --git a/coding-agent/src/coding_agent/cli/remote_commands.py b/coding-agent/src/coding_agent/cli/remote_commands.py index 03dc52f28..094a11531 100644 --- a/coding-agent/src/coding_agent/cli/remote_commands.py +++ b/coding-agent/src/coding_agent/cli/remote_commands.py @@ -787,6 +787,101 @@ def remote_events(name: str, run_id: str) -> None: ) +@remote.command("interactions") +@click.argument("name") +@click.option("--session", "session_id", help="Filter by remote session ID.") +@click.option("--run", "run_id", help="Filter by remote run ID.") +@click.option("--status", "status", help="Filter by interaction status.") +def remote_interactions( + name: str, + session_id: str | None, + run_id: str | None, + status: str | None, +) -> None: + """List durable remote runtime interactions.""" + from coding_agent.remote.client import get_remote, list_remote_interactions + + endpoint = get_remote(name) + interactions = list_remote_interactions( + endpoint, + session_id=session_id, + run_id=run_id, + status=status, + ) + if not interactions: + click.echo("No remote interactions found.") + return + for interaction in interactions: + metadata = interaction.get("metadata") + request_id = ( + metadata.get("request_id") + if isinstance(metadata, dict) + and isinstance(metadata.get("request_id"), str) + else "" + ) + click.echo( + "\t".join( + [ + str(interaction.get("interaction_id", "")), + str(interaction.get("status", "")), + str(interaction.get("interaction_kind", "")), + request_id, + str(interaction.get("run_id", "")), + str(interaction.get("created_at", "")), + ] + ) + ) + + +@remote.command("interaction") +@click.argument("name") +@click.argument("interaction_id") +def remote_interaction(name: str, interaction_id: str) -> None: + """Show one durable remote runtime interaction.""" + from coding_agent.remote.client import get_remote, get_remote_interaction + + endpoint = get_remote(name) + _print_mapping(get_remote_interaction(endpoint, interaction_id)) + + +@remote.command("resolve-interaction") +@click.argument("name") +@click.argument("interaction_id") +@click.option("--approve", "approve", is_flag=True, help="Approve the interaction.") +@click.option("--reject", "reject", is_flag=True, help="Reject the interaction.") +@click.option("--feedback", "feedback", help="Optional approval feedback.") +@click.option( + "--scope", + "scope", + type=click.Choice(["once", "session", "always"]), + default="once", + show_default=True, +) +def remote_resolve_interaction( + name: str, + interaction_id: str, + approve: bool, + reject: bool, + feedback: str | None, + scope: str, +) -> None: + """Resolve a pending approval interaction.""" + from coding_agent.remote.client import get_remote, resolve_remote_interaction + + if approve == reject: + raise click.ClickException("Pass exactly one of --approve or --reject.") + endpoint = get_remote(name) + _print_mapping( + resolve_remote_interaction( + endpoint, + interaction_id, + approved=approve, + feedback=feedback, + scope=scope, + ) + ) + + @remote.command("workers") @click.argument("name") def remote_workers(name: str) -> None: diff --git a/coding-agent/src/coding_agent/remote/client.py b/coding-agent/src/coding_agent/remote/client.py index 1cc91e1bd..93a7dcaf7 100644 --- a/coding-agent/src/coding_agent/remote/client.py +++ b/coding-agent/src/coding_agent/remote/client.py @@ -9,6 +9,7 @@ from collections.abc import Mapping from pathlib import Path from typing import cast +from urllib.parse import quote, urlencode import click import httpx @@ -220,6 +221,83 @@ def list_remote_run_events( return [dict(_expect_mapping(item, "Remote run event entry")) for item in events] +def list_remote_run_interactions( + endpoint: RemoteEndpoint, + run_id: str, + *, + status: str | None = None, +) -> list[dict[str, object]]: + path = f"/runs/{run_id}/interactions" + if status is not None: + path = f"{path}?status={quote(status)}" + data = _get_remote_json(endpoint, path, "list remote run interactions") + interactions = data.get("interactions") + if not isinstance(interactions, list): + raise click.ClickException("Remote interactions response missing interactions") + return [ + dict(_expect_mapping(item, "Remote interaction entry")) for item in interactions + ] + + +def list_remote_interactions( + endpoint: RemoteEndpoint, + *, + session_id: str | None = None, + run_id: str | None = None, + status: str | None = None, +) -> list[dict[str, object]]: + params: list[tuple[str, str]] = [] + if session_id is not None: + params.append(("session_id", session_id)) + if run_id is not None: + params.append(("run_id", run_id)) + if status is not None: + params.append(("status", status)) + path = "/interactions" + if params: + path = f"{path}?{urlencode(params)}" + data = _get_remote_json(endpoint, path, "list remote interactions") + interactions = data.get("interactions") + if not isinstance(interactions, list): + raise click.ClickException("Remote interactions response missing interactions") + return [ + dict(_expect_mapping(item, "Remote interaction entry")) for item in interactions + ] + + +def get_remote_interaction( + endpoint: RemoteEndpoint, + interaction_id: str, +) -> dict[str, object]: + return _get_remote_json( + endpoint, + f"/interactions/{quote(interaction_id, safe='')}", + "get remote interaction", + ) + + +def resolve_remote_interaction( + endpoint: RemoteEndpoint, + interaction_id: str, + *, + approved: bool, + feedback: str | None = None, + scope: str = "once", +) -> dict[str, object]: + payload: dict[str, object] = { + "approved": approved, + "scope": scope, + } + if feedback is not None: + payload["feedback"] = feedback + return _post_remote_json( + endpoint, + f"/interactions/{quote(interaction_id, safe='')}/resolve", + json=payload, + action="resolve remote interaction", + ) + + def list_remote_workers(endpoint: RemoteEndpoint) -> list[dict[str, object]]: data = _get_remote_json(endpoint, "/workers", "list remote workers") workers = data.get("workers") diff --git a/coding-agent/src/coding_agent/remote/worker.py b/coding-agent/src/coding_agent/remote/worker.py index 49ff8e4a2..087ffd3ab 100644 --- a/coding-agent/src/coding_agent/remote/worker.py +++ b/coding-agent/src/coding_agent/remote/worker.py @@ -2,6 +2,7 @@ import asyncio import json +import os import uuid from dataclasses import asdict, is_dataclass from datetime import datetime @@ -128,6 +129,7 @@ async def run_local_worker_once( headers=headers, timeout=_CONTROL_TIMEOUT, ) as client: + worker_instance_id = _new_worker_instance_id(worker_id) session_id = await _create_external_worker_session( client=client, repo_path=repo_path, @@ -147,6 +149,8 @@ async def run_local_worker_once( client=client, session_id=session_id, worker_id=worker_id, + worker_instance_id=worker_instance_id, + repo_path=repo_path, ) if claim is None: raise ExternalWorkerError("requested run was not available to claim") @@ -157,6 +161,7 @@ async def run_local_worker_once( claim=claim, repo_path=repo_path, worker_id=worker_id, + worker_instance_id=worker_instance_id, ) @@ -174,11 +179,14 @@ async def run_worker_loop( headers=headers, timeout=_CONTROL_TIMEOUT, ) as client: + worker_instance_id = _new_worker_instance_id(worker_id) while True: claim = await _claim_run( client=client, session_id=None, worker_id=worker_id, + worker_instance_id=worker_instance_id, + repo_path=repo_path, allow_empty=True, ) if claim is None: @@ -191,6 +199,7 @@ async def run_worker_loop( claim=claim, repo_path=repo_path, worker_id=worker_id, + worker_instance_id=worker_instance_id, ) if once: return status @@ -264,11 +273,17 @@ async def _claim_run( client: httpx.AsyncClient, session_id: str | None, worker_id: str, + worker_instance_id: str, + repo_path: Path, allow_empty: bool = False, ) -> dict[str, Any] | None: payload: dict[str, Any] = { "worker_id": worker_id, "executor_kind": "local_cli", + "worker_instance_id": worker_instance_id, + "process_id": os.getpid(), + "capabilities": _worker_capabilities(), + "workspace_sync": _workspace_sync_metadata(repo_path), } if session_id is not None: payload["session_id"] = session_id @@ -291,6 +306,7 @@ async def _execute_claimed_run( claim: dict[str, Any], repo_path: Path, worker_id: str, + worker_instance_id: str, ) -> int: run_id = _required_claim_str(claim, "run_id") session_id = _required_claim_str(claim, "session_id") @@ -327,8 +343,10 @@ async def _execute_claimed_run( client=client, run_id=run_id, worker_id=worker_id, + worker_instance_id=worker_instance_id, claim_token=claim_token, run_task=run_task, + repo_path=repo_path, ) ) try: @@ -400,14 +418,20 @@ async def _heartbeat_until_complete( client: httpx.AsyncClient, run_id: str, worker_id: str, + worker_instance_id: str, claim_token: str, run_task: asyncio.Task[Any], + repo_path: Path, ) -> None: while not run_task.done(): response = await client.post( f"/worker/runs/{run_id}/heartbeat", json={ "worker_id": worker_id, + "worker_instance_id": worker_instance_id, + "process_id": os.getpid(), + "capabilities": _worker_capabilities(), + "workspace_sync": _workspace_sync_metadata(repo_path), "claim_token": claim_token, "lease_seconds": 30, }, @@ -420,6 +444,27 @@ async def _heartbeat_until_complete( await asyncio.sleep(5) +def _new_worker_instance_id(worker_id: str) -> str: + return f"{worker_id}:{uuid.uuid4().hex}" + + +def _worker_capabilities() -> dict[str, Any]: + return { + "executor": "local_cli", + "process_reconnect": "metadata_only", + "workspace_sync": "metadata_only", + "approval_interactions": True, + } + + +def _workspace_sync_metadata(repo_path: Path) -> dict[str, Any]: + return { + "mode": "none", + "workspace_ref_kind": "local_path", + "display_path": str(repo_path), + } + + def _required_claim_str(claim: dict[str, Any], key: str) -> str: value = claim.get(key) if not isinstance(value, str) or not value: diff --git a/coding-agent/src/coding_agent/server/http_server.py b/coding-agent/src/coding_agent/server/http_server.py index 7df4caab6..6e4811e3f 100644 --- a/coding-agent/src/coding_agent/server/http_server.py +++ b/coding-agent/src/coding_agent/server/http_server.py @@ -76,7 +76,9 @@ record_http_request_metric, ) from coding_agent.runtime_store import ( + AgentInteractionRecord, AgentRunRecord, + JSONObject, RunMessageSnapshotRecord, RuntimeEventRecord, ) @@ -189,8 +191,11 @@ PublishSessionRequest, PublishSessionResponse, ReadinessResponse, + ResolveInteractionRequest, RuntimeEventResponse, RuntimeEventsResponse, + RuntimeInteractionListResponse, + RuntimeInteractionResponse, RuntimeMessageSnapshotResponse, RuntimeRunListResponse, RuntimeRunResponse, @@ -2478,6 +2483,10 @@ async def claim_worker_run( executor_kind=body.executor_kind, session_id=body.session_id, lease_seconds=body.lease_seconds, + worker_instance_id=body.worker_instance_id, + process_id=body.process_id, + capabilities=cast(JSONObject | None, body.capabilities), + workspace_sync=cast(JSONObject | None, body.workspace_sync), ) except RuntimeError as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc @@ -2512,6 +2521,10 @@ async def heartbeat_worker_run( worker_id=body.worker_id, claim_token=body.claim_token, lease_seconds=body.lease_seconds, + worker_instance_id=body.worker_instance_id, + process_id=body.process_id, + capabilities=cast(JSONObject | None, body.capabilities), + workspace_sync=cast(JSONObject | None, body.workspace_sync), ) except KeyError as exc: raise HTTPException(status_code=404, detail=_key_error_detail(exc)) from exc @@ -3239,6 +3252,22 @@ def _runtime_event_response(record: RuntimeEventRecord) -> RuntimeEventResponse: ) +def _runtime_interaction_response( + record: AgentInteractionRecord, +) -> RuntimeInteractionResponse: + return RuntimeInteractionResponse( + interaction_id=record.interaction_id, + run_id=record.run_id, + interaction_kind=record.interaction_kind, + status=record.status, + request_payload=record.request_payload, + response_payload=record.response_payload, + metadata=record.metadata, + created_at=record.created_at, + resolved_at=record.resolved_at, + ) + + def _safe_dict(value: object) -> dict[str, object]: return value if isinstance(value, dict) else {} @@ -3323,6 +3352,9 @@ def _worker_status_from_runs( else: status = "running" workspace_ref = metadata.get("workspace_ref") + capabilities = metadata.get("capabilities") + workspace_sync = metadata.get("workspace_sync") + process_id = metadata.get("process_id") return WorkerStatusResponse( worker_id=worker_id, status=status, @@ -3336,7 +3368,15 @@ def _worker_status_from_runs( if isinstance(metadata.get("worker_pool"), str) else None ), + worker_instance_id=( + metadata.get("worker_instance_id") + if isinstance(metadata.get("worker_instance_id"), str) + else None + ), + process_id=process_id if isinstance(process_id, int) else None, + capabilities=capabilities if isinstance(capabilities, dict) else None, workspace_ref=workspace_ref if isinstance(workspace_ref, dict) else None, + workspace_sync=workspace_sync if isinstance(workspace_sync, dict) else None, current_run_id=current.run_id if current is not None else None, current_session_id=current.session_id if current is not None else None, last_run_id=latest.run_id, @@ -5107,6 +5147,37 @@ async def get_runtime_run( return _runtime_run_response(record) +@app.get( + "/runs/{run_id}/interactions", + response_model=RuntimeInteractionListResponse, +) +@limiter.limit(RateLimits.GET_SESSION) +async def get_runtime_run_interactions( + request: Request, + run_id: str, + status: str | None = Query(None, min_length=1, max_length=100), + auth_context: AuthContext | None = Depends(auth_context_from_headers), +) -> RuntimeInteractionListResponse: + del request + _ = await _get_visible_runtime_run(run_id, auth_context) + try: + interactions = await session_manager.list_runtime_interactions(run_id) + except RuntimeError as exc: + raise HTTPException( + status_code=404, + detail="Runtime interactions not found", + ) from exc + if status is not None: + interactions = [ + interaction for interaction in interactions if interaction.status == status + ] + return RuntimeInteractionListResponse( + interactions=[ + _runtime_interaction_response(interaction) for interaction in interactions + ] + ) + + @app.get( "/runs/{run_id}/message-snapshot", response_model=RuntimeMessageSnapshotResponse, @@ -5156,6 +5227,117 @@ async def get_runtime_events( ) +@app.get("/interactions", response_model=RuntimeInteractionListResponse) +@limiter.limit(RateLimits.GET_SESSION) +async def list_runtime_interactions( + request: Request, + session_id: str | None = Query(None, min_length=1, max_length=100), + run_id: str | None = Query(None, min_length=1, max_length=100), + status: str | None = Query(None, min_length=1, max_length=100), + auth_context: AuthContext | None = Depends(auth_context_from_headers), +) -> RuntimeInteractionListResponse: + del request + if run_id is not None: + runs = [await _get_visible_runtime_run(run_id, auth_context)] + elif session_id is not None: + _ = await _get_visible_session(session_id, auth_context) + runs = await session_manager.list_runtime_runs(session_id) + else: + runs = await _visible_runtime_runs(auth_context) + interactions: list[AgentInteractionRecord] = [] + for run in runs: + if session_id is not None and run.session_id != session_id: + continue + try: + interactions.extend( + await session_manager.list_runtime_interactions(run.run_id) + ) + except RuntimeError: + continue + if status is not None: + interactions = [ + interaction for interaction in interactions if interaction.status == status + ] + interactions.sort(key=lambda interaction: interaction.created_at, reverse=True) + return RuntimeInteractionListResponse( + interactions=[ + _runtime_interaction_response(interaction) for interaction in interactions + ] + ) + + +@app.get( + "/interactions/{interaction_id}", + response_model=RuntimeInteractionResponse, +) +@limiter.limit(RateLimits.GET_SESSION) +async def get_runtime_interaction( + request: Request, + interaction_id: str, + auth_context: AuthContext | None = Depends(auth_context_from_headers), +) -> RuntimeInteractionResponse: + del request + try: + interaction = await session_manager.load_runtime_interaction(interaction_id) + except (KeyError, RuntimeError) as exc: + raise HTTPException( + status_code=404, + detail="Runtime interaction not found", + ) from exc + _ = await _get_visible_runtime_run(interaction.run_id, auth_context) + return _runtime_interaction_response(interaction) + + +@app.post( + "/interactions/{interaction_id}/resolve", + response_model=RuntimeInteractionResponse, +) +@limiter.limit(RateLimits.APPROVE) +async def resolve_runtime_interaction( + request: Request, + interaction_id: str, + body: ResolveInteractionRequest, + auth_context: AuthContext | None = Depends(auth_context_from_headers), +) -> RuntimeInteractionResponse: + del request + try: + interaction = await session_manager.load_runtime_interaction(interaction_id) + except (KeyError, RuntimeError) as exc: + raise HTTPException( + status_code=404, + detail="Runtime interaction not found", + ) from exc + _ = await _get_visible_runtime_run(interaction.run_id, auth_context) + if interaction.interaction_kind != "approval": + raise HTTPException(status_code=400, detail="Interaction is not an approval") + if interaction.status != "pending": + raise HTTPException(status_code=409, detail="Interaction is not pending") + session_id = interaction.metadata.get("session_id") + request_id = interaction.metadata.get("request_id") + if not isinstance(session_id, str) or not isinstance(request_id, str): + raise HTTPException( + status_code=400, + detail="Approval interaction metadata is incomplete", + ) + try: + approval = await session_manager.submit_approval_response( + session_id=session_id, + request_id=request_id, + approved=body.approved, + feedback=body.feedback, + scope=body.scope, + ) + except KeyError as exc: + raise HTTPException(status_code=404, detail="Session not found") from exc + if approval is None: + raise HTTPException( + status_code=409, + detail="Approval request is no longer pending", + ) + resolved = await session_manager.load_runtime_interaction(interaction_id) + return _runtime_interaction_response(resolved) + + @app.get("/workers", response_model=WorkerListResponse) @limiter.limit(RateLimits.GET_SESSION) async def list_workers( diff --git a/coding-agent/src/coding_agent/server/schemas.py b/coding-agent/src/coding_agent/server/schemas.py index f248940bf..d648c5a9b 100644 --- a/coding-agent/src/coding_agent/server/schemas.py +++ b/coding-agent/src/coding_agent/server/schemas.py @@ -234,6 +234,10 @@ class WorkerClaimRequest(BaseModel): executor_kind: str = Field("local_cli", min_length=1, max_length=100) session_id: str | None = Field(None, min_length=1, max_length=100) lease_seconds: int = Field(30, ge=5, le=300) + worker_instance_id: str | None = Field(None, min_length=1, max_length=200) + process_id: int | None = Field(None, ge=1) + capabilities: dict[str, Any] | None = None + workspace_sync: dict[str, Any] | None = None class WorkerClaimResponse(BaseModel): @@ -253,6 +257,10 @@ class WorkerHeartbeatRequest(BaseModel): worker_id: str = Field(..., min_length=1, max_length=200) claim_token: str = Field(..., min_length=1, max_length=500) lease_seconds: int = Field(30, ge=5, le=300) + worker_instance_id: str | None = Field(None, min_length=1, max_length=200) + process_id: int | None = Field(None, ge=1) + capabilities: dict[str, Any] | None = None + workspace_sync: dict[str, Any] | None = None class WorkerHeartbeatResponse(BaseModel): @@ -305,7 +313,11 @@ class WorkerStatusResponse(BaseModel): status: Literal["idle", "running", "stale", "offline"] executor_kind: str | None = None worker_pool: str | None = None + worker_instance_id: str | None = None + process_id: int | None = None + capabilities: dict[str, Any] | None = None workspace_ref: dict[str, Any] | None = None + workspace_sync: dict[str, Any] | None = None current_run_id: str | None = None current_session_id: str | None = None last_run_id: str | None = None @@ -318,6 +330,28 @@ class WorkerListResponse(BaseModel): workers: list[WorkerStatusResponse] +class RuntimeInteractionResponse(BaseModel): + interaction_id: str + run_id: str + interaction_kind: str + status: str + request_payload: dict[str, Any] + response_payload: dict[str, Any] + metadata: dict[str, Any] + created_at: datetime + resolved_at: datetime | None = None + + +class RuntimeInteractionListResponse(BaseModel): + interactions: list[RuntimeInteractionResponse] + + +class ResolveInteractionRequest(BaseModel): + approved: bool + feedback: str | None = Field(None, max_length=1000) + scope: Literal["once", "session", "always"] = "once" + + class WorkspaceDiffFileSchema(BaseModel): path: str status: Literal["added", "modified", "deleted", "renamed", "binary", "unknown"] diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 7f2aa15c2..6ac50cfb8 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/coding-agent/src/coding_agent/server/session_manager.py @@ -342,6 +342,11 @@ async def create_agent_interaction( record: AgentInteractionRecord, ) -> AgentInteractionRecord: ... + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: ... + async def list_agent_interactions( self, run_id: str, @@ -903,6 +908,16 @@ async def list_runtime_interactions( store = self._require_runtime_store() return await store.list_agent_interactions(run_id) + async def load_runtime_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord: + store = self._require_runtime_store() + record = await store.load_agent_interaction(interaction_id) + if record is None: + raise KeyError(f"runtime interaction not found: {interaction_id}") + return record + async def load_tape_debug_info(self, tape_id: str) -> TapeInfo | None: if not isinstance(self._tape_store, TapeDebugStore): return None @@ -1017,6 +1032,10 @@ async def claim_external_worker_run( executor_kind: str, session_id: str | None = None, lease_seconds: int = 30, + worker_instance_id: str | None = None, + process_id: int | None = None, + capabilities: JSONObject | None = None, + workspace_sync: JSONObject | None = None, ) -> ExternalWorkerClaim | None: store = self._require_runtime_store() claim_token = secrets.token_urlsafe(32) @@ -1028,6 +1047,14 @@ async def claim_external_worker_run( "claimed_at": now.isoformat(), "lease_expires_at": lease_expires_at.isoformat(), } + if worker_instance_id is not None: + claim_metadata["worker_instance_id"] = worker_instance_id + if process_id is not None: + claim_metadata["process_id"] = process_id + if capabilities is not None: + claim_metadata["capabilities"] = capabilities + if workspace_sync is not None: + claim_metadata["workspace_sync"] = workspace_sync run = await store.claim_external_worker_run( session_id=session_id, executor_kind=executor_kind, @@ -1051,6 +1078,10 @@ async def heartbeat_external_worker_run( worker_id: str, claim_token: str, lease_seconds: int = 30, + worker_instance_id: str | None = None, + process_id: int | None = None, + capabilities: JSONObject | None = None, + workspace_sync: JSONObject | None = None, ) -> AgentRunRecord: run = await self._load_and_authorize_external_worker_run( run_id=run_id, @@ -1062,6 +1093,14 @@ async def heartbeat_external_worker_run( datetime.now(UTC) + timedelta(seconds=lease_seconds) ).isoformat() metadata["last_heartbeat_at"] = datetime.now(UTC).isoformat() + if worker_instance_id is not None: + metadata["worker_instance_id"] = worker_instance_id + if process_id is not None: + metadata["process_id"] = process_id + if capabilities is not None: + metadata["capabilities"] = capabilities + if workspace_sync is not None: + metadata["workspace_sync"] = workspace_sync status = "running" if run.status == "claimed" else run.status return await self._require_runtime_store().update_agent_run( run_id, diff --git a/coding-agent/tests/cli/test_remote_client.py b/coding-agent/tests/cli/test_remote_client.py index b588feb57..7e30fec95 100644 --- a/coding-agent/tests/cli/test_remote_client.py +++ b/coding-agent/tests/cli/test_remote_client.py @@ -555,6 +555,95 @@ def fake_list_remote_workers(endpoint): ) +def test_remote_interactions_list_and_resolve(tmp_path: Path, monkeypatch) -> None: + config_path = tmp_path / "remotes.json" + monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) + + def fake_list_remote_interactions(endpoint, *, session_id, run_id, status): + assert endpoint.name == "dev" + assert session_id == "sess-1" + assert run_id is None + assert status == "pending" + return [ + { + "interaction_id": "run-1:approval-1", + "status": "pending", + "interaction_kind": "approval", + "run_id": "run-1", + "created_at": "2026-05-31T12:00:00+00:00", + "metadata": {"request_id": "approval-1"}, + } + ] + + def fake_resolve_remote_interaction( + endpoint, + interaction_id, + *, + approved, + feedback, + scope, + ): + assert endpoint.name == "dev" + assert interaction_id == "run-1:approval-1" + assert approved is True + assert feedback == "ok" + assert scope == "once" + return { + "interaction_id": interaction_id, + "status": "approved", + } + + monkeypatch.setattr( + "coding_agent.remote.client.list_remote_interactions", + fake_list_remote_interactions, + ) + monkeypatch.setattr( + "coding_agent.remote.client.resolve_remote_interaction", + fake_resolve_remote_interaction, + ) + runner = CliRunner() + runner.invoke( + main, + ["remote", "add", "dev", "http://agent.example", "--token", "secret-token"], + catch_exceptions=False, + ) + + list_result = runner.invoke( + main, + [ + "remote", + "interactions", + "dev", + "--session", + "sess-1", + "--status", + "pending", + ], + catch_exceptions=False, + ) + resolve_result = runner.invoke( + main, + [ + "remote", + "resolve-interaction", + "dev", + "run-1:approval-1", + "--approve", + "--feedback", + "ok", + ], + catch_exceptions=False, + ) + + assert list_result.exit_code == 0 + assert ( + "run-1:approval-1\tpending\tapproval\tapproval-1\trun-1\t" + "2026-05-31T12:00:00+00:00" + ) in list_result.output + assert resolve_result.exit_code == 0 + assert "status: approved" in resolve_result.output + + def test_remote_runs_lists_session_runs(tmp_path: Path, monkeypatch) -> None: config_path = tmp_path / "remotes.json" monkeypatch.setenv("CODING_AGENT_REMOTES_FILE", str(config_path)) diff --git a/coding-agent/tests/dogfood/test_local_dogfood_run.py b/coding-agent/tests/dogfood/test_local_dogfood_run.py index 11e985c5d..59479cd2b 100644 --- a/coding-agent/tests/dogfood/test_local_dogfood_run.py +++ b/coding-agent/tests/dogfood/test_local_dogfood_run.py @@ -95,6 +95,12 @@ async def create_agent_interaction( self.interactions[record.interaction_id] = record return record + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: + return self.interactions.get(interaction_id) + async def list_agent_interactions( self, run_id: str, diff --git a/coding-agent/tests/dogfood/test_workspace_provider_demo.py b/coding-agent/tests/dogfood/test_workspace_provider_demo.py index 2a89e1916..3ec835a2a 100644 --- a/coding-agent/tests/dogfood/test_workspace_provider_demo.py +++ b/coding-agent/tests/dogfood/test_workspace_provider_demo.py @@ -112,6 +112,12 @@ async def create_agent_interaction( self.interactions[record.interaction_id] = record return record + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: + return self.interactions.get(interaction_id) + async def list_agent_interactions( self, run_id: str, diff --git a/coding-agent/tests/integration/test_durable_runtime_smoke.py b/coding-agent/tests/integration/test_durable_runtime_smoke.py index 80a227f0e..928e1831c 100644 --- a/coding-agent/tests/integration/test_durable_runtime_smoke.py +++ b/coding-agent/tests/integration/test_durable_runtime_smoke.py @@ -144,6 +144,12 @@ async def create_agent_interaction( self.interactions[record.interaction_id] = record return record + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: + return self.interactions.get(interaction_id) + async def resolve_agent_interaction( self, interaction_id: str, diff --git a/coding-agent/tests/ui/test_developer_console.py b/coding-agent/tests/ui/test_developer_console.py index 2aefffa10..89d71d5a4 100644 --- a/coding-agent/tests/ui/test_developer_console.py +++ b/coding-agent/tests/ui/test_developer_console.py @@ -119,38 +119,40 @@ async def search( entries = [] for known_tape_id in self.tape_ids: known_run_id = known_tape_id.replace("tape", "run") - entries.extend([ - TapeSearchResult( - tape_id=known_tape_id, - seq=0, - entry={ - "kind": "message", - "payload": { - "run_id": known_run_id, - "content": "SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT", + entries.extend( + [ + TapeSearchResult( + tape_id=known_tape_id, + seq=0, + entry={ + "kind": "message", + "payload": { + "run_id": known_run_id, + "content": "SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT", + }, }, - }, - ), - TapeSearchResult( - tape_id=known_tape_id, - seq=1, - entry={ - "kind": "tool_call", - "payload": { - "run_id": known_run_id, - "tool_call_id": "tool-alpha", + ), + TapeSearchResult( + tape_id=known_tape_id, + seq=1, + entry={ + "kind": "tool_call", + "payload": { + "run_id": known_run_id, + "tool_call_id": "tool-alpha", + }, }, - }, - ), - TapeSearchResult( - tape_id=known_tape_id, - seq=2, - entry={ - "kind": "anchor", - "meta": {"anchor_type": "handoff", "secret": "hidden"}, - }, - ), - ]) + ), + TapeSearchResult( + tape_id=known_tape_id, + seq=2, + entry={ + "kind": "anchor", + "meta": {"anchor_type": "handoff", "secret": "hidden"}, + }, + ), + ] + ) filtered = [] for result in entries: entry = result.entry @@ -274,6 +276,12 @@ async def create_agent_interaction( self.interactions.setdefault(record.interaction_id, record) return record + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: + return self.interactions.get(interaction_id) + async def resolve_agent_interaction( self, interaction_id: str, @@ -837,38 +845,40 @@ def _topic_runtime_run(run_id: str = "run-alpha") -> AgentRunRecord: def _bee_runtime_run(run_id: str = "run-bee") -> AgentRunRecord: base = _runtime_run(run_id, "session-alpha", status="completed") metadata = dict(base.metadata) - metadata.update({ - "bee_runtime": "task_launch", - "launch_id": "launch-alpha", - "launch_source": "schedule", - "launch_status": "launched", - "launch_kind": "durable_run", - "task_id": "bee-task-alpha", - "node_id": "node-validate", - "topic_id": "topic-auth", - "session_id": "session-alpha", - "template_id": "launch-blueprint-alpha", - "schedule_id": "schedule-alpha", - "task_kind": "maintenance", - "task_profile": "local", - "node_kind": "validation", - "node_profile": "default", - "approval_policy": "existing_runtime_policy", - "action_policy": "existing_action_safety", - "workspace_binding": "existing_workspace_provider", - "workspace_policy": "default", - "context_profile": "repo", - "context_reference": "profile_only", - "validation_profile": "pytest", - "validation_reference": "profile_only", - "executor_run_id": "executor-run-alpha", - "executor_kind": "local", - "executor_status": "succeeded", - "executor_capability": "available", - "executor_summary": "Local executor succeeded", - "prompt": "SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT", - "command_output": "SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT", - }) + metadata.update( + { + "bee_runtime": "task_launch", + "launch_id": "launch-alpha", + "launch_source": "schedule", + "launch_status": "launched", + "launch_kind": "durable_run", + "task_id": "bee-task-alpha", + "node_id": "node-validate", + "topic_id": "topic-auth", + "session_id": "session-alpha", + "template_id": "launch-blueprint-alpha", + "schedule_id": "schedule-alpha", + "task_kind": "maintenance", + "task_profile": "local", + "node_kind": "validation", + "node_profile": "default", + "approval_policy": "existing_runtime_policy", + "action_policy": "existing_action_safety", + "workspace_binding": "existing_workspace_provider", + "workspace_policy": "default", + "context_profile": "repo", + "context_reference": "profile_only", + "validation_profile": "pytest", + "validation_reference": "profile_only", + "executor_run_id": "executor-run-alpha", + "executor_kind": "local", + "executor_status": "succeeded", + "executor_capability": "available", + "executor_summary": "Local executor succeeded", + "prompt": "SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT", + "command_output": "SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT", + } + ) return AgentRunRecord( run_id=base.run_id, session_id=base.session_id, @@ -1146,14 +1156,16 @@ async def _configure_run_detail_fixture( error: str | None = None, ) -> _ConsoleRuntimeStore: _register_console_session("session-detail") - store = _ConsoleRuntimeStore([ - _runtime_run( - "run-detail", - "session-detail", - status=status, - error=error, - ) - ]) + store = _ConsoleRuntimeStore( + [ + _runtime_run( + "run-detail", + "session-detail", + status=status, + error=error, + ) + ] + ) await store.save_message_snapshot( RunMessageSnapshotRecord( snapshot_id="run-detail:latest", @@ -1380,9 +1392,9 @@ async def test_console_observability_renders_configured_links_and_correlation( ) -> None: _register_console_session("session-alpha") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run("run-alpha", "session-alpha", status="completed") - ]) + _ConsoleRuntimeStore( + [_runtime_run("run-alpha", "session-alpha", status="completed")] + ) ) monkeypatch.setattr( http_server, @@ -1716,20 +1728,22 @@ async def test_console_bee_renders_durable_executor_run_summary( monkeypatch.setattr( http_server, "_console_executor_run_store", - lambda: _ConsoleExecutorRunStore([ - ExecutorRunRecord( - executor_run_id="executor-run-store", - executor_kind="kubernetes_job", - task_id="bee-task-store", - node_id="node-validate", - launch_id="launch-store", - topic_id="topic-store", - status="succeeded", - requested_at=datetime(2026, 5, 20, 1, 0, tzinfo=UTC), - sanitized_summary="Kubernetes Job succeeded", - metadata={"capability_status": "available"}, - ) - ]), + lambda: _ConsoleExecutorRunStore( + [ + ExecutorRunRecord( + executor_run_id="executor-run-store", + executor_kind="kubernetes_job", + task_id="bee-task-store", + node_id="node-validate", + launch_id="launch-store", + topic_id="topic-store", + status="succeeded", + requested_at=datetime(2026, 5, 20, 1, 0, tzinfo=UTC), + sanitized_summary="Kubernetes Job succeeded", + metadata={"capability_status": "available"}, + ) + ] + ), ) transport = ASGITransport(app=app) @@ -1758,43 +1772,45 @@ async def test_console_bee_launch_renders_durable_launch_store_summary( monkeypatch.setattr( http_server, "_console_bee_launch_store", - lambda: _ConsoleBeeLaunchStore([ - BeeLaunchRecord( - launch_id="launch-manual", - source="manual", - template_id="template-manual", - status="launched", - requested_at=datetime(2026, 5, 20, 1, 0, tzinfo=UTC), - task_id="bee-task-manual", - topic_id="topic-manual", - session_id="session-alpha", - launched_at=datetime(2026, 5, 20, 1, 1, tzinfo=UTC), - ), - BeeLaunchRecord( - launch_id="launch-scheduled", - source="schedule", - template_id="template-scheduled", - status="launched", - requested_at=datetime(2026, 5, 20, 2, 0, tzinfo=UTC), - task_id="bee-task-scheduled", - topic_id="topic-scheduled", - session_id="session-alpha", - schedule_id="schedule-alpha", - launched_at=datetime(2026, 5, 20, 2, 1, tzinfo=UTC), - ), - BeeLaunchRecord( - launch_id="launch-signal", - source="proactive_signal", - template_id="template-signal", - status="failed", - requested_at=datetime(2026, 5, 20, 3, 0, tzinfo=UTC), - topic_id="topic-signal", - session_id="session-alpha", - signal_id="signal-alpha", - error_type="policy_denied", - error_message="safe policy denied", - ), - ]), + lambda: _ConsoleBeeLaunchStore( + [ + BeeLaunchRecord( + launch_id="launch-manual", + source="manual", + template_id="template-manual", + status="launched", + requested_at=datetime(2026, 5, 20, 1, 0, tzinfo=UTC), + task_id="bee-task-manual", + topic_id="topic-manual", + session_id="session-alpha", + launched_at=datetime(2026, 5, 20, 1, 1, tzinfo=UTC), + ), + BeeLaunchRecord( + launch_id="launch-scheduled", + source="schedule", + template_id="template-scheduled", + status="launched", + requested_at=datetime(2026, 5, 20, 2, 0, tzinfo=UTC), + task_id="bee-task-scheduled", + topic_id="topic-scheduled", + session_id="session-alpha", + schedule_id="schedule-alpha", + launched_at=datetime(2026, 5, 20, 2, 1, tzinfo=UTC), + ), + BeeLaunchRecord( + launch_id="launch-signal", + source="proactive_signal", + template_id="template-signal", + status="failed", + requested_at=datetime(2026, 5, 20, 3, 0, tzinfo=UTC), + topic_id="topic-signal", + session_id="session-alpha", + signal_id="signal-alpha", + error_type="policy_denied", + error_message="safe policy denied", + ), + ] + ), ) transport = ASGITransport(app=app) @@ -2053,15 +2069,17 @@ async def test_console_workspaces_renders_provider_inventory_without_raw_content monkeypatch: pytest.MonkeyPatch, ) -> None: session_manager.configure_workspace_metadata_store( - _ConsoleWorkspaceStore([ - _workspace_record("workspace-alpha"), - _workspace_record( - "workspace-remote", - status="retained", - provider_instance_id="docker-remote", - cleanup_error="safe cleanup failure", - ), - ]) + _ConsoleWorkspaceStore( + [ + _workspace_record("workspace-alpha"), + _workspace_record( + "workspace-remote", + status="retained", + provider_instance_id="docker-remote", + cleanup_error="safe cleanup failure", + ), + ] + ) ) monkeypatch.setattr( http_server, @@ -2262,20 +2280,22 @@ async def test_console_tape_restricts_user_token_to_visible_tapes( owner_label=_owner_label("admin-token"), ) session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run( - "run-user", - "session-user", - status="completed", - tape_id="tape-user", - ), - _runtime_run( - "run-admin", - "session-admin", - status="completed", - tape_id="tape-admin", - ), - ]) + _ConsoleRuntimeStore( + [ + _runtime_run( + "run-user", + "session-user", + status="completed", + tape_id="tape-user", + ), + _runtime_run( + "run-admin", + "session-admin", + status="completed", + tape_id="tape-admin", + ), + ] + ) ) session_manager._tape_store = _ConsoleTapeStore(["tape-user", "tape-admin"]) @@ -2323,9 +2343,9 @@ async def test_console_tape_renders_missing_state() -> None: async def test_console_context_renders_context_pack_evidence_without_body() -> None: _register_console_session("session-alpha") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run("run-alpha", "session-alpha", status="completed") - ]) + _ConsoleRuntimeStore( + [_runtime_run("run-alpha", "session-alpha", status="completed")] + ) ) transport = ASGITransport(app=app) @@ -2351,21 +2371,23 @@ async def test_console_context_renders_empty_state_for_missing_pack() -> None: _register_console_session("session-alpha") run = _runtime_run("run-alpha", "session-alpha", status="completed") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - AgentRunRecord( - run_id=run.run_id, - session_id=run.session_id, - tape_id=run.tape_id, - parent_run_id=run.parent_run_id, - agent_id=run.agent_id, - status=run.status, - started_at=run.started_at, - ended_at=run.ended_at, - metadata={}, - result=run.result, - error=run.error, - ) - ]) + _ConsoleRuntimeStore( + [ + AgentRunRecord( + run_id=run.run_id, + session_id=run.session_id, + tape_id=run.tape_id, + parent_run_id=run.parent_run_id, + agent_id=run.agent_id, + status=run.status, + started_at=run.started_at, + ended_at=run.ended_at, + metadata={}, + result=run.result, + error=run.error, + ) + ] + ) ) transport = ASGITransport(app=app) @@ -2381,9 +2403,9 @@ async def test_console_context_renders_empty_state_for_missing_pack() -> None: async def test_console_memory_renders_memory_evidence_without_raw_content() -> None: _register_console_session("session-alpha") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run("run-alpha", "session-alpha", status="completed") - ]) + _ConsoleRuntimeStore( + [_runtime_run("run-alpha", "session-alpha", status="completed")] + ) ) transport = ASGITransport(app=app) @@ -2454,9 +2476,9 @@ async def test_console_memory_renders_empty_state_for_missing_run() -> None: async def test_console_actions_renders_action_validation_and_policy_summaries() -> None: _register_console_session("session-alpha") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run("run-alpha", "session-alpha", status="completed") - ]) + _ConsoleRuntimeStore( + [_runtime_run("run-alpha", "session-alpha", status="completed")] + ) ) transport = ASGITransport(app=app) @@ -2515,10 +2537,12 @@ async def test_console_memory_and_actions_restrict_user_token_to_visible_run( owner_label=_owner_label("admin-token"), ) session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run("run-user", "session-user", status="completed"), - _runtime_run("run-admin", "session-admin", status="completed"), - ]) + _ConsoleRuntimeStore( + [ + _runtime_run("run-user", "session-user", status="completed"), + _runtime_run("run-admin", "session-admin", status="completed"), + ] + ) ) transport = ASGITransport(app=app) @@ -2545,9 +2569,9 @@ async def test_console_memory_and_actions_restrict_user_token_to_visible_run( @pytest.mark.asyncio async def test_console_interactions_renders_pending_and_resolved_lists() -> None: _register_console_session("session-alpha") - store = _ConsoleRuntimeStore([ - _runtime_run("run-alpha", "session-alpha", status="running") - ]) + store = _ConsoleRuntimeStore( + [_runtime_run("run-alpha", "session-alpha", status="running")] + ) await store.create_agent_interaction( _interaction("interaction-pending", "run-alpha", status="pending") ) @@ -2585,9 +2609,9 @@ async def test_console_interactions_renders_pending_and_resolved_lists() -> None @pytest.mark.asyncio async def test_console_interactions_displays_terminal_duplicate_state_safely() -> None: _register_console_session("session-alpha") - store = _ConsoleRuntimeStore([ - _runtime_run("run-alpha", "session-alpha", status="completed") - ]) + store = _ConsoleRuntimeStore( + [_runtime_run("run-alpha", "session-alpha", status="completed")] + ) await store.create_agent_interaction( _interaction( "interaction-terminal", @@ -2716,16 +2740,18 @@ async def test_console_runs_list_renders_fixture_data_and_status_filter() -> Non _register_console_session("session-alpha") _register_console_session("session-beta") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run("run-complete", "session-alpha", status="completed"), - _runtime_run( - "run-failed", - "session-beta", - status="failed", - error="safe failure summary", - ), - _runtime_run("run-running", "session-alpha", status="running"), - ]) + _ConsoleRuntimeStore( + [ + _runtime_run("run-complete", "session-alpha", status="completed"), + _runtime_run( + "run-failed", + "session-beta", + status="failed", + error="safe failure summary", + ), + _runtime_run("run-running", "session-alpha", status="running"), + ] + ) ) transport = ASGITransport(app=app) @@ -2753,14 +2779,16 @@ async def test_console_runs_list_renders_fixture_data_and_status_filter() -> Non async def test_console_runs_list_redacts_sensitive_error_summary() -> None: _register_console_session("session-sensitive") session_manager.configure_runtime_store( - _ConsoleRuntimeStore([ - _runtime_run( - "run-sensitive", - "session-sensitive", - status="failed", - error="SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT command_output", - ) - ]) + _ConsoleRuntimeStore( + [ + _runtime_run( + "run-sensitive", + "session-sensitive", + status="failed", + error="SECRET_PROMPT_MESSAGE_CONTENT_RESULT_TEXT command_output", + ) + ] + ) ) transport = ASGITransport(app=app) diff --git a/coding-agent/tests/ui/test_http_server.py b/coding-agent/tests/ui/test_http_server.py index c6b5aff31..d0b4025b5 100644 --- a/coding-agent/tests/ui/test_http_server.py +++ b/coding-agent/tests/ui/test_http_server.py @@ -36,6 +36,7 @@ from coding_agent.core.config import settings from coding_agent.environment import CloudCommandResult, CloudEnvironment from coding_agent.runtime_store import ( + AgentInteractionRecord, AgentRunRecord, RunMessageSnapshotRecord, RuntimeEventRecord, @@ -2579,6 +2580,72 @@ async def test_external_worker_workers_endpoint_reports_running_and_stale_worker assert active_offline_response.status_code == 200 assert active_offline_response.json()["status"] == "offline" + async def test_external_worker_worker_metadata_surfaces_in_status( + self, client, tmp_path: Path + ): + store = FakeExternalWorkerRuntimeStore() + session_manager.configure_runtime_store(store) + create_resp = await client.post( + "/sessions", + json={ + "approval_policy": "yolo", + "execution_binding": { + "kind": "external_worker", + "executor_kind": "local_cli", + "worker_pool": "default", + "workspace_ref": { + "kind": "local_path", + "display_path": str(tmp_path), + }, + }, + }, + ) + session_id = create_resp.json()["session_id"] + run = await session_manager.request_external_worker_run(session_id, "hello") + + claim_resp = await client.post( + "/worker/runs/claim", + json={ + "worker_id": "worker-1", + "executor_kind": "local_cli", + "session_id": session_id, + "worker_instance_id": "worker-1:instance-1", + "process_id": 1234, + "capabilities": { + "process_reconnect": "metadata_only", + "workspace_sync": "metadata_only", + }, + "workspace_sync": { + "mode": "none", + "workspace_ref_kind": "local_path", + }, + }, + ) + assert claim_resp.status_code == 200 + claim = claim_resp.json() + + heartbeat_resp = await client.post( + f"/worker/runs/{run.run_id}/heartbeat", + json={ + "worker_id": "worker-1", + "claim_token": claim["claim_token"], + "worker_instance_id": "worker-1:instance-2", + "process_id": 5678, + "capabilities": {"process_reconnect": "metadata_only"}, + "workspace_sync": {"mode": "none"}, + }, + ) + assert heartbeat_resp.status_code == 200 + + status_resp = await client.get("/workers/worker-1") + + assert status_resp.status_code == 200 + worker = status_resp.json() + assert worker["worker_instance_id"] == "worker-1:instance-2" + assert worker["process_id"] == 5678 + assert worker["capabilities"] == {"process_reconnect": "metadata_only"} + assert worker["workspace_sync"] == {"mode": "none"} + async def test_external_worker_approval_uses_server_approval_flow( self, client, monkeypatch ): @@ -2643,6 +2710,99 @@ async def fake_wait_for_approval( assert observed[0].tool == "shell_execute" assert observed[0].args == {"command": "pwd"} + async def test_runtime_run_interactions_endpoint_lists_interactions(self, client): + store = FakeExternalWorkerRuntimeStore() + session_manager.configure_runtime_store(store) + create_resp = await client.post( + "/sessions", + json={ + "approval_policy": "interactive", + "execution_binding": { + "kind": "external_worker", + "executor_kind": "local_cli", + }, + }, + ) + session_id = create_resp.json()["session_id"] + run = await session_manager.request_external_worker_run(session_id, "hello") + approval_req = ApprovalRequest( + session_id=session_id, + agent_id="", + request_id="approval-1", + tool="shell_execute", + args={"command": "pwd"}, + ) + + approval_task = asyncio.create_task( + session_manager.wait_for_http_approval( + session_id, + approval_req, + timeout_seconds=30, + ) + ) + await _wait_for_fake_interaction(store, approval_task) + + response = await client.get(f"/runs/{run.run_id}/interactions") + + assert response.status_code == 200 + interactions = response.json()["interactions"] + assert len(interactions) == 1 + assert interactions[0]["interaction_id"] == f"{run.run_id}:approval:approval-1" + assert interactions[0]["status"] == "pending" + assert interactions[0]["metadata"]["session_id"] == session_id + + approval_task.cancel() + with pytest.raises(asyncio.CancelledError): + await approval_task + + async def test_runtime_interaction_resolve_uses_session_approval_flow(self, client): + store = FakeExternalWorkerRuntimeStore() + session_manager.configure_runtime_store(store) + create_resp = await client.post( + "/sessions", + json={ + "approval_policy": "interactive", + "execution_binding": { + "kind": "external_worker", + "executor_kind": "local_cli", + }, + }, + ) + session_id = create_resp.json()["session_id"] + run = await session_manager.request_external_worker_run(session_id, "hello") + approval_req = ApprovalRequest( + session_id=session_id, + agent_id="", + request_id="approval-1", + tool="shell_execute", + args={"command": "pwd"}, + ) + approval_task = asyncio.create_task( + session_manager.wait_for_http_approval( + session_id, + approval_req, + timeout_seconds=30, + ) + ) + await _wait_for_fake_interaction(store, approval_task) + + response = await client.post( + f"/interactions/{run.run_id}:approval:approval-1/resolve", + json={ + "approved": True, + "feedback": "ok", + "scope": "once", + }, + ) + approval = await approval_task + + assert response.status_code == 200 + resolved = response.json() + assert resolved["status"] == "approved" + assert resolved["response_payload"]["approved"] is True + assert approval.approved is True + assert approval.feedback == "ok" + async def test_external_worker_recovery_expires_stale_claim(self, client): store = FakeExternalWorkerRuntimeStore() session_manager.configure_runtime_store(store) @@ -7182,6 +7342,7 @@ class FakeExternalWorkerRuntimeStore: def __init__(self) -> None: self.runs: dict[str, AgentRunRecord] = {} self.events: list[RuntimeEventRecord] = [] + self.interactions: dict[str, AgentInteractionRecord] = {} async def create_agent_run(self, record: AgentRunRecord) -> AgentRunRecord: self.runs[record.run_id] = record @@ -7246,6 +7407,65 @@ async def append_runtime_event( self.events.append(stored) return stored + async def create_agent_interaction( + self, + record: AgentInteractionRecord, + ) -> AgentInteractionRecord: + stored = self.interactions.setdefault(record.interaction_id, record) + return stored + + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: + return self.interactions.get(interaction_id) + + async def list_agent_interactions( + self, + run_id: str, + ) -> list[AgentInteractionRecord]: + return [ + interaction + for interaction in self.interactions.values() + if interaction.run_id == run_id + ] + + async def resolve_agent_interaction( + self, + interaction_id: str, + *, + status: str, + response_payload: dict[str, object], + resolved_at: datetime, + ) -> AgentInteractionRecord: + interaction = self.interactions[interaction_id] + if interaction.resolved_at is not None: + return interaction + resolved = replace( + interaction, + status=status, + response_payload=response_payload, + resolved_at=resolved_at, + ) + self.interactions[interaction_id] = resolved + return resolved + + +async def _wait_for_fake_interaction( + store: FakeExternalWorkerRuntimeStore, + approval_task: asyncio.Task[ApprovalResponse], +) -> None: + async def wait_until_registered() -> None: + while not store.interactions: + if approval_task.done(): + _ = approval_task.result() + raise AssertionError( + "approval task finished before interaction was registered" + ) + await asyncio.sleep(0) + + await asyncio.wait_for(wait_until_registered(), timeout=1) + class TestRuntimeReplayEndpoints: async def test_runtime_replay_endpoints_return_run_snapshot_and_events( diff --git a/coding-agent/tests/ui/test_session_manager_runtime.py b/coding-agent/tests/ui/test_session_manager_runtime.py index dbdbc71fd..9cb4b2769 100644 --- a/coding-agent/tests/ui/test_session_manager_runtime.py +++ b/coding-agent/tests/ui/test_session_manager_runtime.py @@ -149,6 +149,15 @@ async def create_agent_interaction( self.interactions.append(record) return record + async def load_agent_interaction( + self, + interaction_id: str, + ) -> AgentInteractionRecord | None: + for interaction in self.interactions: + if interaction.interaction_id == interaction_id: + return interaction + return None + async def resolve_agent_interaction( self, interaction_id: str, @@ -1918,7 +1927,9 @@ def create_failover_agent(**kwargs): return fake_pipeline, rebuilt_ctx monkeypatch = pytest.MonkeyPatch() - monkeypatch.setattr("coding_agent.server.session_manager.PipelineAdapter", FakeAdapter) + monkeypatch.setattr( + "coding_agent.server.session_manager.PipelineAdapter", FakeAdapter + ) original_task: asyncio.Task[object] | None = None try: