2f09ad4a - Loosen error-fix ownership to device-level, add a one-shot decide dispatcher

DESIGN.md

@@ -419,6 +419,7 @@ agent watch grok-usage                         # one scan; knock child (under th
 agent watch assigned [--follow]                # allowlisted GitHub assignments → runner session + knock
 agent watch errors                             # one scan; $AGENT_HOME/error-fix.json; knock daemon polls with grok-usage
 agent watch error-fix                         # one scan; find-or-create implement task + isolated worktree; knock daemon polls with grok-usage
+agent watch error-decide                      # one scan; one-shot decide session per unconcluded error.seen; not wired into agent daemon in this revision
 agent supervise --session ID [--repo OWNER/REPO --number N] [--once|--follow]
 agent status
 agent dashboard [--port 7845]
@@ -577,7 +578,7 @@ Attach `error-fix` together with `spine`, `review-loop`, and `pr-review` on the
 |---|---|
 | Log credentials and adapter config | `$AGENT_HOME` on this device, not git, not the hub |
 | Watcher process | This device. Script, not the model. |
-| Analysis and “fix or skip” | The attached runner session on this device |
+| Analysis and “fix or skip” | Any session on this device (same `_origin_device_id`; not the scanning session only) |
 | Isolated worktree, checks, draft pull request | This device |
 | Merge | A human |
 | Hub | Replica + fan-out of the rows this device already wrote |
