Handle task.failed events to prevent DAG deadlocks

[mvillmow]

Keystone only reacts to 'completed' events. If a task fails, all downstream tasks remain permanently 'pending' — the DAG is deadlocked with no recovery path. Hermes publishes hi.tasks.{team}.{task}.failed events, and Telemachy treats failed/error/cancelled as terminal. Keystone has no equivalent concept.

Impact: A single failed task silently stalls the entire downstream DAG forever.

[mvillmow]

Implementation Plan

The src/keystone/ directory only has __pycache__ — no Python source files! The implementation lives only in the worktree. Now I have all the context I need to write the implementation plan.

---

Implementation Plan: Issue #87 — Handle task.failed Events to Prevent DAG Deadlocks

1. Objective

Add terminal-state handling for task.failed (and related failure statuses) to the Keystone Python daemon so that when a task fails, downstream dependents are unblocked rather than permanently stalled. Specifically:

• Subscribe to hi.tasks.{team}.{task}.failed events on the existing homeric-tasks JetStream stream
• Treat FAILED, ERROR, and CANCELLED as terminal states equivalent to COMPLETED for DAG-advancement purposes
• Cascade: when a task enters a terminal state, mark its downstream pending dependents as CANCELLED and call advance_dag() so they don't silently stall
• Ensure every NATS code path ACKs its message to prevent redelivery loops

---

2. Approach

Key decisions (informed by Prior Learnings):

