Pipeline Tracker Split-Brain — Server-Authoritative Step Completion

docs/safety/hooks.md

@@ -1,6 +1,6 @@
 # Hooks
 
-AutoSkillit registers 43 Claude Code hook scripts: 32 PreToolUse, 10 PostToolUse,
+AutoSkillit registers 42 Claude Code hook scripts: 32 PreToolUse, 9 PostToolUse,
 and 1 SessionStart. Every script is stdlib-only Python so it can run before the
 project virtualenv is on the path. Scripts live in `src/autoskillit/hooks/`
 and are bound to event types in `src/autoskillit/hook_registry.py` via the
@@ -147,7 +147,7 @@ dependencies. Permission is always `allow` — the server-side `_check_pipeline_
 in `run_skill` is the primary enforcer. Fails open on missing tracker or
 malformed input.
 
-## PostToolUse hooks (10)
+## PostToolUse hooks (9)
 
 ### `pretty_output_hook.py`
 **Guarded tools:** all AutoSkillit MCP tools
@@ -188,12 +188,12 @@ state file. When `run_python` calls `check_review_loop`, marks
 `check_review_loop_called: True` in the state so `review_loop_gate.py` will
 unblock `wait_for_ci`/`enqueue_pr`.
 
-### `pipeline_step_post_hook.py`
-**Guarded tool:** `run_skill`
-After a successful `run_skill`, auto-marks the step as `complete` in the
-pipeline tracker file. Appends a progress banner via `updatedMCPToolOutput`.
-Uses `AUTOSKILLIT_DISPATCH_ID` env fallback for `order_id` resolution to
-handle fleet-dispatched pipelines. Fails open on missing tracker or errors.
+### ~~`pipeline_step_post_hook.py`~~ (RETIRED)
+Step completion marking is now **server-authoritative**: the `run_skill`
+handler in `tools_execution.py` writes step completion at the adjudication
+point, using the same tracker resolver as the dependency enforcer. This
+eliminates the split-brain between client-side hook writes and server-side
+enforcement reads that caused false `DEPENDENCY UNMET` denials (#4293).
 
 ### `recipe_confirmed_post_hook.py`
 **Guarded tool:** `run_skill`

src/autoskillit/cli/_prompts_orchestrator.py

@@ -281,6 +281,10 @@ def _build_orchestrator_prompt(
 SPECIFIC HOOK DENIAL PATTERNS:
 - "QUOTA WAIT REQUIRED": Temporary — sleep and retry (see QUOTA DENIAL ROUTING below).
 - "REVIEW LOOP REQUIRED": Call check_review_loop before retrying wait_for_ci/enqueue_pr.
+- "DEPENDENCY UNMET": A prerequisite pipeline step has not completed. \
+Call record_pipeline_step(op="status") to inspect tracker state. \
+Run the missing prerequisite, or escalate if the tracker is stale. \
+Never blind-retry the denied call.
 - All other denials: Follow the corrective instruction in the deny reason text.
 
 QUOTA DENIAL ROUTING — run_skill only (check BEFORE on_failure):

src/autoskillit/core/__init__.pyi

@@ -17,6 +17,8 @@ from ._plugin_cache import _retire_old_versions as _retire_old_versions
 from ._plugin_cache import any_kitchen_open as any_kitchen_open
 from ._plugin_cache import append_retiring_entry as append_retiring_entry
 from ._plugin_cache import clear_kitchens_for_pid as clear_kitchens_for_pid
+from ._plugin_cache import kitchen_entry_alive as kitchen_entry_alive
+from ._plugin_cache import read_active_kitchens_registry as read_active_kitchens_registry
 from ._plugin_cache import register_active_kitchen as register_active_kitchen
 from ._plugin_cache import sweep_retiring_cache as sweep_retiring_cache
 from ._plugin_cache import unregister_active_kitchen as unregister_active_kitchen

src/autoskillit/core/_plugin_cache.py

@@ -148,6 +148,36 @@ def __exit__(self, *_: object) -> None:
             self._lock_file = None
 
 
+def kitchen_entry_alive(entry: dict) -> bool:
+    """Return True if an active_kitchens.json entry's process is still running."""
+    pid = entry.get("pid")
+    if not isinstance(pid, int):
+        return False
+    create_time = entry.get("create_time")
+    stored: float | None = float(create_time) if isinstance(create_time, (int, float)) else None
+    return _pid_alive(pid, stored_create_time=stored)
+
+
+def read_active_kitchens_registry() -> list[dict]:
+    """Return the current active_kitchens.json entries (locked read).
+
+    Public counterpart to the private ``_active_kitchens_path``/``_active_kitchens_lock``
+    pair — callers outside this module must not reach into private submodule internals
+    (REQ-ARCH-001), so this is the sanctioned read surface for registry consumers such
+    as ``prune_stale_kitchen_state``.
+    """
+    akp = _active_kitchens_path()
+    lock = _active_kitchens_lock()
+    if not akp.exists():
+        return []
+    fh = _open_lock(lock)
+    try:
+        data = read_versioned_json(akp, _SCHEMA_VERSION, logger=logger)
+        return data.get("kitchens", []) if data is not None else []
+    finally:
+        fh.close()
+
+
 def _pid_alive(pid: int, stored_create_time: float | None = None) -> bool:
     try:
         os.kill(pid, 0)

src/autoskillit/fleet/_prompts.py

@@ -254,6 +254,10 @@ def _build_food_truck_prompt(
 SPECIFIC HOOK DENIAL PATTERNS:
 - "QUOTA WAIT REQUIRED": Temporary — sleep and retry (see QUOTA DENIAL ROUTING below).
 - "REVIEW LOOP REQUIRED": Call check_review_loop before retrying wait_for_ci/enqueue_pr.
+- "DEPENDENCY UNMET": A prerequisite pipeline step has not completed. \
+Call record_pipeline_step(op="status") to inspect tracker state. \
+Run the missing prerequisite, or escalate if the tracker is stale. \
+Never blind-retry the denied call.
 - All other denials: Follow the corrective instruction in the deny reason text.
 
 QUOTA DENIAL ROUTING — run_skill only (check BEFORE on_failure):

src/autoskillit/hook_registry.py

@@ -299,7 +299,6 @@ def __post_init__(self) -> None:
         scripts=[
             "token_summary_hook.py",
             "quota_post_hook.py",
-            "pipeline_step_post_hook.py",
             "recipe_confirmed_post_hook.py",
         ],
         mechanism="output-rewrite",
@@ -404,6 +403,7 @@ def __post_init__(self) -> None:
         "write_guard.py",
         "pretty_output_hook.py",
         "output_budget_guard.py",
+        "pipeline_step_post_hook.py",
         # Append any future retired basenames here, atomically with the rename commit.
     }
 )

src/autoskillit/hooks/AGENTS.md

@@ -16,12 +16,12 @@ Sub-packages: guards/ (see guards/AGENTS.md), formatters/ (see formatters/AGENTS
 | `token_summary_hook.py` | Appends Token Usage Summary to PR body |
 | `session_start_hook.py` | Injects open-kitchen reminder on resume |
 | `skill_load_post_hook.py` | `PostToolUse`: writes skill-loaded flag for non-Anthropic provider guard |
-| `pipeline_step_post_hook.py` | `PostToolUse`: auto-marks pipeline steps complete after `run_skill` |
+| ~~`pipeline_step_post_hook.py`~~ | **RETIRED** — step completion is now server-authoritative (written at `run_skill` adjudication point in `tools_execution.py`) |
 | `recipe_confirmed_post_hook.py` | `PostToolUse`: writes recipe-load-confirmed marker after first successful `run_skill` |
 | `resume_gate_post_hook.py` | `PostToolUse`: records resume attempts to `resume_gate_state.json` for the reset_dispatch resume gate |
 | `quota_guard_state_post_hook.py` | `PostToolUse`: writes / clears the per-session quota-disable marker after `disable_quota_guard` / `close_kitchen` |
 | `ingredient_lock_guard.py` | PreToolUse guard script (see guards/AGENTS.md) |
-| `_hook_utils.py` | Shared stdlib-only utilities for hook scripts (e.g., `find_project_root`) |
+| `_hook_utils.py` | Shared stdlib-only utilities for hook scripts (e.g., `find_project_root`, `STEP_SUFFIX_RE`) |
 | `_command_classification.py` | Shared stdlib-only command classification primitives for guard scripts (interpreter/wrapper detection, git command classification) |
 | `_policy_event.py` | Typed policy-event formatter for hook provenance messages (stdlib-only) |
 | `shell_capture_hook.py` | `PreToolUse`: input-rewrite hook for Codex shell capture — wraps commands in a lossless capture harness (#4286 / ADR-0006) |

src/autoskillit/hooks/_hook_utils.py

@@ -6,8 +6,11 @@
 
 from __future__ import annotations
 
+import re
 from pathlib import Path
 
+STEP_SUFFIX_RE = re.compile(r"-\d+$")
+
 
 def find_project_root() -> Path:
     """Walk up from CWD to find nearest ancestor containing .autoskillit/."""
@@ -16,15 +19,3 @@ def find_project_root() -> Path:
         if (ancestor / ".autoskillit").is_dir():
             return ancestor
     return cwd
-
-
-def discover_single_tracker_order_id(tracker_dir: Path) -> str:
-    """Return the order_id of the sole tracker file in tracker_dir, or "" if not exactly one."""
-    if not tracker_dir.is_dir():
-        return ""
-    trackers = [
-        f for f in tracker_dir.iterdir() if f.suffix == ".json" and not f.name.startswith(".")
-    ]
-    if len(trackers) == 1:
-        return trackers[0].stem
-    return ""

src/autoskillit/hooks/formatters/AGENTS.md

@@ -19,3 +19,5 @@ PostToolUse output formatters — MCP JSON to Markdown-KV reformatter (30-77% to
 ## Architecture Notes
 
 All `_fmt_*` modules use bare relative imports (`from _fmt_primitives import ...`) because hook scripts run as standalone executables with this directory as CWD — not via the Python package system. `pretty_output_hook.py` is the only entry point.
+
+**Error-fidelity invariant:** `_format_response` in `pretty_output_hook.py` enforces a single-exit-point gate: if a payload dict has a truthy top-level `error` whose text is absent from the rendered output, a final `error: {text}` line is appended. This guarantees no formatter can silently drop the `error` field — the gate fires for all code paths including `gate_error` and `tool_exception` early returns.

src/autoskillit/hooks/formatters/_fmt_execution.py

@@ -38,6 +38,18 @@ def _maybe_provider_line(data: dict, lines: list[str]) -> None:
         lines.append(f"provider: {provider_used}{suffix}")
 
 
+def _maybe_tracker_line(data: dict, lines: list[str], *, blank_before: bool = False) -> None:
+    tracker = data.get("pipeline_tracker")
+    if not isinstance(tracker, dict):
+        return
+    line = (
+        f"--- Pipeline Tracker: {tracker.get('step', '')} {tracker.get('status', '')}"
+        f" ({tracker.get('order_id', '')}) ---"
+    )
+    lines.extend([""] if blank_before else [])
+    lines.append(line)
+
+
 def _fmt_run_skill(data: dict, pipeline: bool) -> str:
     """Format run_skill result as Markdown-KV."""
     success = data.get("success", False)
@@ -62,6 +74,7 @@ def _fmt_run_skill(data: dict, pipeline: bool) -> str:
         if worktree:
             lines.append(f"worktree_path: {worktree}")
         _maybe_provider_line(data, lines)
+        _maybe_tracker_line(data, lines)
         result = data.get("result", "")
         if result:
             lines.append(f"\nresult:\n{result}")
@@ -99,6 +112,7 @@ def _fmt_run_skill(data: dict, pipeline: bool) -> str:
         cw = token_usage.get("cache_write_tokens", 0)
         if cw:
             lines.append(f"tokens_cache_write: {_fmt_tokens(cw)}")
+    _maybe_tracker_line(data, lines, blank_before=True)
     result = data.get("result", "")
     if result:
         lines.extend(["", "### Result", result])
@@ -206,6 +220,8 @@ def _fmt_test_check(data: dict, _pipeline: bool) -> str:
         "has_progress_evidence",
         "provider_used",
         "provider_fallback",
+        "pipeline_tracker",
+        "error",
     }
 )
 _FMT_RUN_SKILL_SUPPRESSED: frozenset[str] = frozenset(
@@ -231,6 +247,8 @@ def _fmt_test_check(data: dict, _pipeline: bool) -> str:
         "completion_required",
         "ndjson_unknown_event_count",
         "ndjson_unknown_item_count",
+        "stage",
+        "retriable",
     }
 )
 
@@ -242,9 +260,10 @@ def _fmt_test_check(data: dict, _pipeline: bool) -> str:
         "stderr",
         "stdout_artifact_path",
         "stderr_artifact_path",
+        "error",
     }
 )
-_FMT_RUN_CMD_SUPPRESSED: frozenset[str] = frozenset({"error"})
+_FMT_RUN_CMD_SUPPRESSED: frozenset[str] = frozenset()
 
 _FMT_TEST_CHECK_RENDERED: frozenset[str] = frozenset(
     {

src/autoskillit/hooks/formatters/_fmt_status.py

@@ -129,6 +129,9 @@ def _fmt_kitchen_status(data: dict, _pipeline: bool) -> str:
     warning = data.get("warning")
     if warning:
         lines.extend(["", f"warning: {warning}"])
+    error = data.get("error")
+    if error:
+        lines.extend(["", f"error: {error}"])
     return "\n".join(lines)
 
 
@@ -156,20 +159,18 @@ def _fmt_clone_repo(data: dict, _pipeline: bool) -> str:
 
 
 _FMT_TOKEN_SUMMARY_RENDERED: frozenset[str] = frozenset(
-    {"steps", "total", "mcp_responses", "model_totals"}
+    {"steps", "total", "mcp_responses", "model_totals", "error"}
 )
 _FMT_TOKEN_SUMMARY_SUPPRESSED: frozenset[str] = frozenset(
     {
         "success",  # only emitted from exception guard paths, not accessed by the formatter
-        "error",  # only emitted from exception guard paths, not accessed by the formatter
     }
 )
 
-_FMT_TIMING_SUMMARY_RENDERED: frozenset[str] = frozenset({"steps", "total"})
+_FMT_TIMING_SUMMARY_RENDERED: frozenset[str] = frozenset({"steps", "total", "error"})
 _FMT_TIMING_SUMMARY_SUPPRESSED: frozenset[str] = frozenset(
     {
         "success",  # only emitted from exception guard paths, not accessed by the formatter
-        "error",  # only emitted from exception guard paths, not accessed by the formatter
     }
 )
 

src/autoskillit/hooks/formatters/pretty_output_hook.py

@@ -285,20 +285,21 @@ def with_spill(formatted: str) -> str:
         return with_spill(str(data["preview"]))
 
     if data.get("subtype") == "gate_error":
-        return with_spill(_fmt_gate_error(data, pipeline))
-    if data.get("subtype") == "tool_exception":
-        return with_spill(_fmt_tool_exception(data, pipeline))
-
-    if short_name in _UNFORMATTED_TOOLS:
-        return with_spill(
-            _fmt_generic(short_name, data, pipeline, artifact_backed=artifact_backed)
-        )
-
-    formatter = _FORMATTERS.get(short_name)
-    if formatter is not None:
-        return with_spill(formatter(data, pipeline))
-
-    return with_spill(_fmt_generic(short_name, data, pipeline, artifact_backed=artifact_backed))
+        rendered = _fmt_gate_error(data, pipeline)
+    elif data.get("subtype") == "tool_exception":
+        rendered = _fmt_tool_exception(data, pipeline)
+    elif short_name in _UNFORMATTED_TOOLS:
+        rendered = _fmt_generic(short_name, data, pipeline, artifact_backed=artifact_backed)
+    elif (formatter := _FORMATTERS.get(short_name)) is not None:
+        rendered = formatter(data, pipeline)
+    else:
+        rendered = _fmt_generic(short_name, data, pipeline, artifact_backed=artifact_backed)
+
+    error_text = data.get("error", "")
+    if error_text and isinstance(error_text, str) and error_text not in rendered:
+        rendered = f"{rendered}\nerror: {error_text}"
+
+    return with_spill(rendered)
 
 
 def main() -> None:

src/autoskillit/hooks/guards/pipeline_step_guard.py

@@ -4,26 +4,55 @@
 Non-blocking advisory — permissionDecision is always "allow". The server-side
 _check_pipeline_deps in run_skill is the primary enforcer.
 
+Tracker resolution uses the kitchen_id from the merged hook config (the same
+rule the server uses) rather than the fragile single-file discovery heuristic.
+
 Stdlib-only — runs under any Python interpreter without the autoskillit package.
 """
 
 from __future__ import annotations
 
 import json
 import os
-import re
 import sys
 from pathlib import Path
 
 _HOOKS_DIR = str(Path(__file__).resolve().parent.parent)
 if _HOOKS_DIR not in sys.path:
     sys.path.insert(0, _HOOKS_DIR)
 
-from _hook_utils import (  # type: ignore[import-not-found]  # noqa: E402
-    discover_single_tracker_order_id,
-)
-
-_SUFFIX_RE = re.compile(r"-\d+$")
+from _hook_settings import read_merged_hook_config  # type: ignore[import-not-found]  # noqa: E402
+from _hook_utils import STEP_SUFFIX_RE  # type: ignore[import-not-found]  # noqa: E402
+
+
+def _resolve_order_id_from_kitchen(tracker_dir: Path, kitchen_id: str) -> str:
+    """Select tracker by internal kitchen_id field (same rule as the server).
+
+    Skips the self-named file (``{kitchen_id}.json``) from the candidate scan,
+    matching ``resolve_tracker_order_id`` in ``tools_pipeline_tracker.py``.
+    When exactly one non-self candidate exists, returns that candidate's stem.
+    When no non-self candidates exist, returns ``kitchen_id`` (the self-named
+    file is the implicit default). Returns ``""`` on ambiguity (>1 candidate).
+    """
+    if not tracker_dir.is_dir():
+        return ""
+    active: set[str] = set()
+    for f in tracker_dir.iterdir():
+        if f.suffix != ".json" or f.name.startswith("."):
+            continue
+        if f.stem == kitchen_id:
+            continue
+        try:
+            data = json.loads(f.read_text())
+        except (json.JSONDecodeError, OSError):
+            continue
+        if data.get("kitchen_id") == kitchen_id:
+            active.add(f.stem)
+    if len(active) > 1:
+        return ""
+    if len(active) == 1:
+        return next(iter(active))
+    return kitchen_id
 
 
 def main() -> None:
@@ -40,11 +69,29 @@ def main() -> None:
     order_id = tool_input.get("order_id", "") or os.environ.get("AUTOSKILLIT_DISPATCH_ID", "")
     if not order_id:
         tracker_dir = Path.cwd() / ".autoskillit" / "temp" / "pipeline_tracker"
-        order_id = discover_single_tracker_order_id(tracker_dir)
+        hook_config = read_merged_hook_config()
+        kitchen_id = hook_config.get("kitchen_id", "")
+        if kitchen_id:
+            order_id = _resolve_order_id_from_kitchen(tracker_dir, kitchen_id)
         if not order_id:
+            print(
+                json.dumps(
+                    {
+                        "hookSpecificOutput": {
+                            "hookEventName": "PreToolUse",
+                            "permissionDecision": "allow",
+                            "additionalContext": (
+                                "Pipeline step guard: cannot resolve tracker — "
+                                "no order_id, no kitchen_id, or ambiguous tracker state. "
+                                "The server-side enforcer will handle dependency checks."
+                            ),
+                        }
+                    }
+                )
+            )
             sys.exit(0)
 
-    canonical = _SUFFIX_RE.sub("", step_name)
+    canonical = STEP_SUFFIX_RE.sub("", step_name)
     tracker_path = Path.cwd() / ".autoskillit" / "temp" / "pipeline_tracker" / f"{order_id}.json"
     if not tracker_path.exists():
         sys.exit(0)