@@ -634,28 +635,36 @@ Log lines, stack traces, and error messages are untrusted data (§19.2). They ar
 }
 ```
 
-`repo` may be omitted when the adapter cannot map the stream; the session then `error.skip`s with reason `unmapped-repo`. `line_fingerprint` is optional: `sha256(server + newline + container + newline + exact line)` as 64 lowercase hex, computed from the raw line before redaction. Omit it when `server` or `container` is missing. Host adapters may print the hex on `error.fix` stdout; it is not a mandate and not a log-host name.
+`repo` may be omitted when the adapter cannot map the stream; a session on this device then `error.skip`s with reason `unmapped-repo`. `line_fingerprint` is optional: `sha256(server + newline + container + newline + exact line)` as 64 lowercase hex, computed from the raw line before redaction. Omit it when `server` or `container` is missing. Host adapters may print the hex on `error.fix` stdout; it is not a mandate and not a log-host name.
 
 ### 21.4 Analysis and eligibility
 
-After the knock, the session reads the row and writes `investigate.step` immediately (hypothesis, check, ruled out — each a new row). Then it inserts **one** typed conclusion. Both conclusion payloads include `error_id` (the `error.seen` id) and `fingerprint`:
+After the knock, a session on **this device** reads the row and writes `investigate.step` immediately (hypothesis, check, ruled out — each a new row). Then it inserts **one** typed conclusion. Ownership is device-level: any session with the same `_origin_device_id` as the `error.seen` row may conclude it; the scanning session named in `error-fix.json` need not be the writer. Both conclusion payloads include `error_id` (the `error.seen` id) and `fingerprint`:
 
 - `error.skip` — not a code fix (infra, noisy duplicate, unmapped repo, forbidden path, already an open draft for this fingerprint). Also `reason` (short token plus optional note).
-- `error.fix` — `execution_status=pending`. Local intent only.
+- `error.fix` — `execution_status=pending`. Local intent only. Also `brief` (short text: what's broken, likely cause, where to look — written from this session's own investigation, never a placeholder).
+
+```json
+{
+  "error_id": "…",
+  "fingerprint": "service|class|stack-sig|env",
+  "brief": "TimeoutError in the payout worker; the retry loop never backs off; add exponential backoff before retrying."
+}
+```
 
 The model does not certify eligibility by saying “this is safe”. The typed row is the decision. Confidence scores are not stored as proof.
 
-`agent activity add` now enforces the error-fix skill, payload, fingerprint, one-conclusion, unmapped-repo, and already-open-draft guards.
+`agent activity add` now enforces the error-fix skill, payload (including required `brief` on `error.fix` / `reason` on `error.skip`), fingerprint, one-conclusion, unmapped-repo, and already-open-draft guards.
 
-The adapter decides open vs closed by that `error_id` / `fingerprint`, plus the spine task whose `payload.error_id` matches. A later `error.skip` or a terminal task (`done` / `failed`) for the same `error_id` closes the incident. `pr.merged` knocks as today; it is not a second close signal. `agent task create` for a given `error_id` is find-or-create; a second `error.fix` does not open a second task.
+The adapter decides open vs closed by that `error_id` / `fingerprint`, plus any spine task on this device whose `payload.error_id` matches (not only a task under the scanning session). A later `error.skip` or a terminal task (`done` / `failed`) for the same `error_id` closes the incident. `pr.merged` knocks as today; it is not a second close signal. `agent task create` for a given `error_id` is find-or-create across this device; a second `error.fix` does not open a second task.
 
 Same fingerprint while the incident is **open**: enrich `error.seen`. Do not create a second task or a second pull request. After close: the next match is a **new** `error.seen` (new id, first insert knocks).
 
 ### 21.5 Patch and draft pull request
 
 On `error.fix`:
 
-1. `agent task create --workflow implement --error-id <error.seen-id>` on this session (find-or-create). That copies `error_id` and `repo` from the `error.seen` row into the task payload.
+1. `agent task create --workflow implement --error-id <error.seen-id>` on this device (find-or-create; any session on the same `_origin_device_id`). That copies `error_id` and `repo` from the `error.seen` row into the task payload.
 2. Isolated worktree of that task `payload.repo` at the allowed base revision. Git operations are scripts. `payload.repo` is already on the task because analysis refused `error.fix` when `repo` was missing. Never fall back to the origin checkout.
 3. Spine implement: mandatory checks must `pass`, then `pr.open` opens a **draft** (spine `pushed`). Title/body may be model-drafted; the GitHub API call is a script. A retry finds an existing draft for this fingerprint instead of opening a second one.
 4. pr-review gates run on that head after `pushed`.
@@ -669,6 +678,18 @@ The model never receives production credentials. Analysis that only reads the ex
 
 - A second hub state machine, leases, or autonomous merge
 
+### 21.7 One-shot decide dispatcher
+
+`agent watch error-decide` drains unconcluded `error.seen` rows on this device one at a time (oldest `payload.first_seen` first; `id` as tiebreaker). For each row it:
+
+1. Ensures a deterministic runner session `error-decide-<error_id>` with skills `error-fix`, `spine`, `review-loop`, and `pr-review` (`status: active`).
+2. Starts that session’s tmux pane (grok). Waits for the session to reach an idle prompt (bounded by a timeout), then knocks `da ist Post id <uuid>` directly into it (not via `knock.deliver`, which would target the scanning session on the `error.seen` row) and retries the Enter keypress (up to a bounded number of attempts) until the session shows as busy, confirming the knock was actually accepted; if either wait is never satisfied, that row is recorded as failed and the dispatcher moves on to the next one.
+3. Polls until this device writes `error.fix` or `error.skip` for that `error_id`, or until timeout (default 30 minutes).
+4. Stops the pane (`runtime.control: stopped`) and leaves the session `active` — it does not `session close`, so `agent watch error-fix` can still create the implement task under that session without racing the “open tasks” guard.
+5. Moves to the next unconcluded row. No auto-continue / keep-working wiring; each decide session handles one error. A row whose session setup or knock fails is recorded as an error line and does not block the rest of the backlog scan.
+
+Empty backlog prints `error.seen decide none`. The exclusive lock `error-decide-act:<device>` is held for the whole scan — the backlog read and the per-row start/knock/wait/stop sequence — so two overlapping invocations of this scan cannot both dispatch the same row. That key never collided with `error-fix-act:<device>`'s own lock in the first place (they are different `pg_advisory_lock(hashtext(...))` keys), so widening the scope does not block conclusion writes; the reason to hold it for the whole scan is purely to serialize overlapping dispatcher invocations against each other. Not wired into `agent daemon` in this revision.
+
 ## 22. Static supervise loop (v1)
 
 A second model must not orchestrate the first. `agent supervise` is a **script** with locked questions and locked answers. Model text is not a state transition.

README.md

@@ -96,6 +96,7 @@ agent watch grok-usage  # one scan of SuperGrok weekly credits into usage.snapsh
 agent watch assigned [--follow]  # allowlisted assignments; needs `gh` and `$AGENT_HOME/watch.json`
 agent watch errors      # one scan; $AGENT_HOME/error-fix.json; no log host in this package
 agent watch error-fix   # one scan; find-or-create implement task + isolated worktree
+agent watch error-decide  # one scan; one-shot decide session per unconcluded error.seen; not polled by the daemon yet
 agent supervise --session ID [--repo OWNER/REPO --number N] [--once|--follow]
 # agent knock (daemon, no --once) polls grok-usage, pending, pr.merged, github pending, mail pending, errors, and error-fix every 60s
 ```

src/agent_cli/error_decide_act.py