• Re-use the existing homeric-tasks / hi.tasks.> subscription — do not create a new stream or consumer. The stream already delivers both .completed and .failed subjects; add a dispatch branch in the listener.
• Treat all terminal states uniformly for DAG advancement — FAILED, ERROR, CANCELLED are all terminal; downstream tasks whose upstream is in any terminal state become "ready" (so the DAG engine can cancel them).
• Cascade on receipt — when advance_dag() is called after a failure event, the DAGWalker identifies PENDING tasks whose all dependencies are terminal, then marks them CANCELLED via update_task before assigning anything. This prevents them from being dispatched to agents in an invalid state.
• Guard DAG state with mutex — if multi-threaded access is introduced, std::mutex wraps the node map (per Prior Learnings TSan guidance). In Python's asyncio context, the event loop serializes coroutines naturally, but any executor-thread usage must use asyncio.Lock.
• ACK on every exit path — wrap _process_message in a try/except that ACKs even on unhandled exceptions (prevents infinite redelivery of bad messages, per issue Add NATS JetStream message acknowledgment in NATSListener #86 pattern).
• Test without a live NATS server — use a FakeMsg dataclass in tests; no mocking of the NATS library required.

---

3. Files to Create

File	Description
src/keystone/__init__.py	Package marker (empty or minimal)
src/keystone/__main__.py	python -m keystone entry point
src/keystone/models.py	TaskStatus enum + TERMINAL_STATUSES frozenset + Task/Agent dataclasses
src/keystone/config.py	Lazy settings singleton (get_settings())
src/keystone/logging.py	Structured JSON logger (get_logger())
src/keystone/nats_listener.py	NATSListener — subscribes to hi.tasks.>, dispatches by verb/status
src/keystone/dag_walker.py	DAGWalker — iterative cycle detection, get_ready_tasks(), advance_dag()
src/keystone/maestro_client.py	MaestroClient — HTTP client for ai-maestro API with retries
src/keystone/daemon.py	Main daemon: NATS connect, startup scan, background scan, advance_dag() orchestrator
src/keystone/health.py	HTTP health-check endpoint
tests/__init__.py	Test package marker
tests/conftest.py	Shared pytest fixtures
tests/helpers.py	make_task() / make_agent() factory helpers
tests/test_nats_listener.py	Tests: ACK paths, task.failed handling, nested payload parsing, path-traversal validation
tests/test_dag_walker.py	Tests: cycle detection (iterative, deep chain), terminal-state readiness, agent availability, DAG advancement
tests/test_daemon.py	Tests: advance_dag(), _startup_scan(), _background_scan_task(), NATS callbacks
tests/test_config.py	Settings singleton tests
tests/test_health.py	Health endpoint tests
tests/test_maestro_client.py	API client parsing/retry tests

---

4. Files to Modify

File	Change
(none currently exist in main src/keystone/)	All are new — the Python daemon layer does not yet exist in the main branch

If a setup.py / pyproject.toml exists at project root, add keystone to packages.

---

5. Implementation Order

1. src/keystone/models.py — Define TaskStatus enum with FAILED, ERROR, CANCELLED, COMPLETED, PENDING, IN_PROGRESS; define TERMINAL_STATUSES; define Task and Agent dataclasses. This is the foundation everything else depends on.

2. src/keystone/logging.py — Structured JSON logger. Low complexity; needed early by all other modules.

3. src/keystone/config.py — Lazy settings singleton: NATS URL (nats://localhost:4222), maestro base URL, NATS subject (hi.tasks.>), background scan interval. Read from env vars with defaults.

4. tests/helpers.py — make_task(task_id, team_id, status, dependencies) and make_agent(agent_id, status, session_status, current_task_id) factory functions. Needed by all test files.

5. src/keystone/dag_walker.py — DAGWalker class:

   • validate_no_cycles() — iterative DFS (avoid RecursionError on 2000-deep chains, per issue validate_no_cycles uses unbounded recursion — risks RecursionError on large DAGs #100)
   • get_ready_tasks() — returns PENDING tasks whose all deps are in TERMINAL_STATUSES; importantly, tasks downstream of a FAILED dep are ready (not blocked) so the engine can cancel them
   • get_available_agents() — filters by status == "active", session_status == "online", current_task_id is None
   • advance_dag() — matches ready tasks to available agents, returns list[tuple[Task, Agent]]

6. tests/test_dag_walker.py — Write before finalizing dag_walker.py (TDD). Key cases:

   • Linear chain, diamond DAG, cycle detection, self-cycle, 2000-deep chain (no RecursionError)
   • FAILED, ERROR, CANCELLED deps → downstream task becomes ready (issue Handle task.failed events to prevent DAG deadlocks #87 core case)
   • IN_PROGRESS task is not ready
   • Offline/inactive/busy agent not available
   • Full advance_dag() matching

7. src/keystone/nats_listener.py — NATSListener:

   • _validate_id(value, label) — rejects path-traversal chars; only [a-zA-Z0-9_-] allowed (issue Security: NATS subject tokens from events used in API URLs without validation #113)
   • _extract_status(payload) — extracts from payload["data"]["status"] (Hermes format, issue NATS payload schema mismatch: Keystone reads top-level status but Hermes nests it inside data #107), falls back to top-level status/newStatus
   • on_task_event(msg) — outer try/except that always ACKs (issue Add NATS JetStream message acknowledgment in NATSListener #86)
   • _process_message(msg) — parse subject into (team_id, task_id, verb); dispatch: if verb in ("completed", "failed") → call advance_dag; if verb == "updated" and status in TERMINAL_STATUSES → call advance_dag; otherwise ignore
   • Do NOT catch Exception before the specific terminal-state dispatch (per Prior Learnings failure)

8. tests/test_nats_listener.py — Key cases:

   • ACK on success, malformed subject, invalid JSON, unknown verb, callback exception (issue Add NATS JetStream message acknowledgment in NATSListener #86)
   • hi.tasks.team1.task1.failed → advance_dag called (issue Handle task.failed events to prevent DAG deadlocks #87 primary test)
   • hi.tasks.team1.task1.completed → advance_dag called
   • updated with data.status = "failed" → advance_dag called
   • updated with data.status = "in_progress" → advance_dag NOT called
   • Path-traversal in team_id → ACKed but not processed

9. src/keystone/maestro_client.py — MaestroClient: get_teams(), get_tasks(team_id), get_agents(), update_task(team_id, task_id, patch), retry on transient HTTP errors, close().

10. src/keystone/daemon.py — Main daemon:

    • advance_dag(team_id, *, client=None) — create client if not provided; call DAGWalker.advance_dag(); call update_task for each assignment; always close client if created internally; log and swallow exceptions
    • _startup_scan(*, client=None) — iterate all teams, call advance_dag for each
    • _background_scan_task() — periodic loop; asyncio.wait_for(_startup_scan(), ...) on interval; exit when _shutdown event set; survive scan exceptions
    • NATS callbacks: _nats_error_cb, _nats_disconnected_cb, _nats_reconnected_cb, _nats_closed_cb
    • main() — connect NATS, subscribe to hi.tasks.> with durable consumer keystone-daemon, run startup scan, start background scan, await shutdown

11. tests/test_daemon.py — Key cases (already exist in worktree, port them):

    • advance_dag assigns tasks, closes client only when internally created
    • advance_dag handles get_tasks exceptions without raising
    • _startup_scan scans all teams, skips teams without id/_id
    • _background_scan_task runs loop, exits on shutdown, survives exceptions
    • NATS callbacks log/mark correct state

12. src/keystone/health.py and src/keystone/__init__.py / src/keystone/__main__.py** — health endpoint and entry points.

13. Wire up setup.py/pyproject.toml to include keystone package, add pytest-asyncio to test deps.

---

6. Verification

Unit tests (no live NATS/maestro required):

# From project root
python -m pytest tests/ -v

Key test assertions:

• test_failed_verb_advances_dag — hi.tasks.team1.task1.failed triggers advance_dag(team_id="team1", task_id="task1")
• test_task_with_failed_dep_is_ready — DAGWalker returns downstream task as ready when upstream is FAILED
• test_ack_on_callback_exception — ACK called even when callback raises
• test_cycle_detected / test_deep_chain_no_recursion_error — iterative DFS correctness

Integration smoke test (with live NATS):

# Start NATS with JetStream
nats-server -js &
# Publish a fake task.failed event
nats pub hi.tasks.team1.task1.failed '{"data":{"status":"failed"}}'
# Observe daemon logs: "advancing_dag" with team_id=team1, task_id=task1

Sanitizers (C++ layer — not directly affected by this change, but run for regression):

cmake --preset tsan && cmake --build --preset tsan && ctest --preset tsan

Formatting and lint:

just format-check
just lint

---

7. Skills Used

Skills invoked during planning:

• hephaestus:advise — searched team knowledge base before implementation
• Explore subagent — explored codebase structure to understand existing DAG/NATS infrastructure

Team knowledge base skills referenced (Prior Learnings section):

Skill	How it informed the plan
natsc-fetchcontent-cpp20-integration	Confirmed hi.tasks.> wildcard subscription pattern; avoid jsErrCode jerr = 0 (use static_cast<jsErrCode>(0)); use #include "nats.h" not <nats/nats.h>
cpp-ci-sanitizer-tsan-data-race-fixes	Guard DAG state maps with std::mutex; use std::atomic<int> for scalar hot-path fields; relevant if C++ DAG structures are added later
cpp-coverage-container-pinned-clang-format	Use real disconnected NatsClient("nats://127.0.0.1:1") for unit tests; exclude NATS integration layer from gcovr
state-machine-interrupt-handling	Catch specific terminal-state type before generic Exception; propagate FAILED explicitly rather than letting it time out
resume-from-failed-experiment	Three-bug pattern: failed upstream → downstream is_complete() returns true immediately → loop exits → deadlock; fixed by treating FAILED as terminal that unblocks dependents
always-retry-infra-failures	Distinguish infra failure (task.failed verb) from bad result (task.completed with bad payload); do not conflate when cascading cancellations

---

Generated by Claude Code Planner

[mvillmow]

Created 5 follow-up issue(s): #183, #184, #185, #186, #187