Harden cancellation, recovery, and concurrency

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.
 
 ---
 

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

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")

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.

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)
 

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)

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:

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,