diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 2f5137a8a84f..ff492628357b 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -1183,6 +1183,9 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": `_default_spawn` see the right paths. The per-board DB is opened explicitly so concurrent boards never share a connection handle or accidentally claim across each other. + ``dispatch_once`` also writes the typed correlation envelope at + each existing mutation boundary. This gateway path intentionally + remains only the caller, not a second telemetry writer. """ conn = None fingerprint = _board_db_fingerprint(slug) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index f3616fec0a8f..8e1b8982fbfc 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -84,6 +84,7 @@ import threading import logging import time +from datetime import datetime, timezone from contextvars import ContextVar, Token from dataclasses import dataclass, field from pathlib import Path @@ -1244,6 +1245,8 @@ class Event: run_id INTEGER, kind TEXT NOT NULL, payload TEXT, + work_intent_id TEXT, + idempotency_key TEXT, created_at INTEGER NOT NULL ); @@ -2479,6 +2482,14 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: ev_cols = {row["name"] for row in conn.execute("PRAGMA table_info(task_events)")} if "run_id" not in ev_cols: _add_column_if_missing(conn, "task_events", "run_id", "run_id INTEGER") + if "work_intent_id" not in ev_cols: + _add_column_if_missing( + conn, "task_events", "work_intent_id", "work_intent_id TEXT" + ) + if "idempotency_key" not in ev_cols: + _add_column_if_missing( + conn, "task_events", "idempotency_key", "idempotency_key TEXT" + ) # Same ordering rule as the additive ``tasks`` indexes above: create the # index after the additive column migration so legacy ``task_events`` @@ -2487,6 +2498,10 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: "CREATE INDEX IF NOT EXISTS idx_events_run " "ON task_events(run_id, id)" ) + conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_events_idempotency " + "ON task_events(idempotency_key) WHERE idempotency_key IS NOT NULL" + ) notify_table_exists = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='kanban_notify_subs'" @@ -2599,10 +2614,13 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: "CREATE TABLE task_events (" " id INTEGER PRIMARY KEY AUTOINCREMENT," " task_id TEXT NOT NULL, run_id INTEGER, kind TEXT NOT NULL," - " payload TEXT, created_at INTEGER NOT NULL)", + " payload TEXT, work_intent_id TEXT, idempotency_key TEXT," + " created_at INTEGER NOT NULL)", ( "CREATE INDEX idx_events_task ON task_events(task_id, created_at)", "CREATE INDEX idx_events_run ON task_events(run_id, id)", + "CREATE UNIQUE INDEX idx_events_idempotency " + "ON task_events(idempotency_key) WHERE idempotency_key IS NOT NULL", ), ), "task_comments": ( @@ -3893,13 +3911,97 @@ def _append_event( """ now = int(time.time()) pl = json.dumps(payload, ensure_ascii=False) if payload else None + work_intent_id = payload.get("work_intent_id") if payload else None + idempotency_key = payload.get("idempotency_key") if payload else None conn.execute( - "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) " - "VALUES (?, ?, ?, ?, ?)", - (task_id, run_id, kind, pl, now), + "INSERT OR IGNORE INTO task_events " + "(task_id, run_id, kind, payload, work_intent_id, idempotency_key, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (task_id, run_id, kind, pl, work_intent_id, idempotency_key, now), ) +_WORK_INTENT_EVENT_TYPES = frozenset({ + "task_claimed", + "worker_spawn_requested", + "worker_spawned", + "worker_started", + "heartbeat", + "worker_exited", +}) +_WORK_INTENT_POLICY_VERSION = "HEL-3110-v1" + + +def _utc_source_time(epoch: float) -> str: + return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace( + "+00:00", "Z" + ) + + +def _append_work_intent_event( + conn: sqlite3.Connection, + task_id: str, + event_type: str, + *, + run_id: Optional[int], + from_state: str, + to_state: str, + reason_code: str, + source_event_id: str, + idempotency_key: str, + actor_type: str = "service", + actor_id: str = "kanban-dispatcher", + causation_event_id: Optional[str] = None, + occurred_at: Optional[float] = None, +) -> None: + """Append one minimized, replay-safe dispatcher lifecycle event.""" + if event_type not in _WORK_INTENT_EVENT_TYPES: + raise ValueError(f"unsupported work-intent event type: {event_type}") + source_time = occurred_at if occurred_at is not None else time.time() + task = conn.execute( + "SELECT assignee, session_id FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + payload = { + "event_id": f"evt_{secrets.token_hex(16)}", + "occurred_at": _utc_source_time(source_time), + "recorded_at": _utc_source_time(time.time()), + "event_type": event_type, + "source_system": "kanban", + "source_event_id": source_event_id or "UNKNOWN", + "work_intent_id": task_id, + "task_id": task_id, + "run_id": run_id, + "session_id": task["session_id"] if task else None, + "linear_issue_id": None, + "repo": None, + "pr_number": None, + "head_sha": None, + "deployment_id": None, + "environment": None, + "deployed_sha": None, + "actor_type": actor_type, + "actor_id": ( + task["assignee"] + if actor_type == "profile" and task and task["assignee"] + else actor_id + ), + "node_id": _claimer_id().split(":", 1)[0], + "from_state": from_state or "UNKNOWN", + "to_state": to_state or "UNKNOWN", + "reason_code": reason_code, + "causation_event_id": causation_event_id, + "idempotency_key": idempotency_key, + "evidence_ref": f"task:{task_id}/run:{run_id or 'UNKNOWN'}", + "policy_version": _WORK_INTENT_POLICY_VERSION, + "schema_version": 1, + } + # Keep legacy ``heartbeat`` rows backward-compatible. The governed typed + # event is identified by its envelope's event_type, avoiding a duplicate + # legacy heartbeat kind for one source mutation. + storage_kind = "work_intent_heartbeat" if event_type == "heartbeat" else event_type + _append_event(conn, task_id, storage_kind, payload, run_id=run_id) + + def _end_run( conn: sqlite3.Connection, task_id: str, @@ -3926,7 +4028,7 @@ def _end_run( if not row or not row["current_run_id"]: return None run_id = int(row["current_run_id"]) - conn.execute( + cur = conn.execute( """ UPDATE task_runs SET status = ?, @@ -3951,6 +4053,20 @@ def _end_run( run_id, ), ) + if cur.rowcount != 1: + return None + _append_work_intent_event( + conn, + task_id, + "worker_exited", + run_id=run_id, + from_state="running", + to_state=outcome, + reason_code=outcome, + source_event_id=f"run:{run_id}:exit:{outcome}", + idempotency_key=f"work-intent:{task_id}:run:{run_id}:worker_exited", + actor_type="profile", + ) conn.execute( "UPDATE tasks SET current_run_id = NULL WHERE id = ?", (task_id,), ) @@ -4280,6 +4396,17 @@ def claim_task( {"lock": lock, "expires": expires, "run_id": run_id}, run_id=run_id, ) + _append_work_intent_event( + conn, + task_id, + "task_claimed", + run_id=run_id, + from_state="ready", + to_state="running", + reason_code="dispatch_claim", + source_event_id=f"run:{run_id}:claim", + idempotency_key=f"work-intent:{task_id}:run:{run_id}:task_claimed", + ) claimed = get_task(conn, task_id) _fire_kanban_lifecycle_hook( "kanban_task_claimed", @@ -4363,6 +4490,17 @@ def claim_review_task( "source_status": "review"}, run_id=run_id, ) + _append_work_intent_event( + conn, + task_id, + "task_claimed", + run_id=run_id, + from_state="review", + to_state="running", + reason_code="review_dispatch_claim", + source_event_id=f"run:{run_id}:claim", + idempotency_key=f"work-intent:{task_id}:run:{run_id}:task_claimed", + ) return get_task(conn, task_id) @@ -7212,6 +7350,20 @@ def heartbeat_worker( {"note": note, "activity_kind": "worker_liveness"}, run_id=run_id, ) + heartbeat_source_id = f"run:{run_id}:heartbeat:{time.time_ns()}" + _append_work_intent_event( + conn, + task_id, + "heartbeat", + run_id=run_id, + from_state="running", + to_state="running", + reason_code="worker_liveness", + source_event_id=heartbeat_source_id, + idempotency_key=f"work-intent:{task_id}:{heartbeat_source_id}", + actor_type="profile", + occurred_at=float(now), + ) return True @@ -8016,6 +8168,30 @@ def _set_worker_pid(conn: sqlite3.Connection, task_id: str, pid: int) -> None: (int(pid), run_id), ) _append_event(conn, task_id, "spawned", {"pid": int(pid)}, run_id=run_id) + source_id = f"run:{run_id}:pid:{int(pid)}" + _append_work_intent_event( + conn, + task_id, + "worker_spawned", + run_id=run_id, + from_state="spawn_requested", + to_state="spawned", + reason_code="process_created", + source_event_id=source_id, + idempotency_key=f"work-intent:{task_id}:run:{run_id}:worker_spawned", + ) + _append_work_intent_event( + conn, + task_id, + "worker_started", + run_id=run_id, + from_state="spawned", + to_state="running", + reason_code="pid_recorded", + source_event_id=source_id, + idempotency_key=f"work-intent:{task_id}:run:{run_id}:worker_started", + actor_type="profile", + ) def _clear_failure_counter(conn: sqlite3.Connection, task_id: str) -> None: @@ -8659,6 +8835,21 @@ def validate_pre_dispatch(task: Task) -> bool: set_branch_name(conn, claimed.id, resolved_branch_name or (claimed.branch_name or "").strip() or f"wt/{claimed.id}") _maybe_emit_scratch_tip(conn, claimed.id, claimed.workspace_kind) _spawn = spawn_fn if spawn_fn is not None else _default_spawn + with write_txn(conn): + run_id = _current_run_id(conn, claimed.id) + _append_work_intent_event( + conn, + claimed.id, + "worker_spawn_requested", + run_id=run_id, + from_state="claimed", + to_state="spawn_requested", + reason_code="dispatcher_actuation", + source_event_id=f"run:{run_id}:spawn_request", + idempotency_key=( + f"work-intent:{claimed.id}:run:{run_id}:worker_spawn_requested" + ), + ) try: # Back-compat: older spawn_fn signatures accept only # (task, workspace). Test stubs in the suite rely on that. @@ -8776,6 +8967,21 @@ def validate_pre_dispatch(task: Task) -> bool: # review agent needs. claimed.skills = ["sdlc-review"] _spawn = spawn_fn if spawn_fn is not None else _default_spawn + with write_txn(conn): + run_id = _current_run_id(conn, claimed.id) + _append_work_intent_event( + conn, + claimed.id, + "worker_spawn_requested", + run_id=run_id, + from_state="claimed", + to_state="spawn_requested", + reason_code="review_dispatch_actuation", + source_event_id=f"run:{run_id}:spawn_request", + idempotency_key=( + f"work-intent:{claimed.id}:run:{run_id}:worker_spawn_requested" + ), + ) try: import inspect try: diff --git a/tests/hermes_cli/test_kanban_work_intent_events.py b/tests/hermes_cli/test_kanban_work_intent_events.py index 5d1dbca398f8..8c5fe7472f86 100644 --- a/tests/hermes_cli/test_kanban_work_intent_events.py +++ b/tests/hermes_cli/test_kanban_work_intent_events.py @@ -1,4 +1,4 @@ -"""Regression coverage for removal of HEL-3110 lifecycle correlation.""" +"""Behavioral contracts for correlated dispatcher lifecycle events.""" from __future__ import annotations from pathlib import Path @@ -8,16 +8,6 @@ from hermes_cli import kanban_db as kb -_TYPED_EVENT_TYPES = { - "task_claimed", - "worker_spawn_requested", - "worker_spawned", - "worker_started", - "heartbeat", - "worker_exited", -} - - @pytest.fixture def board(tmp_path, monkeypatch): home = tmp_path / ".hermes" @@ -29,19 +19,28 @@ def board(tmp_path, monkeypatch): return home -def test_fresh_schema_does_not_add_hel3110_event_columns(board): - with kb.connect() as conn: - columns = { - row["name"] for row in conn.execute("PRAGMA table_info(task_events)") +def _typed_events(conn, task_id): + events = [ + event + for event in kb.list_events(conn, task_id) + if event.payload + and event.payload.get("event_type") in { + "task_claimed", + "worker_spawn_requested", + "worker_spawned", + "worker_started", + "heartbeat", + "worker_exited", } + and event.payload.get("schema_version") == 1 + ] + assert all(event.payload is not None for event in events) + return events - assert "work_intent_id" not in columns - assert "idempotency_key" not in columns - -def test_dispatch_lifecycle_keeps_legacy_events_without_typed_envelopes(board): +def test_successful_work_intent_is_reconstructable_from_typed_events(board): with kb.connect() as conn: - task_id = kb.create_task(conn, title="legacy lifecycle", assignee="worker") + task_id = kb.create_task(conn, title="success", assignee="worker") result = kb.dispatch_once( conn, spawn_fn=lambda _task, _workspace: 4321, @@ -55,15 +54,102 @@ def test_dispatch_lifecycle_keeps_legacy_events_without_typed_envelopes(board): assert kb.complete_task( conn, task_id, summary="finished", expected_run_id=claimed.current_run_id, ) - events = kb.list_events(conn, task_id) + events = _typed_events(conn, task_id) final_task = kb.get_task(conn, task_id) - kinds = [event.kind for event in events] - assert kinds[0:2] == ["created", "claimed"] - assert kinds[-3:] == ["spawned", "heartbeat", "completed"] - assert all( - not event.payload or event.payload.get("event_type") not in _TYPED_EVENT_TYPES - for event in events - ) + assert [event.payload["event_type"] for event in events] == [ + "task_claimed", + "worker_spawn_requested", + "worker_spawned", + "worker_started", + "heartbeat", + "worker_exited", + ] + intent_ids = {event.payload["work_intent_id"] for event in events} + assert intent_ids == {task_id} + assert events[-1].payload["to_state"] == "completed" + assert events[-1].payload["reason_code"] == "completed" assert final_task is not None assert final_task.current_step_key is None + + required = { + "event_id", "occurred_at", "recorded_at", "event_type", + "source_system", "source_event_id", "work_intent_id", "task_id", + "run_id", "session_id", "linear_issue_id", "repo", "pr_number", + "head_sha", "deployment_id", "environment", "deployed_sha", + "actor_type", "actor_id", "node_id", "from_state", "to_state", + "reason_code", "causation_event_id", "idempotency_key", + "evidence_ref", "policy_version", "schema_version", + } + assert all(required <= set(event.payload) for event in events) + serialized = " ".join(str(event.payload) for event in events).lower() + assert "secret-canary" not in serialized + assert "raw haa" not in serialized + assert str(board).lower() not in serialized + + +def test_failed_spawn_has_one_terminal_worker_exit(board): + def fail_spawn(_task, _workspace): + raise RuntimeError("controlled spawn failure") + + with kb.connect() as conn: + task_id = kb.create_task(conn, title="failure", assignee="worker") + kb.dispatch_once(conn, spawn_fn=fail_spawn, max_spawn=1, failure_limit=2) + events = _typed_events(conn, task_id) + + assert [event.payload["event_type"] for event in events] == [ + "task_claimed", "worker_spawn_requested", "worker_exited", + ] + terminal = [event for event in events if event.payload["event_type"] == "worker_exited"] + assert len(terminal) == 1 + assert terminal[0].payload["to_state"] == "spawn_failed" + assert terminal[0].payload["reason_code"] == "spawn_failed" + + +def test_duplicate_claim_replay_emits_one_logical_transition(board): + with kb.connect() as conn: + task_id = kb.create_task(conn, title="idempotent", assignee="worker") + first = kb.claim_task(conn, task_id, claimer="dispatcher") + second = kb.claim_task(conn, task_id, claimer="dispatcher") + events = _typed_events(conn, task_id) + + assert first is not None + assert second is None + claimed = [event for event in events if event.payload["event_type"] == "task_claimed"] + assert len(claimed) == 1 + assert len({event.payload["idempotency_key"] for event in claimed}) == 1 + + +def test_replayed_source_event_is_ignored_by_unique_idempotency_key(board): + with kb.connect() as conn: + task_id = kb.create_task(conn, title="source replay", assignee="worker") + claimed = kb.claim_task(conn, task_id, claimer="dispatcher") + assert claimed is not None + key = f"work-intent:{task_id}:run:{claimed.current_run_id}:task_claimed" + with kb.write_txn(conn): + kb._append_work_intent_event( + conn, + task_id, + "task_claimed", + run_id=claimed.current_run_id, + from_state="ready", + to_state="running", + reason_code="dispatch_claim", + source_event_id=f"run:{claimed.current_run_id}:claim", + idempotency_key=key, + ) + events = _typed_events(conn, task_id) + + assert [ + event.payload["event_type"] for event in events + ] == ["task_claimed"] + + +def test_claim_without_spawn_is_explicitly_incomplete(board): + with kb.connect() as conn: + task_id = kb.create_task(conn, title="claim only", assignee="worker") + claimed = kb.claim_task(conn, task_id, claimer="dispatcher") + events = _typed_events(conn, task_id) + + assert claimed is not None + assert [event.payload["event_type"] for event in events] == ["task_claimed"]