From 171afa73111f446f97ee241b62b5c9d410fb9af3 Mon Sep 17 00:00:00 2001 From: Yichen Wu Date: Mon, 20 Jul 2026 23:02:08 -0700 Subject: [PATCH 1/2] Harden cancellation recovery and concurrency --- ARCHITECTURE.md | 12 +- README.md | 3 + .../versions/0010_operation_journal.py | 25 ++ apps/api/app/core/config.py | 5 +- apps/api/app/db/models/task.py | 4 + apps/api/app/provider_gateway/providers.py | 5 +- apps/api/app/provider_gateway/runtime.py | 7 + apps/api/app/repositories/task.py | 18 ++ apps/api/app/services/agent_react.py | 289 +++++++++++++++--- apps/api/app/services/changeset.py | 38 +-- apps/api/app/services/runner.py | 16 +- apps/api/app/services/task.py | 3 + apps/api/app/tools/calendar.py | 6 +- apps/api/app/tools/email.py | 6 + apps/api/app/tools/mcp/stdio.py | 4 + apps/api/app/tools/provider_gateway.py | 8 + apps/api/app/tools/shell.py | 5 + apps/api/pyproject.toml | 1 + apps/api/requirements.lock | 4 + apps/api/tests/test_agent_react.py | 151 +++++++++ apps/api/tests/test_calendar.py | 26 +- apps/api/tests/test_changeset.py | 25 ++ apps/api/tests/test_deployment_boundaries.py | 3 +- apps/api/tests/test_email.py | 14 +- apps/api/tests/test_mcp.py | 26 ++ apps/api/tests/test_provider_gateway.py | 104 ++++++- apps/api/tests/test_runner.py | 43 ++- apps/api/tests/test_sandbox.py | 113 +++++++ apps/api/tests/test_tools.py | 19 ++ apps/api/tests/test_worker_queue.py | 44 +++ apps/api/uv.lock | 2 + docs/STRATEGY.md | 17 ++ docs/guides/security.md | 15 +- docs/loop.md | 20 +- scripts/k8s-deployment-acceptance.sh | 29 +- 35 files changed, 1021 insertions(+), 89 deletions(-) create mode 100644 apps/api/alembic/versions/0010_operation_journal.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index fd0edec..95c35da 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -174,7 +174,10 @@ flowchart LR - **Background jobs:** Redis Streams consumer groups provide an atomic producer, cross-pod consumers, visibility leases, `XAUTOCLAIM` recovery, retries, and a dead-letter stream. The task row is claimed with a compare-and-update before the - loop starts, so duplicate deliveries do not execute a task twice. + loop starts, so duplicate deliveries do not execute a task twice. Every tool call + commits a write-ahead operation journal before execution; its Step clears the journal + in the same commit, and recovery fails closed on an unknown outcome rather than + duplicating a mutation. - **Why a separate worker tier?** Long or bursty work must not block request threads or share the API's scaling signal. The checked-in worker HPA uses CPU as a portable baseline; queue-depth scaling requires KEDA or a custom metrics adapter. @@ -297,7 +300,12 @@ Full playbook: [`docs/guides/scaling.md`](./docs/guides/scaling.md). drains and upgrades. - **Idempotency:** publishes persist an owner-scoped idempotency key with a database uniqueness constraint. Trigger fires use an atomic timestamp claim plus a derived - idempotency key, so multiple scheduler replicas are safe. + idempotency key, so multiple scheduler replicas are safe. External writes require a + stable operation id (also used as SMTP Message-ID/CalDAV UID), without claiming that + an upstream protocol provides exactly-once delivery. +- **Cancellation:** a DB watcher cancels the active run coroutine rather than waiting + for the next step. Cancellation reaches provider transports, descendant runs, shell + process groups, Docker cleanup, and Kubernetes Job deletion. --- diff --git a/README.md b/README.md index 09938fd..37a6466 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,9 @@ an instruction to an accepted result. It separates **work** from **acceptance**: runtime provenance, output hashes, and the head of a hash-chained step ledger. - **Proof is portable.** The Receipt can be replayed later through the API, CLI, or bundled GitHub Action. +- **Interruption is a first-class state.** In-flight cancellation tears down the active + execution boundary, while a durable operation journal refuses crash recovery that + could repeat a mutation with an unknown outcome. If the evidence fails, the task does not become `completed` merely because the model called `finish`. Limit, stuck, cancelled, and error outcomes still receive an diff --git a/apps/api/alembic/versions/0010_operation_journal.py b/apps/api/alembic/versions/0010_operation_journal.py new file mode 100644 index 0000000..34b2e2f --- /dev/null +++ b/apps/api/alembic/versions/0010_operation_journal.py @@ -0,0 +1,25 @@ +"""Add the crash-safe in-flight operation journal. + +Revision ID: 0010_operation_journal +Revises: 0009_contract_draft +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0010_operation_journal" +down_revision: str | None = "0009_contract_draft" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column("tasks", sa.Column("operation_journal", sa.JSON(), nullable=True)) + + +def downgrade() -> None: + op.drop_column("tasks", "operation_journal") diff --git a/apps/api/app/core/config.py b/apps/api/app/core/config.py index ad4dabc..0c24c97 100644 --- a/apps/api/app/core/config.py +++ b/apps/api/app/core/config.py @@ -133,7 +133,10 @@ def parse_cors_origins(cls, value: object) -> object: execution_mode: Literal["inline", "worker"] = "inline" # Cap concurrent inline runs: each pins a DB connection for its whole run, so # without a bound a burst of publishes exhausts the pool. Excess runs queue. - agent_max_concurrent_runs: int = 8 + agent_max_concurrent_runs: int = Field(default=8, ge=1, le=256) + # A separate DB session watches a running task so API cancellation interrupts + # an in-flight provider/tool call instead of waiting for the next agent step. + agent_cancellation_poll_seconds: float = Field(default=0.25, ge=0.05, le=5.0) # A task RUNNING with no update for longer than this is treated as stranded by a # crash and failed on reconcile. Must exceed the longest gap between step commits # (one LLM call + one command + retries), so a live run is never wrongly failed. diff --git a/apps/api/app/db/models/task.py b/apps/api/app/db/models/task.py index 208e2ed..4671d72 100644 --- a/apps/api/app/db/models/task.py +++ b/apps/api/app/db/models/task.py @@ -71,6 +71,10 @@ class TaskModel(UUIDPrimaryKeyMixin, TimestampMixin, Base): chat_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) # The action awaiting approval while paused: {"tool": ..., "args": {...}}. pending_action: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + # Durable write-ahead record for a tool that may have crossed a side-effect + # boundary but whose Step has not committed yet. Recovery fails closed instead + # of blindly replaying an operation with an unknown outcome. + operation_journal: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) idempotency_key: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=1) diff --git a/apps/api/app/provider_gateway/providers.py b/apps/api/app/provider_gateway/providers.py index 4597df7..900260a 100644 --- a/apps/api/app/provider_gateway/providers.py +++ b/apps/api/app/provider_gateway/providers.py @@ -55,10 +55,12 @@ def _send(self, args: dict[str, Any], relay: tuple[str, int]) -> str: to = str(args.get("to", "")).strip() if not to: return "send_email needs a 'to' address." + operation_id = str(uuid.UUID(str(args["operation_id"]))) msg = EmailMessage() msg["From"] = self.settings.email_from or self.settings.smtp_user or "" msg["To"] = to msg["Subject"] = str(args.get("subject", "")).strip() + msg["Message-ID"] = f"<{operation_id}@loop>" msg.set_content(str(args.get("body", ""))) with _SMTPThroughRelay( self.settings.smtp_host or "", @@ -161,6 +163,7 @@ def _create(self, args: dict[str, Any], proxy_url: str) -> str: summary = str(args.get("summary", "")).strip() if not summary: return "create_event needs a 'summary'." + operation_id = str(uuid.UUID(str(args["operation_id"]))) try: start = datetime.fromisoformat(str(args["start"])) end = ( @@ -178,7 +181,7 @@ def _create(self, args: dict[str, Any], proxy_url: str) -> str: event.add("summary", summary) event.add("dtstart", start) event.add("dtend", end) - event.add("uid", f"{uuid.uuid4().hex}@loop") + event.add("uid", f"{operation_id}@loop") if args.get("description"): event.add("description", str(args["description"])) obj.add_component(event) diff --git a/apps/api/app/provider_gateway/runtime.py b/apps/api/app/provider_gateway/runtime.py index f632371..9afc6e3 100644 --- a/apps/api/app/provider_gateway/runtime.py +++ b/apps/api/app/provider_gateway/runtime.py @@ -80,6 +80,13 @@ async def invoke( raise AuthorityTokenError(f"Authority token does not grant {tool!r}") if required is None and not grant.permits(Capability.NET_BROWSER): raise AuthorityTokenError("Authority token does not grant browser access") + if tool in {"send_email", "create_event"}: + try: + uuid.UUID(str(args["operation_id"])) + except (KeyError, ValueError) as exc: + raise AuthorityTokenError( + f"{tool} requires a Loop operation_id for duplicate protection" + ) from exc if browser is not None and tool in browser_tools: if not egress_token: diff --git a/apps/api/app/repositories/task.py b/apps/api/app/repositories/task.py index 96a9b30..7080fd7 100644 --- a/apps/api/app/repositories/task.py +++ b/apps/api/app/repositories/task.py @@ -85,6 +85,24 @@ async def list_children(self, parent_id: uuid.UUID) -> list[TaskModel]: ) return list((await self.session.scalars(stmt)).all()) + async def list_descendants(self, parent_id: uuid.UUID) -> list[TaskModel]: + descendants: list[TaskModel] = [] + frontier = [parent_id] + seen = {parent_id} + while frontier: + children = list( + ( + await self.session.scalars( + select(TaskModel).where(TaskModel.parent_id.in_(frontier)) + ) + ).all() + ) + fresh = [child for child in children if child.id not in seen] + descendants.extend(fresh) + seen.update(child.id for child in fresh) + frontier = [child.id for child in fresh] + return descendants + async def recent_for_chat( self, chat_id: str, diff --git a/apps/api/app/services/agent_react.py b/apps/api/app/services/agent_react.py index b1da953..f44448c 100644 --- a/apps/api/app/services/agent_react.py +++ b/apps/api/app/services/agent_react.py @@ -20,11 +20,12 @@ from __future__ import annotations import asyncio +import contextlib import json import re import shutil import uuid -from collections.abc import Callable +from collections.abc import Awaitable, Callable from datetime import date from pathlib import Path from typing import Any @@ -213,11 +214,13 @@ def __init__( steps: StepRepository, llm: LLMClient, verifier_llm: LLMClient | None = None, + cancellation_probe: Callable[[uuid.UUID], Awaitable[bool]] | None = None, ) -> None: self.tasks = tasks self.steps = steps self.llm = llm self.verifier_llm = verifier_llm or llm + self._cancellation_probe = cancellation_probe self.session = tasks.session self._history: list[HistoryEntry] = [] self._last_hash = "" # head of the step hash chain @@ -238,8 +241,134 @@ def __init__( self._sandbox_backend: str | None = None self._egress_allowed = False # resolved egress; verification checks mirror it self._authority_token_factory: Callable[[str], str] | None = None + self._active_resources: list[Any] = [] + self._active_authority_clients: list[tuple[str, Any]] = [] + self._active_run_id: str | None = None + self._authority_revoked = False async def run(self, task_id: uuid.UUID) -> None: + if self._cancellation_probe is None: + try: + await self._run_claimed(task_id) + finally: + with contextlib.suppress(Exception): + await self._revoke_terminal_authority(task_id) + await self._stop_active_resources() + return + execution = asyncio.create_task(self._run_claimed(task_id), name=f"loop-run-{task_id}") + cancellation = asyncio.create_task( + self._wait_for_cancellation(task_id), name=f"loop-cancel-{task_id}" + ) + try: + done, _ = await asyncio.wait( + {execution, cancellation}, return_when=asyncio.FIRST_COMPLETED + ) + if cancellation in done and cancellation.result(): + execution.cancel() + with contextlib.suppress(asyncio.CancelledError): + await execution + await self._finalize_cancellation(task_id) + return + await execution + except asyncio.CancelledError: + execution.cancel() + with contextlib.suppress(asyncio.CancelledError): + await execution + with contextlib.suppress(Exception): + await self._finalize_cancellation(task_id) + raise + finally: + cancellation.cancel() + with contextlib.suppress(asyncio.CancelledError): + await cancellation + with contextlib.suppress(Exception): + await self._revoke_terminal_authority(task_id) + await self._stop_active_resources() + + async def _stop_active_resources(self) -> None: + resources = self._active_resources + self._active_resources = [] + seen: set[int] = set() + for resource in reversed(resources): + if id(resource) in seen: + continue + seen.add(id(resource)) + with contextlib.suppress(Exception): + await resource.stop() + + async def _wait_for_cancellation(self, task_id: uuid.UUID) -> bool: + while True: + await asyncio.sleep(settings.agent_cancellation_poll_seconds) + try: + assert self._cancellation_probe is not None + cancelled = await self._cancellation_probe(task_id) + except asyncio.CancelledError: + raise + except Exception: + log.warning("agent.cancellation_probe_failed", task_id=str(task_id)) + continue + if cancelled: + return True + + async def _finalize_cancellation(self, task_id: uuid.UUID) -> None: + await self.session.rollback() + task = await self.tasks.get(task_id) + if task is None or task.status != TaskStatus.CANCELLED.value: + return + if task.steps_used and not self._last_hash: + await self._rebuild_history(task.id) + task.stop_reason = StopReason.CANCELLED.value + if not task.summary: + task.summary = self._stop_summary(task, StopReason.CANCELLED) + self._ensure_unverified_receipt(task) + await self._commit() + await self._revoke_active_authority(task) + + async def _revoke_active_authority(self, task: TaskModel) -> None: + if self._authority_revoked or not self._active_authority_clients: + return + self._authority_revoked = True + events: list[dict[str, Any]] = [] + for service, client in self._active_authority_clients: + try: + event = await client.revoke() + if isinstance(event, list): + events.extend(event) + elif event: + events.append(event) + except Exception as exc: + events.append( + { + "kind": "authority", + "decision": "revocation_unavailable", + "run_id": self._active_run_id, + "service": service, + "error": type(exc).__name__, + } + ) + if not events: + return + task.authority_audit = [*(task.authority_audit or []), *events][-200:] + if task.receipt_hash and task.workspace_path: + try: + task.receipt_hash = refresh_receipt_authority( + Workspace(Path(task.workspace_path)), task.authority_audit + ) + except Exception: + log.warning("agent.receipt_authority_refresh_failed", task_id=str(task.id)) + await self._commit() + + async def _revoke_terminal_authority(self, task_id: uuid.UUID) -> None: + task = await self.tasks.get(task_id) + if task is not None and task.status in { + TaskStatus.COMPLETED.value, + TaskStatus.STOPPED.value, + TaskStatus.CANCELLED.value, + TaskStatus.FAILED.value, + }: + await self._revoke_active_authority(task) + + async def _run_claimed(self, task_id: uuid.UUID) -> None: """Run, or resume, a task. A task is resumable when it was paused on an ask_user question and the user has since answered (status back to pending with steps already on record).""" @@ -256,11 +385,29 @@ async def run(self, task_id: uuid.UUID) -> None: self._sandbox_image = None self._sandbox_backend = None self._egress_allowed = False + self._active_authority_clients = [] + self._active_run_id = None + self._authority_revoked = False workspace = Workspace( Path(task.workspace_path or settings.agent_workspaces_root) / ("" if task.workspace_path else str(task.id)) ) + if task.operation_journal is not None: + operation = task.operation_journal + tool = str(operation.get("tool", "unknown")) + operation_id = str(operation.get("id", "unknown")) + task.status = TaskStatus.FAILED.value + task.stop_reason = StopReason.ERROR.value + task.error = ( + f"Interrupted with {tool!r} operation {operation_id} in flight. Its outcome " + "is unknown, so Loop refused to replay it and risk a duplicate mutation." + ) + task.workspace_path = str(workspace.root) + await self._rebuild_history(task.id) + self._ensure_unverified_receipt(task) + await self._commit() + return self.memory = MemoryStore( scoped_memory_root(Path(settings.agent_memory_root), task.owner_id, task.project_id) ) @@ -540,6 +687,7 @@ def configured_provider_hosts(value: str) -> frozenset[str]: return run_id = f"{task.id}:{task.attempt}" + self._active_run_id = run_id def scoped_token_factory( granted_capabilities: set[Capability], @@ -574,6 +722,8 @@ def factory(audience: str) -> str: and settings.agent_egress_proxy_audit_url else None ) + if egress_audit is not None: + self._active_authority_clients.append(("egress-proxy", egress_audit)) if not isinstance(task.authority_audit, list): task.authority_audit = [] @@ -710,6 +860,7 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any ) ) provider_gateway = ProviderGatewayPool(gateway_clients) + self._active_authority_clients.append(("provider-gateway", provider_gateway)) try: await provider_gateway.start() except Exception as exc: @@ -720,6 +871,7 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any self._ensure_unverified_receipt(task) await self._commit() return + self._active_resources.append(provider_gateway) executor.provider_gateway = provider_gateway self._mcp_tools |= provider_gateway.tool_names self._browser_specs = provider_gateway.specs("net.browser") @@ -752,6 +904,7 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any if Capability.NET_BROWSER not in gateway_capabilities: browser = await self._start_browser(envelope) if browser is not None: + self._active_resources.append(browser) executor.mcp = browser self._browser_specs = browser.specs() self._mcp_tools |= set(browser.tool_names) @@ -864,6 +1017,7 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any self._ensure_unverified_receipt(task) await self._commit() return + self._active_resources.append(host_mcp) executor.auxiliary_mcp = host_mcp self._mcp_tools |= host_mcp.tool_names self._mcp_specs = host_mcp.specs() @@ -904,6 +1058,16 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any try: await self._run_loop(task, workspace, executor, start=task.steps_used + 1) + except asyncio.CancelledError: + await self.session.rollback() + await self.session.refresh(task) + if task.status == TaskStatus.CANCELLED.value: + task.stop_reason = StopReason.CANCELLED.value + if not task.summary: + task.summary = self._stop_summary(task, StopReason.CANCELLED) + self._ensure_unverified_receipt(task) + await self._commit() + raise except Exception as exc: # any unhandled error fails the task cleanly log.exception("agent.failed", task_id=str(task.id)) task.status = TaskStatus.FAILED.value @@ -922,45 +1086,7 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any TaskStatus.CANCELLED.value, TaskStatus.FAILED.value, }: - revocation_events: list[dict[str, Any]] = [] - for service, client in ( - ("provider-gateway", provider_gateway), - ("egress-proxy", egress_audit), - ): - if client is None: - continue - try: - event = await client.revoke() - if isinstance(event, list): - revocation_events.extend(event) - elif event: - revocation_events.append(event) - except Exception as exc: - revocation_events.append( - { - "kind": "authority", - "decision": "revocation_unavailable", - "run_id": run_id, - "service": service, - "error": type(exc).__name__, - } - ) - if revocation_events: - task.authority_audit = [ - *(task.authority_audit or []), - *revocation_events, - ][-200:] - if task.receipt_hash and task.workspace_path: - try: - task.receipt_hash = refresh_receipt_authority( - Workspace(Path(task.workspace_path)), task.authority_audit - ) - except Exception: - log.warning( - "agent.receipt_authority_refresh_failed", - task_id=str(task.id), - ) - await self._commit() + await self._revoke_active_authority(task) if provider_gateway is not None: await provider_gateway.stop() @@ -1152,6 +1278,9 @@ async def _start_browser(self, envelope: CapabilityEnvelope) -> McpBrowser | Non try: await browser.start() return browser + except asyncio.CancelledError: + await browser.stop() + raise except Exception: # browser unavailable -> run without it, don't fail the task log.warning("agent.browser_unavailable") await browser.stop() @@ -1207,13 +1336,22 @@ async def _run_loop( if task.pending_action is not None: action = dict(task.pending_action) task.pending_action = None - result = await executor.execute(str(action["tool"]), dict(action.get("args", {}))) + approved_tool = str(action["tool"]) + args = await self._begin_operation( + task, + number=start, + thought="(approved by the user)", + tool=approved_tool, + args=dict(action.get("args", {})), + tokens=0, + ) + result = await executor.execute(approved_tool, args) await self._record_step( task, start, "(approved by the user)", - str(action["tool"]), - dict(action.get("args", {})), + approved_tool, + args, result.observation, result.status, 0, @@ -1358,6 +1496,14 @@ async def _run_loop( if tool == "remember": note = str(args.get("note", "")).strip() topic = args.get("topic") + args = await self._begin_operation( + task, + number=number, + thought=thought, + tool=tool, + args=args, + tokens=step_tokens, + ) observation = self.memory.remember(note, str(topic) if topic else None) if note: # make it visible to the rest of this run too self._memory_snapshot = f"{self._memory_snapshot}\n- {note}".strip() @@ -1388,6 +1534,14 @@ async def _run_loop( step_tokens, ) else: + args = await self._begin_operation( + task, + number=number, + thought=thought, + tool=tool, + args=args, + tokens=step_tokens, + ) await self._handle_spawn(task, args, thought, number, step_tokens) last = self._history[-1] guard.observe(tool, args, last.observation, last.status) @@ -1468,9 +1622,25 @@ async def _run_loop( if verdict is Verdict.NEEDS_APPROVAL: await self._pause_for_approval(task, args, thought, number, step_tokens, reason) return # resumes when the user approves or denies + args = await self._begin_operation( + task, + number=number, + thought=thought, + tool=tool, + args=args, + tokens=step_tokens, + ) tool_result = await executor.execute(tool, args) observation, status = tool_result.observation, tool_result.status else: + args = await self._begin_operation( + task, + number=number, + thought=thought, + tool=tool, + args=args, + tokens=step_tokens, + ) tool_result = await executor.execute(tool, args) observation, status = tool_result.observation, tool_result.status if tool in ("write_file", "edit_file") and same_path_writes == 2: @@ -1753,7 +1923,11 @@ async def _handle_spawn( await self._commit() child_service = AgentReactService( - self.tasks, self.steps, self.llm, verifier_llm=self.verifier_llm + self.tasks, + self.steps, + self.llm, + verifier_llm=self.verifier_llm, + cancellation_probe=self._cancellation_probe, ) await child_service.run(child.id) await self.session.refresh(child) @@ -2016,6 +2190,32 @@ async def _handle_finish( # --- Persistence helpers --------------------------------------------- + async def _begin_operation( + self, + task: TaskModel, + *, + number: int, + thought: str, + tool: str, + args: dict[str, Any], + tokens: int, + ) -> dict[str, Any]: + operation_id = str(uuid.uuid4()) + execution_args = dict(args) + if tool in {"send_email", "create_event"}: + execution_args["operation_id"] = operation_id + task.operation_journal = { + "schema": "loop.operation/v1", + "id": operation_id, + "step": number, + "thought": thought, + "tool": tool, + "args": execution_args, + "tokens": tokens, + } + await self._commit() + return execution_args + @staticmethod def _record_model_use(task: TaskModel, result: LLMResult, *, verifier: bool = False) -> None: identity = {"provider": result.provider, "model": result.model} @@ -2077,6 +2277,7 @@ async def _record_step( hash=this_hash, ) self._last_hash = this_hash + task.operation_journal = None task.steps_used = number task.tokens_used += tokens await self._commit() diff --git a/apps/api/app/services/changeset.py b/apps/api/app/services/changeset.py index e3b1273..35ca50d 100644 --- a/apps/api/app/services/changeset.py +++ b/apps/api/app/services/changeset.py @@ -7,16 +7,16 @@ import shutil import subprocess import tempfile -import time import uuid from dataclasses import dataclass from pathlib import Path +from filelock import FileLock, Timeout + from app.core.config import settings from app.exceptions import ConflictError, ValidationError _RECEIPT_PATHS = ("receipt.json", "RECEIPT.md") -_STALE_LOCK_SECONDS = 600 def _git_environment(extra: dict[str, str] | None = None) -> dict[str, str]: @@ -367,29 +367,23 @@ def _metadata_root() -> Path: return root -def acquire_source_lock(source_path: str) -> Path: +def acquire_source_lock(source_path: str) -> FileLock: digest = hashlib.sha256(str(Path(source_path).resolve()).encode()).hexdigest() - lock = _metadata_root() / f"source-{digest}.lock" - for attempt in range(2): - try: - lock.mkdir() - return lock - except FileExistsError as exc: - try: - stale = time.time() - lock.stat().st_mtime > _STALE_LOCK_SECONDS - except OSError: - stale = False - if attempt == 0 and stale: - shutil.rmtree(lock, ignore_errors=True) - continue - raise ConflictError( - "Another change-set operation is in progress for this project." - ) from exc - raise ConflictError("Could not acquire the project change-set lock.") + lock = FileLock( + _metadata_root() / f"source-{digest}.filelock", + thread_local=False, + ) + try: + lock.acquire(timeout=0) + except Timeout as exc: + raise ConflictError( + "Another change-set operation is in progress for this project." + ) from exc + return lock -def release_source_lock(lock: Path) -> None: - shutil.rmtree(lock, ignore_errors=True) +def release_source_lock(lock: FileLock) -> None: + lock.release() def patch_path(task_id: uuid.UUID) -> Path: diff --git a/apps/api/app/services/runner.py b/apps/api/app/services/runner.py index c17d6d6..5917a93 100644 --- a/apps/api/app/services/runner.py +++ b/apps/api/app/services/runner.py @@ -69,6 +69,16 @@ async def _heartbeat_task(task_id: uuid.UUID, stop: asyncio.Event) -> None: await session.commit() +async def _task_cancelled(task_id: uuid.UUID) -> bool: + from sqlalchemy import select + + from app.db.models.task import TaskModel + + async with get_sessionmaker()() as session: + status = await session.scalar(select(TaskModel.status).where(TaskModel.id == task_id)) + return status == TaskStatus.CANCELLED.value + + async def reconcile_interrupted_tasks( session: AsyncSession, *, stale_seconds: int = 0, requeue: bool = False ) -> int: @@ -153,7 +163,11 @@ async def execute_task(task_id: uuid.UUID) -> None: tasks = TaskRepository(session) steps = StepRepository(session) service = AgentReactService( - tasks, steps, get_llm_client(), verifier_llm=get_verifier_client() + tasks, + steps, + get_llm_client(), + verifier_llm=get_verifier_client(), + cancellation_probe=_task_cancelled, ) stop = asyncio.Event() heartbeat = asyncio.create_task(_heartbeat_task(task_id, stop)) diff --git a/apps/api/app/services/task.py b/apps/api/app/services/task.py index 4cef88b..2f4a90d 100644 --- a/apps/api/app/services/task.py +++ b/apps/api/app/services/task.py @@ -722,6 +722,9 @@ async def cancel(self, task_id: uuid.UUID) -> TaskModel: if task.status not in active: raise ConflictError(f"Task is {task.status} and cannot be cancelled") task.status = TaskStatus.CANCELLED.value + for descendant in await self.tasks.list_descendants(task.id): + if descendant.status in active: + descendant.status = TaskStatus.CANCELLED.value await self.tasks.session.flush() await self.tasks.session.refresh(task) return task diff --git a/apps/api/app/tools/calendar.py b/apps/api/app/tools/calendar.py index 74ca076..29ddfdf 100644 --- a/apps/api/app/tools/calendar.py +++ b/apps/api/app/tools/calendar.py @@ -69,6 +69,10 @@ def _create(self, args: dict[str, Any]) -> str: summary = str(args.get("summary", "")).strip() if not summary: return "create_event needs a 'summary'." + try: + operation_id = str(uuid.UUID(str(args["operation_id"]))) + except (KeyError, ValueError): + return "create_event requires a Loop operation_id for duplicate protection." try: start = datetime.fromisoformat(str(args["start"])) end = ( @@ -86,7 +90,7 @@ def _create(self, args: dict[str, Any]) -> str: event.add("summary", summary) event.add("dtstart", start) event.add("dtend", end) - event.add("uid", f"{uuid.uuid4().hex}@loop") + event.add("uid", f"{operation_id}@loop") if args.get("description"): event.add("description", str(args["description"])) obj.add_component(event) diff --git a/apps/api/app/tools/email.py b/apps/api/app/tools/email.py index 85bd083..43dc12d 100644 --- a/apps/api/app/tools/email.py +++ b/apps/api/app/tools/email.py @@ -14,6 +14,7 @@ import imaplib import smtplib import ssl +import uuid from email.message import EmailMessage from typing import Any, ClassVar @@ -37,10 +38,15 @@ def _send(self, args: dict[str, Any]) -> str: to = str(args.get("to", "")).strip() if not to: return "send_email needs a 'to' address." + try: + operation_id = str(uuid.UUID(str(args["operation_id"]))) + except (KeyError, ValueError): + return "send_email requires a Loop operation_id for duplicate protection." msg = EmailMessage() msg["From"] = settings.email_from or settings.smtp_user or "" msg["To"] = to msg["Subject"] = str(args.get("subject", "")).strip() + msg["Message-ID"] = f"<{operation_id}@loop>" msg.set_content(str(args.get("body", ""))) with smtplib.SMTP(settings.smtp_host or "", settings.smtp_port, timeout=30) as server: if settings.smtp_starttls: diff --git a/apps/api/app/tools/mcp/stdio.py b/apps/api/app/tools/mcp/stdio.py index 6603ca6..60e3719 100644 --- a/apps/api/app/tools/mcp/stdio.py +++ b/apps/api/app/tools/mcp/stdio.py @@ -132,6 +132,10 @@ async def start(self) -> None: raise RuntimeError(f"Duplicate MCP tool names: {sorted(overlap)}") self.tool_names |= provider.tool_names self._by_tool.update(dict.fromkeys(provider.tool_names, provider)) + except asyncio.CancelledError: + for provider in reversed(started): + await provider.stop() + raise except Exception: for provider in reversed(started): await provider.stop() diff --git a/apps/api/app/tools/provider_gateway.py b/apps/api/app/tools/provider_gateway.py index 7a9977f..8c538d4 100644 --- a/apps/api/app/tools/provider_gateway.py +++ b/apps/api/app/tools/provider_gateway.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import base64 import contextlib from collections.abc import Callable @@ -53,6 +54,9 @@ async def start(self) -> None: self.tool_names = { str(item["name"]) for item in self.tools if isinstance(item.get("name"), str) } + except asyncio.CancelledError: + await self.client.aclose() + raise except Exception: await self.client.aclose() raise @@ -149,6 +153,10 @@ async def start(self) -> None: self.tools.extend(client.tools) self.tool_names.update(client.tool_names) self._routes.update(dict.fromkeys(client.tool_names, client)) + except asyncio.CancelledError: + for client in reversed(started): + await client.stop() + raise except Exception: for client in reversed(started): await client.stop() diff --git a/apps/api/app/tools/shell.py b/apps/api/app/tools/shell.py index c7f2cd5..8d1fa80 100644 --- a/apps/api/app/tools/shell.py +++ b/apps/api/app/tools/shell.py @@ -86,6 +86,11 @@ async def drain() -> bytes: try: raw = await asyncio.wait_for(drain(), timeout=timeout_seconds) + except asyncio.CancelledError: + kill_process_group(proc) + with contextlib.suppress(Exception): + await asyncio.shield(proc.wait()) + raise except TimeoutError: kill_process_group(proc) with contextlib.suppress(Exception): diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml index 9e6b8bd..41345d3 100644 --- a/apps/api/pyproject.toml +++ b/apps/api/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [ "pyjwt>=2.10", "prometheus-client>=0.21", "kubernetes-asyncio>=36.1,<37", + "filelock>=3.20,<4", # --- Observability (OpenTelemetry) --- "opentelemetry-sdk>=1.43", "opentelemetry-exporter-otlp>=1.43", diff --git a/apps/api/requirements.lock b/apps/api/requirements.lock index 41186b8..0760afd 100644 --- a/apps/api/requirements.lock +++ b/apps/api/requirements.lock @@ -373,6 +373,10 @@ fastapi==0.139.0 \ --hash=sha256:99ab7b2d92223c76d6cf10757ab3f89d45b38267fc20b2a136cf02f6beac3145 \ --hash=sha256:cf15e1e9e667ddb0ad63811e60bd11390d1aac838ca4a7a23f421807b2308189 # via loop-api +filelock==3.29.7 \ + --hash=sha256:5b481979797ae69e72f0b389d89a80bdd585c260c5b3f1fb9c0a5ba9bb3f195d \ + --hash=sha256:987db6f789a3a2a59f55081801b2b3697cb97e2a736b5f1a9e99b559285fbc51 + # via loop-api frozenlist==1.8.0 \ --hash=sha256:0325024fe97f94c41c08872db482cf8ac4800d80e79222c6b0b7b162d5b13686 \ --hash=sha256:032efa2674356903cd0261c4317a561a6850f3ac864a63fc1583147fb05a79b0 \ diff --git a/apps/api/tests/test_agent_react.py b/apps/api/tests/test_agent_react.py index b7f33e6..848d529 100644 --- a/apps/api/tests/test_agent_react.py +++ b/apps/api/tests/test_agent_react.py @@ -4,7 +4,9 @@ from __future__ import annotations +import asyncio import json +import os from pathlib import Path from typing import Any @@ -173,6 +175,155 @@ async def test_goal_achieved_when_verifier_accepts_finish(session: AsyncSession) assert task.summary == "wrote result.txt" # The file the agent "wrote" really exists in its workspace. assert (Path(task.workspace_path) / "result.txt").read_text() == "done" + assert task.operation_journal is None + + +async def test_running_shell_is_killed_when_task_is_cancelled( + session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + command = ( + 'python3 -c "import pathlib,subprocess,time; ' + "p=subprocess.Popen(['sleep','30']); " + "pathlib.Path('child.pid').write_text(str(p.pid)); time.sleep(30)\"" + ) + task = await _make_task(session, max_steps=4, token_budget=1_000_000) + requested = False + + async def cancellation_probe(_task_id: object) -> bool: + if not requested: + return False + task.status = TaskStatus.CANCELLED.value + await session.commit() + return True + + monkeypatch.setattr(settings, "agent_sandbox", "off") + monkeypatch.setattr(settings, "agent_cancellation_poll_seconds", 0.05) + service = AgentReactService( + TaskRepository(session), + StepRepository(session), + ScriptedLLM([{"thought": "run", "tool": "run_command", "args": {"command": command}}]), + cancellation_probe=cancellation_probe, + ) + running = asyncio.create_task(service.run(task.id)) + pid_file = Path(settings.agent_workspaces_root) / str(task.id) / "child.pid" + for _ in range(100): + if pid_file.is_file(): + break + await asyncio.sleep(0.02) + assert pid_file.is_file() + child_pid = int(pid_file.read_text()) + + requested = True + await asyncio.wait_for(running, timeout=2) + + for _ in range(50): + try: + os.kill(child_pid, 0) + except ProcessLookupError: + break + await asyncio.sleep(0.02) + with pytest.raises(ProcessLookupError): + os.kill(child_pid, 0) + await session.refresh(task) + assert task.status == TaskStatus.CANCELLED.value + assert task.stop_reason == StopReason.CANCELLED.value + assert (Path(task.workspace_path) / "receipt.json").is_file() + + +async def test_interrupted_mutation_is_not_replayed( + session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + import app.services.agent_react as agent_module + + task = await _make_task(session, max_steps=4, token_budget=1_000_000) + calls = 0 + + async def crash_after_mutation(executor: object, _tool: str, _args: dict[str, Any]) -> object: + nonlocal calls + calls += 1 + workspace = executor.workspace # type: ignore[attr-defined] + with (workspace.root / "mutation.log").open("a") as output: + output.write("mutated\n") + raise asyncio.CancelledError + + monkeypatch.setattr(agent_module.ToolExecutor, "execute", crash_after_mutation) + plans = [{"thought": "mutate", "tool": "run_command", "args": {"command": "echo x"}}] + with pytest.raises(asyncio.CancelledError): + await _service(session, ScriptedLLM(plans)).run(task.id) + + await session.refresh(task) + assert task.operation_journal is not None + task.status = TaskStatus.PENDING.value + await session.commit() + + await _service(session, ScriptedLLM(plans)).run(task.id) + + await session.refresh(task) + assert calls == 1 + assert (Path(task.workspace_path) / "mutation.log").read_text() == "mutated\n" + assert task.status == TaskStatus.FAILED.value + assert "refused to replay" in (task.error or "") + + +async def test_receipt_write_failure_cannot_false_complete( + session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + import app.services.agent_react as agent_module + + def fail_receipt(*_args: object, **_kwargs: object) -> object: + raise OSError("disk unavailable") + + monkeypatch.setattr(agent_module, "build_receipt", fail_receipt) + task = await _make_task(session, max_steps=4, token_budget=1_000_000) + plans = [{"thought": "done", "tool": "finish", "args": {"summary": "done"}}] + + await _service(session, ScriptedLLM(plans)).run(task.id) + + await session.refresh(task) + assert task.status == TaskStatus.FAILED.value + assert task.receipt_hash is None + assert "disk unavailable" in (task.error or "") + + +async def test_cancelling_parent_cancels_active_descendants(session: AsyncSession) -> None: + parent = await _make_task(session, max_steps=4, token_budget=10_000) + repo = TaskRepository(session) + child = await repo.create( + goal="child", + status=TaskStatus.RUNNING.value, + parent_id=parent.id, + depth=1, + rubric=[], + max_steps=4, + token_budget=10_000, + summary=None, + verification_score=0, + steps_used=0, + tokens_used=0, + workspace_path=None, + ) + grandchild = await repo.create( + goal="grandchild", + status=TaskStatus.AWAITING_INPUT.value, + parent_id=child.id, + depth=2, + rubric=[], + max_steps=4, + token_budget=10_000, + summary=None, + verification_score=0, + steps_used=0, + tokens_used=0, + workspace_path=None, + ) + await session.commit() + + await TaskService(repo, StepRepository(session)).cancel(parent.id) + await session.commit() + + for item in (parent, child, grandchild): + await session.refresh(item) + assert item.status == TaskStatus.CANCELLED.value async def test_agent_scopes_protocol_and_browser_gateway_tokens( diff --git a/apps/api/tests/test_calendar.py b/apps/api/tests/test_calendar.py index b1fcba4..6b79772 100644 --- a/apps/api/tests/test_calendar.py +++ b/apps/api/tests/test_calendar.py @@ -43,10 +43,18 @@ async def test_create_event(monkeypatch: pytest.MonkeyPatch) -> None: tools = CalendarTools() fake = _FakeCalendar() monkeypatch.setattr(tools, "_calendar", lambda: fake) - out = await tools.call("create_event", {"summary": "Dentist", "start": "2026-07-02T15:00:00"}) + out = await tools.call( + "create_event", + { + "summary": "Dentist", + "start": "2026-07-02T15:00:00", + "operation_id": "66e20fcc-5887-46f5-b315-9c6b2630fec4", + }, + ) assert "Event created" in out and "Dentist" in out assert len(fake.saved) == 1 assert "Dentist" in fake.saved[0] # the iCalendar payload carries the summary + assert "66e20fcc-5887-46f5-b315-9c6b2630fec4@loop" in fake.saved[0] async def test_create_event_needs_summary() -> None: @@ -55,5 +63,19 @@ async def test_create_event_needs_summary() -> None: async def test_create_event_needs_valid_start() -> None: - out = await CalendarTools().call("create_event", {"summary": "X", "start": "not-a-date"}) + out = await CalendarTools().call( + "create_event", + { + "summary": "X", + "start": "not-a-date", + "operation_id": "c9a37b93-288c-4e7a-ab2f-146634d52639", + }, + ) assert "ISO 'start'" in out + + +async def test_create_event_requires_operation_id() -> None: + out = await CalendarTools().call( + "create_event", {"summary": "X", "start": "2026-07-02T15:00:00"} + ) + assert "operation_id" in out diff --git a/apps/api/tests/test_changeset.py b/apps/api/tests/test_changeset.py index ebb5743..f6f76d6 100644 --- a/apps/api/tests/test_changeset.py +++ b/apps/api/tests/test_changeset.py @@ -1,10 +1,12 @@ from __future__ import annotations +import asyncio import subprocess import uuid from pathlib import Path import pytest +from filelock import FileLock from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker @@ -41,6 +43,29 @@ def _repository(root: Path, name: str = "project") -> Path: return repo +async def test_source_lock_admits_one_of_twenty_concurrent_writers( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + workspaces = tmp_path / "workspaces" + monkeypatch.setattr(settings, "agent_workspaces_root", str(workspaces)) + source = tmp_path / "source" + source.mkdir() + + async def acquire() -> FileLock | None: + try: + return await asyncio.to_thread(acquire_source_lock, str(source)) + except Exception as exc: + assert "in progress" in str(exc) + return None + + locks = await asyncio.gather(*(acquire() for _ in range(20))) + acquired = [lock for lock in locks if lock is not None] + assert len(acquired) == 1 + release_source_lock(acquired[0]) + reacquired = await asyncio.to_thread(acquire_source_lock, str(source)) + await asyncio.to_thread(release_source_lock, reacquired) + + async def _mark_verified(engine: AsyncEngine, task_id: str) -> None: sessionmaker = async_sessionmaker(engine, expire_on_commit=False) async with sessionmaker() as session: diff --git a/apps/api/tests/test_deployment_boundaries.py b/apps/api/tests/test_deployment_boundaries.py index 4eccf4e..1b96a3c 100644 --- a/apps/api/tests/test_deployment_boundaries.py +++ b/apps/api/tests/test_deployment_boundaries.py @@ -305,7 +305,8 @@ def test_kubernetes_acceptance_migrates_runs_task_and_rolls_back() -> None: assert "kubectl rollout undo deployment/api" in script assert 'task["sandbox"] != "kubernetes"' in script assert 'report.get("authentic")' in script - assert "0009_contract_draft" in script + assert "0010_operation_journal" in script + assert "run_cluster_probe after-postgres-restart true" in script assert "api web worker" in smoke assert 'cluster="${LOOP_ACCEPTANCE_CLUSTER:-la-' in script assert 'registry_container="$registry"' in script diff --git a/apps/api/tests/test_email.py b/apps/api/tests/test_email.py index 56f5540..0c0678b 100644 --- a/apps/api/tests/test_email.py +++ b/apps/api/tests/test_email.py @@ -46,7 +46,13 @@ async def test_send_email(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("app.tools.email.smtplib.SMTP", _FakeSMTP) out = await EmailTools().call( - "send_email", {"to": "you@example.com", "subject": "Hi", "body": "hello there"} + "send_email", + { + "to": "you@example.com", + "subject": "Hi", + "body": "hello there", + "operation_id": "47e20763-0997-43fe-ab29-c36112e9f495", + }, ) assert "sent to you@example.com" in out sent = _FakeSMTP.last @@ -54,6 +60,7 @@ async def test_send_email(monkeypatch: pytest.MonkeyPatch) -> None: assert sent.creds == ("me@example.com", "app-pass") assert sent.msg["To"] == "you@example.com" assert sent.msg["Subject"] == "Hi" + assert sent.msg["Message-ID"] == "<47e20763-0997-43fe-ab29-c36112e9f495@loop>" assert "hello there" in sent.msg.get_content() @@ -62,6 +69,11 @@ async def test_send_email_requires_to() -> None: assert "needs a 'to'" in out +async def test_send_email_requires_operation_id() -> None: + out = await EmailTools().call("send_email", {"to": "you@example.com"}) + assert "operation_id" in out + + class _FakeIMAP: def __init__(self, host: str) -> None: self.host = host diff --git a/apps/api/tests/test_mcp.py b/apps/api/tests/test_mcp.py index 24212ef..b2e1862 100644 --- a/apps/api/tests/test_mcp.py +++ b/apps/api/tests/test_mcp.py @@ -3,10 +3,13 @@ from __future__ import annotations +import asyncio from pathlib import Path from types import SimpleNamespace from typing import ClassVar +import pytest + from app.domain.capability import Capability from app.tools import CapabilityEnvelope, ToolExecutor, ToolStatus, Workspace from app.tools.mcp import McpBrowser, McpPool, McpStdioProvider @@ -24,6 +27,29 @@ async def call(self, name: str, args: dict) -> str: return f"navigated to {args.get('url')}" +async def test_pool_cancellation_stops_partially_started_provider() -> None: + started = asyncio.Event() + stopped = asyncio.Event() + + class BlockingProvider: + tool_names: ClassVar[set[str]] = set() + + async def start(self) -> None: + started.set() + await asyncio.Event().wait() + + async def stop(self) -> None: + stopped.set() + + pool = McpPool([BlockingProvider()]) # type: ignore[list-item] + running = asyncio.create_task(pool.start()) + await started.wait() + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert stopped.is_set() + + async def test_executor_dispatches_mcp_tool(tmp_path: Path) -> None: mcp = _FakeMcp() ex = ToolExecutor( diff --git a/apps/api/tests/test_provider_gateway.py b/apps/api/tests/test_provider_gateway.py index 15dbb74..9c2a9f9 100644 --- a/apps/api/tests/test_provider_gateway.py +++ b/apps/api/tests/test_provider_gateway.py @@ -141,6 +141,51 @@ async def test_gateway_rejects_unsigned_requests() -> None: assert response.status_code == 401 +async def test_gateway_requires_operation_id_for_external_write( + monkeypatch: pytest.MonkeyPatch, +) -> None: + private, public = _keys() + app = create_app( + ProviderGatewaySettings( + authority_public_key=public, + browser_enabled=False, + smtp_host="smtp.example.com", + smtp_user="me@example.com", + smtp_password="secret", + ) + ) + runtime: ProviderGatewayRuntime = app.state.runtime + called = False + + async def fake_send(_name: str, _args: dict[str, Any], _token: str | None) -> str: + nonlocal called + called = True + return "sent" + + monkeypatch.setattr(runtime.email, "call", fake_send) + capabilities = [Capability.EMAIL_SEND] + hosts = ["smtp.example.com"] + headers = { + "Authorization": f"Bearer {_token(private, capabilities, egress_hosts=hosts)}", + "X-Loop-Egress-Token": _token( + private, + capabilities, + audience=EGRESS_PROXY_AUDIENCE, + egress_hosts=hosts, + ), + } + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://gateway") as client: + response = await client.post( + "/v1/tools/send_email", + headers=headers, + json={"args": {"to": "recipient@example.com"}}, + ) + assert response.status_code == 403 + assert "operation_id" in response.json()["detail"] + assert called is False + + async def test_browser_gateway_rejects_provider_audience_token() -> None: private, public = _keys() app = create_app( @@ -511,6 +556,58 @@ def token_factory(audience: str) -> str: await gateway.client.aclose() +async def test_gateway_client_cancellation_aborts_inflight_request(tmp_path) -> None: + started = asyncio.Event() + cancelled = asyncio.Event() + + async def handler(_request: httpx.Request) -> httpx.Response: + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + return httpx.Response(200, json={"result": "late"}) + + gateway = ProviderGatewayClient( + "http://gateway", + Workspace(tmp_path / "workspace"), + lambda _audience: "token", + ) + await gateway.client.aclose() + gateway.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + running = asyncio.create_task(gateway.call("read_inbox", {})) + await started.wait() + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert cancelled.is_set() + await gateway.client.aclose() + + +async def test_gateway_start_cancellation_closes_transport(tmp_path) -> None: + started = asyncio.Event() + + async def handler(_request: httpx.Request) -> httpx.Response: + started.set() + await asyncio.Event().wait() + return httpx.Response(200, json={"tools": []}) + + gateway = ProviderGatewayClient( + "http://gateway", + Workspace(tmp_path / "workspace"), + lambda _audience: "token", + ) + await gateway.client.aclose() + gateway.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + running = asyncio.create_task(gateway.start()) + await started.wait() + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert gateway.client.is_closed + + async def test_gateway_clients_separate_audiences_and_egress_authority(tmp_path) -> None: requested: list[str] = [] @@ -866,7 +963,12 @@ async def resolver(host: str, port: int) -> tuple[int, str]: try: result = await EmailProvider(settings).call( "send_email", - {"to": "recipient@example.com", "subject": "Proxy proof", "body": "Delivered"}, + { + "to": "recipient@example.com", + "subject": "Proxy proof", + "body": "Delivered", + "operation_id": "1a9ffad3-a357-49ce-a556-94c93dbe17a4", + }, token, ) finally: diff --git a/apps/api/tests/test_runner.py b/apps/api/tests/test_runner.py index 567277f..c00bc2c 100644 --- a/apps/api/tests/test_runner.py +++ b/apps/api/tests/test_runner.py @@ -12,7 +12,7 @@ async def test_execute_task_bounds_concurrency(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(settings, "agent_max_concurrent_runs", 2) + monkeypatch.setattr(settings, "agent_max_concurrent_runs", 4) runner._run_gates.clear() # fresh gate at the new limit active = 0 @@ -40,9 +40,46 @@ async def __aexit__(self, *_a: object) -> bool: monkeypatch.setattr(runner, "get_sessionmaker", lambda: _NullSession) monkeypatch.setattr(runner, "get_llm_client", lambda: None) - await asyncio.gather(*(runner.execute_task(uuid4()) for _ in range(6))) + await asyncio.gather(*(runner.execute_task(uuid4()) for _ in range(20))) - assert peak == 2 # reached the cap, never exceeded it + assert peak == 4 # reached the cap, never exceeded it + + +async def test_duplicate_delivery_claims_one_task_exactly_once(tmp_path) -> None: + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + import app.db.models as _models # noqa: F401 + from app.db.base import Base + from app.repositories.task import TaskRepository + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'claims.db'}") + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + sessions = async_sessionmaker(engine, expire_on_commit=False, autoflush=False) + async with sessions() as session: + task = await TaskRepository(session).create( + goal="claim once", + status="pending", + rubric=[], + max_steps=4, + token_budget=10_000, + summary=None, + verification_score=0, + steps_used=0, + tokens_used=0, + workspace_path=None, + ) + await session.commit() + + async def claim() -> bool: + async with sessions() as session: + return await TaskRepository(session).claim_pending(task.id) is not None + + try: + claims = await asyncio.gather(*(claim() for _ in range(20))) + assert sum(claims) == 1 + finally: + await engine.dispose() async def test_worker_run_task_handler_dispatches(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/apps/api/tests/test_sandbox.py b/apps/api/tests/test_sandbox.py index 8ee7c6d..2dd2095 100644 --- a/apps/api/tests/test_sandbox.py +++ b/apps/api/tests/test_sandbox.py @@ -5,10 +5,12 @@ from __future__ import annotations +import asyncio from pathlib import Path import pytest +from app.core.config import settings from app.tools import CapabilityEnvelope, ToolExecutor, ToolResult, ToolStatus, Workspace from app.tools.egress import resolve_proxy_endpoint from app.tools.sandbox import image_present @@ -192,6 +194,117 @@ async def fake_remove(_name: str) -> None: ) +async def test_cancelling_docker_command_force_removes_container( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import app.tools.sandbox as sandbox + + started = asyncio.Event() + removed = asyncio.Event() + + async def fake_create(*_argv: str, **_kwargs: object) -> object: + return object() + + async def blocked_collect( + _proc: object, *, timeout_seconds: int, output_limit: int + ) -> tuple[bytes, int]: + del timeout_seconds, output_limit + started.set() + await asyncio.Event().wait() + return b"", 0 + + async def fake_remove(_name: str) -> None: + removed.set() + + monkeypatch.setattr(sandbox.asyncio, "create_subprocess_exec", fake_create) + monkeypatch.setattr(sandbox, "collect_output", blocked_collect) + monkeypatch.setattr(sandbox, "_force_remove", fake_remove) + workspace = tmp_path / "workspace" + workspace.mkdir() + + running = asyncio.create_task( + sandbox.run_command_in_container( + "sleep 30", + workspace, + image="sandbox:latest", + network=False, + ) + ) + await started.wait() + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert removed.is_set() + + +async def test_cancelling_kubernetes_command_deletes_job( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from kubernetes_asyncio import client, config + + from app.tools.kubernetes_sandbox import run_command_in_kubernetes + + created = asyncio.Event() + deleted = asyncio.Event() + closed = asyncio.Event() + + class FakeApiClient: + async def close(self) -> None: + closed.set() + + class FakeBatch: + def __init__(self, _client: object) -> None: + pass + + async def create_namespaced_job(self, **_kwargs: object) -> None: + created.set() + + async def read_namespaced_job_status(self, **_kwargs: object) -> object: + class Status: + succeeded = 0 + failed = 0 + + class Job: + status = Status() + + return Job() + + async def delete_namespaced_job(self, **_kwargs: object) -> None: + deleted.set() + + class FakeCore: + def __init__(self, _client: object) -> None: + pass + + mount = tmp_path / "data" + workspace = mount / "workspaces" / "task-id" + workspace.mkdir(parents=True) + monkeypatch.setattr(settings, "agent_kubernetes_data_mount", str(mount)) + monkeypatch.setattr(config, "load_incluster_config", lambda: None) + monkeypatch.setattr(client, "ApiClient", FakeApiClient) + monkeypatch.setattr(client, "BatchV1Api", FakeBatch) + monkeypatch.setattr(client, "CoreV1Api", FakeCore) + + running = asyncio.create_task( + run_command_in_kubernetes( + "sleep 30", + workspace, + image="sandbox:latest", + network=False, + timeout_seconds=60, + output_limit=4000, + memory="512m", + cpus="1", + ) + ) + await created.wait() + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert deleted.is_set() + assert closed.is_set() + + def test_docker_available_does_not_cache_negative(monkeypatch: pytest.MonkeyPatch) -> None: import app.tools.sandbox as sb diff --git a/apps/api/tests/test_tools.py b/apps/api/tests/test_tools.py index e0a0303..e113448 100644 --- a/apps/api/tests/test_tools.py +++ b/apps/api/tests/test_tools.py @@ -4,6 +4,7 @@ from __future__ import annotations +import asyncio from pathlib import Path import pytest @@ -164,6 +165,24 @@ async def test_run_command_reports_nonzero_exit(tmp_path: Path) -> None: assert "exit code 3" in result.observation +async def test_twenty_concurrent_workspaces_do_not_leak(tmp_path: Path) -> None: + async def write(index: int) -> None: + executor = ToolExecutor( + Workspace(tmp_path / f"task-{index}"), + envelope=CapabilityEnvelope.from_capabilities(["fs.write"]), + ) + result = await executor.execute( + "write_file", {"path": "result.txt", "content": f"task-{index}"} + ) + assert result.status is ToolStatus.OK + + await asyncio.gather(*(write(index) for index in range(20))) + + assert {(tmp_path / f"task-{index}" / "result.txt").read_text() for index in range(20)} == { + f"task-{index}" for index in range(20) + } + + def test_envelope_permits_logic() -> None: full = CapabilityEnvelope.from_tools(None) assert full.permits("run_command") is True diff --git a/apps/api/tests/test_worker_queue.py b/apps/api/tests/test_worker_queue.py index 3a81c00..62f2869 100644 --- a/apps/api/tests/test_worker_queue.py +++ b/apps/api/tests/test_worker_queue.py @@ -42,6 +42,16 @@ def pipeline(self, **_kwargs: Any) -> FakePipeline: return FakePipeline(self.actions) +class AckFailurePipeline(FakePipeline): + async def execute(self) -> list[object]: + raise ConnectionError("Redis disappeared before acknowledgement") + + +class AckFailureRedis(FakeRedis): + def pipeline(self, **_kwargs: Any) -> FakePipeline: + return AckFailurePipeline(self.actions) + + async def failing_handler(_payload: dict[str, Any]) -> None: raise RuntimeError("transient failure") @@ -107,6 +117,40 @@ async def test_successful_job_is_acknowledged_and_removed( assert [action[0] for action in redis.actions] == ["xack", "xdel"] +async def test_acknowledgement_crash_leaves_job_reclaimable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = 0 + + async def handled(_payload: dict[str, Any]) -> None: + nonlocal calls + calls += 1 + + async def reconciled() -> None: + return None + + monkeypatch.setattr(worker, "_reset_stale_tasks", reconciled) + monkeypatch.setitem(worker.HANDLERS, "ack-crash", handled) + with pytest.raises(ConnectionError, match="before acknowledgement"): + await worker._process( + AckFailureRedis(), # type: ignore[arg-type] + "3-1", + {"type": "ack-crash", "payload": "{}", "attempt": "1"}, + reclaimed=False, + ) + assert calls == 1 + + reclaimed = FakeRedis() + await worker._process( + reclaimed, # type: ignore[arg-type] + "3-1", + {"type": "ack-crash", "payload": "{}", "attempt": "1"}, + reclaimed=True, + ) + assert calls == 2 + assert [action[0] for action in reclaimed.actions] == ["xack", "xdel"] + + async def test_reclaimed_job_reconciles_stale_tasks_before_execution( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/apps/api/uv.lock b/apps/api/uv.lock index 14db5ab..631c892 100644 --- a/apps/api/uv.lock +++ b/apps/api/uv.lock @@ -1227,6 +1227,7 @@ dependencies = [ { name = "asyncpg" }, { name = "cryptography" }, { name = "fastapi" }, + { name = "filelock" }, { name = "httpx" }, { name = "kubernetes-asyncio" }, { name = "mcp" }, @@ -1278,6 +1279,7 @@ requires-dist = [ { name = "cryptography", specifier = ">=42" }, { name = "fakeredis", marker = "extra == 'dev'", specifier = "==2.36.2" }, { name = "fastapi", specifier = ">=0.136,<1.0" }, + { name = "filelock", specifier = ">=3.20,<4" }, { name = "httpx", specifier = ">=0.28" }, { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.28" }, { name = "icalendar", marker = "extra == 'calendar'", specifier = ">=5.0" }, diff --git a/docs/STRATEGY.md b/docs/STRATEGY.md index aaa03cd..287c3f4 100644 --- a/docs/STRATEGY.md +++ b/docs/STRATEGY.md @@ -157,6 +157,23 @@ strict verified change set, and Apply/Undo without manually authored criteria. ### Gate 2 — Cancellation, recovery, and concurrency +**Implementation status (2026-07-20): gate complete.** Every run has a database-backed +cancellation watcher. Cancellation tears through the active model/tool coroutine and +its delegated task tree; host process groups, Docker containers, Kubernetes Jobs, and +gateway requests have explicit cleanup tests. Tool execution uses a durable +`loop.operation/v1` write-ahead journal: a crash after the possible side effect but +before the hash-chained Step commit leaves an unknown outcome that recovery refuses to +replay. This deliberately provides at-most-once mutation, not a false exactly-once +claim. SMTP Message-ID and CalDAV UID receive stable operation ids, while already +accepted upstream effects remain non-reversible. + +The automated fault matrix covers a failed queue acknowledgement, abandoned Redis +delivery and Redis restart, Postgres container restart, terminal Receipt write failure, +20 duplicate claims, 20 isolated workspaces, 20 source-lock contenders, and 20 runs +through the resource gate. Atomic pending-to-running claims prevent duplicate queue +delivery from executing a run twice; OS-backed project locks serialize Apply/Undo +across processes without stale-lock eviction. + - Propagate cancellation into in-flight provider calls, shell process groups, Docker containers, Kubernetes Jobs, gateways, and delegated work. - Inject crashes around plan persistence, tool completion, verification, queue diff --git a/docs/guides/security.md b/docs/guides/security.md index d6492e5..5fd6f61 100644 --- a/docs/guides/security.md +++ b/docs/guides/security.md @@ -85,8 +85,16 @@ not prompt instructions. session backend. - Revocation prevents new calls and closes browser/proxy connections, but cannot roll back a completed side effect or guarantee interruption of an SMTP/CalDAV operation - already executing in an upstream library. API cancellation is observed between - agent steps, then the worker publishes the signed revocation at the terminal boundary. + already accepted by an upstream service. API cancellation polls durable task state, + cancels the in-flight coroutine, tears down sandbox processes/connections, and then + publishes signed revocation. Host-only development SMTP/CalDAV adapters use blocking + libraries; cancellation stops awaiting them but cannot forcibly stop their worker + thread, which is another reason production routes these capabilities through gateways. +- Before any tool side effect, Loop commits a durable operation id and action. The Step + and journal deletion commit atomically after execution. Recovery never blindly replays + a journal left in flight; it fails closed because the upstream outcome is unknowable. + Email Message-ID and calendar UID carry that operation id, but this is not an + exactly-once guarantee. ## Receipts and provenance @@ -121,5 +129,6 @@ not prompt instructions. a dead-letter stream. Task execution uses an atomic pending-to-running claim. - CI exercises enforcement against real Redis with AOF, including cross-process live tunnel revocation, fail-closed readiness during outage, and state recovery after - restart. The post-deploy smoke script separately verifies the cluster egress policy. + restart. Kubernetes acceptance kills and recovers Postgres before running another + verified task. The post-deploy smoke script separately verifies cluster egress policy. - Ingress must terminate TLS and add HSTS. NetworkPolicies default-deny ingress. diff --git a/docs/loop.md b/docs/loop.md index 14305db..ec40e70 100644 --- a/docs/loop.md +++ b/docs/loop.md @@ -22,6 +22,7 @@ discover project checks and record the pre-change baseline repeat: plan exactly one action enforce capability, budget, approval, egress, and workspace policy + persist an operation journal before crossing a side-effect boundary execute and persist one hash-chained step when finish is proposed: copy the final workspace independently for each check @@ -129,10 +130,23 @@ prevents duplicate deliveries from executing the same run concurrently. Terminal runs revoke their authority tokens and close live proxy/browser connections where the protocol supports it. +A planned tool call is journaled and committed before execution. The resulting Step +and journal deletion commit together. If a process disappears between those boundaries, +recovery reports the operation outcome as unknown and fails closed instead of replaying +a command, external write, memory append, or delegated task that may already have +completed. This is an explicit at-most-once policy. Email and calendar writes also carry +the stable operation id into Message-ID/UID, but Loop does not claim an upstream server +will deduplicate or roll back a request it already accepted. + +API cancellation is not a between-step flag: a database watcher cancels the active +coroutine. Async transports abort model/gateway requests, shell cancellation kills the +whole process group, sandbox cleanup removes the Docker container or Kubernetes Job, +and cancelling a parent marks all active descendants cancelled. + The enforcement acceptance harness abandons a claimed job, restarts Redis, and proves another worker reclaims and finishes it. Kubernetes acceptance additionally exercises -migration, per-command Jobs, Receipt authenticity, NetworkPolicy, a deliberately -broken API rollout, and rollback. +migration, per-command Jobs, Receipt authenticity, NetworkPolicy, Postgres interruption +and recovery, a deliberately broken API rollout, and rollback. ## MCP and provider surfaces @@ -173,8 +187,6 @@ untouched. Undo reverses only the exact applied patch and refuses overlapping ch ## Residual edges -- Cancellation is observed between steps; it does not interrupt an in-flight provider - request or shell command. - Protocol operations already accepted by an upstream service cannot be rolled back. - Browser sessions are pod-local; a gateway restart loses them. - Inline mode is not a security sandbox. diff --git a/scripts/k8s-deployment-acceptance.sh b/scripts/k8s-deployment-acceptance.sh index be2cb0e..f40ff7c 100755 --- a/scripts/k8s-deployment-acceptance.sh +++ b/scripts/k8s-deployment-acceptance.sh @@ -312,7 +312,7 @@ migration="$( kubectl exec deployment/postgres --namespace "$namespace" -- \ psql -U app -d app -tAc 'SELECT version_num FROM alembic_version' )" -if [[ "$migration" != "0009_contract_draft" ]]; then +if [[ "$migration" != "0010_operation_journal" ]]; then echo "Unexpected Alembic revision: $migration" >&2 exit 1 fi @@ -334,6 +334,31 @@ kubectl apply -f "$tmp/acceptance.yaml" bash "$root/scripts/k8s-enforcement-smoke.sh" "$namespace" run_cluster_probe before-rollback true +postgres_restarts_before="$( + kubectl get pod --namespace "$namespace" -l app.kubernetes.io/name=postgres \ + -o jsonpath='{.items[0].status.containerStatuses[0].restartCount}' +)" +kubectl exec deployment/postgres --namespace "$namespace" -- sh -c 'kill -9 1' || true +for _ in {1..60}; do + postgres_restarts_after="$( + kubectl get pod --namespace "$namespace" -l app.kubernetes.io/name=postgres \ + -o jsonpath='{.items[0].status.containerStatuses[0].restartCount}' 2>/dev/null || true + )" + if [[ "$postgres_restarts_after" =~ ^[0-9]+$ ]] && \ + (( postgres_restarts_after > postgres_restarts_before )); then + break + fi + sleep 1 +done +if [[ ! "$postgres_restarts_after" =~ ^[0-9]+$ ]] || \ + (( postgres_restarts_after <= postgres_restarts_before )); then + echo "Postgres fault injection did not restart the container" >&2 + exit 1 +fi +kubectl wait --namespace "$namespace" --for=condition=ready \ + pod -l app.kubernetes.io/name=postgres --timeout=3m +run_cluster_probe after-postgres-restart true + stable_image="$( kubectl get deployment api --namespace "$namespace" \ -o jsonpath='{.spec.template.spec.containers[?(@.name=="api")].image}' @@ -357,5 +382,5 @@ if [[ "$restored_image" != "$stable_image" ]]; then fi run_cluster_probe after-rollback false -printf '{"migration":"%s","rollback":true,"sandbox_digest":"%s","status":"passed"}\n' \ +printf '{"migration":"%s","postgres_recovery":true,"rollback":true,"sandbox_digest":"%s","status":"passed"}\n' \ "$migration" "$sandbox_digest" From 2b91b7562e243fcd389c6f59b096ee74dce6b01d Mon Sep 17 00:00:00 2001 From: Yichen Wu Date: Mon, 20 Jul 2026 23:14:25 -0700 Subject: [PATCH 2/2] Fix Postgres restart fault injection --- apps/api/tests/test_deployment_boundaries.py | 2 ++ scripts/k8s-deployment-acceptance.sh | 3 ++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/api/tests/test_deployment_boundaries.py b/apps/api/tests/test_deployment_boundaries.py index 1b96a3c..43e6258 100644 --- a/apps/api/tests/test_deployment_boundaries.py +++ b/apps/api/tests/test_deployment_boundaries.py @@ -307,6 +307,8 @@ def test_kubernetes_acceptance_migrates_runs_task_and_rolls_back() -> None: assert 'report.get("authentic")' in script assert "0010_operation_journal" in script assert "run_cluster_probe after-postgres-restart true" in script + assert "gosu postgres pg_ctl -D /var/lib/postgresql/data -m immediate stop" in script + assert "kill -9 1" not in script assert "api web worker" in smoke assert 'cluster="${LOOP_ACCEPTANCE_CLUSTER:-la-' in script assert 'registry_container="$registry"' in script diff --git a/scripts/k8s-deployment-acceptance.sh b/scripts/k8s-deployment-acceptance.sh index f40ff7c..fa9e910 100755 --- a/scripts/k8s-deployment-acceptance.sh +++ b/scripts/k8s-deployment-acceptance.sh @@ -338,7 +338,8 @@ postgres_restarts_before="$( kubectl get pod --namespace "$namespace" -l app.kubernetes.io/name=postgres \ -o jsonpath='{.items[0].status.containerStatuses[0].restartCount}' )" -kubectl exec deployment/postgres --namespace "$namespace" -- sh -c 'kill -9 1' || true +kubectl exec deployment/postgres --namespace "$namespace" -- \ + gosu postgres pg_ctl -D /var/lib/postgresql/data -m immediate stop || true for _ in {1..60}; do postgres_restarts_after="$( kubectl get pod --namespace "$namespace" -l app.kubernetes.io/name=postgres \