@@ -0,0 +1,171 @@
+"""Dispatch one fresh, one-shot session per unconcluded error.seen row."""
+
+from __future__ import annotations
+
+import socket
+import time
+from collections.abc import Callable
+from typing import Any
+
+from .store import Store, StoreError, utcnow
+
+DEFAULT_TIMEOUT_S = 1800.0
+DEFAULT_POLL_INTERVAL_S = 5.0
+
+
+def decide_session_id(error_id: str) -> str:
+    """Deterministic per-error session id, so a retry after a timeout reuses the
+    same session row instead of piling up a new one per attempt."""
+    return f"error-decide-{error_id}"
+
+
+def _has_conclusion(store: Store, error_id: str) -> bool:
+    origin = store.device_id()
+    for row in store.rows("activity"):
+        if row.get("_origin_device_id") != origin:
+            continue
+        if row.get("type") not in ("error.fix", "error.skip"):
+            continue
+        inner = row.get("payload")
+        if isinstance(inner, dict) and inner.get("error_id") == error_id:
+            return True
+    return False
+
+
+def unconcluded_seen_rows(store: Store) -> list[dict[str, Any]]:
+    origin = store.device_id()
+    rows: list[dict[str, Any]] = []
+    for row in store.rows("activity"):
+        if row.get("_origin_device_id") != origin:
+            continue
+        if row.get("type") != "error.seen":
+            continue
+        rid = row.get("id")
+        if not isinstance(rid, str) or rid == "":
+            continue
+        if _has_conclusion(store, rid):
+            continue
+        rows.append(row)
+
+    def sort_key(row: dict[str, Any]) -> tuple[str, str]:
+        inner = row.get("payload")
+        payload = inner if isinstance(inner, dict) else {}
+        first = payload.get("first_seen")
+        first_s = first if isinstance(first, str) else ""
+        return (first_s, str(row.get("id") or ""))
+
+    rows.sort(key=sort_key)
+    return rows
+
+
+def _ensure_decide_session(store: Store, sid: str, now: str) -> None:
+    existing = store.row("session", sid)
+    if existing is None:
+        store.write(
+            "session",
+            "insert",
+            sid,
+            {
+                "id": sid,
+                "kind": "runner",
+                "started_at": now,
+                "last_seen_at": now,
+                "host": socket.gethostname(),
+                "status": "active",
+                "skills": ["error-fix", "spine", "review-loop", "pr-review"],
+            },
+        )
+        return
+    if existing.get("_origin_device_id") != store.device_id():
+        raise StoreError(f"session {sid} is owned by another device")
+    if existing.get("kind") != "runner":
+        raise StoreError(f"session {sid} is kind={existing.get('kind')}, error-decide worker must be runner")
+    required = ["error-fix", "spine", "review-loop", "pr-review"]
+    current_skills = existing.get("skills")
+    current = list(current_skills) if isinstance(current_skills, list) else []
+    missing = [s for s in required if s not in current]
+    was_closed = existing.get("status") == "closed"
+    if missing or was_closed:
+        updated = dict(existing)
+        if missing:
+            updated["skills"] = current + missing
+        if was_closed:
+            updated["status"] = "active"
+            updated["last_seen_at"] = now
+        store.write("session", "update", sid, {k: v for k, v in updated.items() if not k.startswith("_")})
+
+
+def _wait_for_conclusion(
+    store: Store,
+    error_id: str,
+    *,
+    timeout_s: float,
+    poll_interval_s: float,
+    sleep: Callable[[float], None],
+) -> bool:
+    deadline = time.monotonic() + timeout_s
+    while True:
+        if _has_conclusion(store, error_id):
+            return True
+        if time.monotonic() >= deadline:
+            return False
+        sleep(poll_interval_s)
+
+
+def scan_error_decide(
+    store: Store,
+    *,
+    start: Callable[[str], None],
+    stop: Callable[[str], None],
+    knock: Callable[[str, str], None],
+    sleep: Callable[[float], None] = time.sleep,
+    timeout_s: float = DEFAULT_TIMEOUT_S,
+    poll_interval_s: float = DEFAULT_POLL_INTERVAL_S,
+) -> list[str]:
+    # Held for the whole scan so two overlapping invocations don't double-dispatch
+    # the same row. This key never collides with error-fix-act's own lock (different
+    # pg_advisory_lock hashtext keys), so it doesn't block conclusion writes from the
+    # sessions this dispatcher starts.
+    with store.exclusive("error-decide-act:" + store.device_id()):
+        backlog = unconcluded_seen_rows(store)
+        lines: list[str] = []
+        for row in backlog:
+            error_id = str(row["id"])
+            if _has_conclusion(store, error_id):
+                continue
+            sid = decide_session_id(error_id)
+            now = utcnow()
+            try:
+                _ensure_decide_session(store, sid, now)
+            except (StoreError, SystemExit, OSError) as exc:
+                lines.append(f"error.seen {error_id} error session={sid}: {exc}")
+                continue
+            failure: StoreError | SystemExit | OSError | None = None
+            stop_failure: StoreError | SystemExit | OSError | None = None
+            decided = False
+            try:
+                start(sid)
+                knock(sid, error_id)
+                decided = _wait_for_conclusion(
+                    store,
+                    error_id,
+                    timeout_s=timeout_s,
+                    poll_interval_s=poll_interval_s,
+                    sleep=sleep,
+                )
+            except (StoreError, SystemExit, OSError) as exc:
+                failure = exc
+            finally:
+                try:
+                    stop(sid)
+                except (StoreError, SystemExit, OSError) as stop_exc:
+                    stop_failure = stop_exc
+            if failure is not None:
+                lines.append(f"error.seen {error_id} error session={sid}: {failure}")
+                continue
+            note = f" (stop failed: {stop_failure})" if stop_failure is not None else ""
+            if decided:
+                lines.append(f"error.seen {error_id} decided session={sid}{note}")
+            else:
+                lines.append(f"error.seen {error_id} timeout session={sid}{note}")
+        return lines