[MINOR] §15: No audit trail for workflow execution events

[mvillmow]

Evidence

src/telemachy/executor.py (logging only)

Description

Workflow execution is logged via Python's logging module but there is no tamper-evident audit trail (e.g., structured event log to a file or external service) tracking who ran which workflow, when, with what parameters, and what the outcome was.

Part of #92

[mvillmow]

Implementation Plan

Good — verified. Re-planning now to fix the major findings (DIP/singleton, hash chain continuity) and the minor ones.

Implementation Plan

Objective

Add a tamper-evident, structured audit trail for workflow execution events to WorkflowExecutor (src/telemachy/executor.py), satisfying §15 by durably recording who ran a workflow, when, with what parameters, and the outcome of every task and the workflow as a whole. The audit sink is injected into WorkflowExecutor, not coupled via a process-global singleton — this resolves the DIP/modularity finding from the prior review. The hash chain (used by default when an audit log is enabled, so the §15 "tamper-evident" requirement holds without an extra opt-in) is continuity-aware across process restarts: it resumes from the last record's hash on the existing file rather than silently restarting at the zero genesis, which closes the chain-continuity gap the reviewer flagged.

Approach

Decision 1 — Dependency-injected AuditSink, no process-global singleton. Prior review flagged _sink = _AuditSink() at module-import time as a live DIP/Modularity defect. Resolution: AuditSink is a class with no module-level instance; WorkflowExecutor.__init__ takes a sink: AuditSink | None = None parameter; when omitted, the executor constructs one from settings inside __init__ (lazy, per-executor, not at module import). Tests inject a AuditSink(path=tmp_path/"audit.jsonl", ...) directly — no monkeypatching of module internals. Evidence: matches the existing injection pattern for AgamemnonClient (executor.py:31-39, cli.py:187) where the client is constructed once in cli.run and passed in.

Decision 2 — JSONL file sink, stdlib only. Confirmed pixi.toml [pypi-dependencies] lists only pydantic/httpx/yaml/rich/typer — adding structlog/loguru would force a new docs/license-audit.md row per CLAUDE.md guardrails. Stdlib json + pathlib.Path + hashlib covers everything. JSONL streams cleanly into Loki/Splunk later (the §15 "external service" tail-and-forward path).

Decision 3 — Hash chain ON by default when audit is enabled; continuity-aware on restart. Prior review correctly flagged that opt-in tamper-evidence does not satisfy "tamper-evident audit trail" in the default deployment, and that resetting _prev_hash to zero on process restart silently fakes a clean chain start. Resolution:

• When AUDIT_LOG_PATH is set, hash-chain is on by default (AUDIT_HASH_CHAIN=true is the new default; false is allowed as an explicit opt-out for tooling/tests).
• On AuditSink.__init__, if the target file already exists and is non-empty, the sink reads the last JSON line and seeds _prev_hash from its hash field. If the last line is malformed or missing hash, the sink raises AuditChainError at startup rather than silently restarting at zero. Tests cover both the resume and the corrupt-tail paths.

Decision 4 — actor field documents OS-portability limitation in code. Prior review flagged os.environ.get("USER", "unknown") silently degrades on Windows/CI. Resolution: source user from USER then fall back to USERNAME (Windows) before "unknown", and add an inline comment naming the limitation. Evidence: pixi.toml lists platforms = ["linux-64"] only, so Linux is the supported target, but the fallback keeps the field meaningful on dev machines and CI runners that set USERNAME.

Decision 5 — Sync emit (not async) to close the await-forgotten POLA hole. Prior review flagged that an async def emit silently drops calls if a caller forgets await. Resolution: AuditSink.emit is a sync method protected by a threading.Lock. Disk writes here are line-sized (~1 KB) and flushed per record; the executor's hot path already does heavyweight I/O via httpx. A sync call cannot be silently dropped. The lock guards both asyncio.gather fan-outs (single-threaded but reentrant via coroutines) and any future thread-pool callers.

Decision 6 — Import-failure isolation via a thin façade. Prior review flagged no rollback if audit.py raises at import. Resolution: executor.py does from telemachy.audit import AuditSink, NullSink once. NullSink is a zero-dependency drop-in with the same interface that no-ops every method; if construction of a real AuditSink raises (bad path, unreadable existing log), WorkflowExecutor.__init__ falls back to NullSink() and emits a single logger.warning. The executor never depends on the file path being writable at construction time.

Decision 7 — Emit-points at existing state-transition boundaries. Confirmed the 10 sites in executor.py via re-read: workflow start (:103), workflow completed (:128), workflow cancelled (:134), workflow failed (:141), agent created+woken (:194-196), team created (:235), task submitted (:307), task completed/failed via the existing hook branches (:362-367), team/agent deleted in teardown (:399, :407). Verified state.started_at / state.completed_at exist on WorkflowState (models.py:180-181) so the _duration helper is safe.

Decision 8 — Test ordering: contract tests land first. Prior review noted P3/TDD ordering. The Implementation Order below writes tests/test_audit.py (defining the sink contract) before audit.py is implemented, then writes the executor integration test before wiring executor.py. Production code lands red-then-green per step.

Files to Create

src/telemachy/audit.py

Defines AuditSink (injectable, per-instance), NullSink (drop-in no-op for fallback / disabled), and AuditChainError. Sync emit. Continuity-aware hash-chain resume. No module-level singleton; no settings access at import.

"""Append-only JSONL audit log for workflow execution events.

Sinks are CONSTRUCTED, not module-global. Inject into WorkflowExecutor.
"""

from __future__ import annotations

import atexit
import hashlib
import json
import logging
import os
import threading
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Protocol, TextIO

logger = logging.getLogger(__name__)

_GENESIS_HASH = "0" * 64


class AuditChainError(RuntimeError):
    """Raised when an existing audit log cannot be resumed (corrupt tail / missing hash)."""


class AuditSinkProtocol(Protocol):
    def emit(self, event_type: str, *, workflow_id: str | None = None, **fields: Any) -> None: ...
    def close(self) -> None: ...


class NullSink:
    """No-op sink. Used when AUDIT_LOG_PATH is unset or sink construction failed."""

    def emit(self, event_type: str, *, workflow_id: str | None = None, **fields: Any) -> None:
        return None

    def close(self) -> None:
        return None


class AuditSink:
    """Append-only JSONL sink with optional SHA-256 hash chain.

    Continuity: if `path` already exists and is non-empty, the previous run's
    last `hash` is read and used as the seed for `prev_hash`. This prevents a
    process restart from silently creating a false chain start at the zero
    genesis value (see issue #187 prior review).
    """

    def __init__(
        self,
        *,
        path: Path,
        host_id: str,
        hash_chain: bool = True,
    ) -> None:
        self._path: Path = path
        self._chain: bool = hash_chain
        self._lock = threading.Lock()
        self._fh: TextIO | None = None
        # USER works on POSIX; USERNAME is the Windows fallback. CI sandboxes that
        # strip both env vars get "unknown" — fidelity loss is logged once below.
        user = os.environ.get("USER") or os.environ.get("USERNAME") or "unknown"
        if user == "unknown":
            logger.warning("Audit actor.user resolved to 'unknown' — USER/USERNAME unset")
        self._actor: dict[str, str] = {"host_id": host_id, "user": user}
        self._prev_hash: str = _GENESIS_HASH
        if self._chain:
            self._prev_hash = self._resume_chain()
        atexit.register(self.close)

    def _resume_chain(self) -> str:
        if not self._path.exists() or self._path.stat().st_size == 0:
            return _GENESIS_HASH
        last_line = ""
        with self._path.open("rb") as fh:
            for line in fh:
                if line.strip():
                    last_line = line.decode("utf-8")
        if not last_line:
            return _GENESIS_HASH
        try:
            rec = json.loads(last_line)
            prev = rec["hash"]
        except (json.JSONDecodeError, KeyError) as exc:
            raise AuditChainError(
                f"Cannot resume hash chain from {self._path}: last line missing 'hash'"
            ) from exc
        if not isinstance(prev, str) or len(prev) != 64:
            raise AuditChainError(f"Invalid prior hash in {self._path}: {prev!r}")
        return prev

    def _open(self) -> TextIO:
        if self._fh is None:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            self._fh = self._path.open("a", encoding="utf-8")
        return self._fh

    def emit(self, event_type: str, *, workflow_id: str | None = None, **fields: Any) -> None:
        record: dict[str, Any] = {
            "timestamp": datetime.now(tz=UTC).isoformat(timespec="microseconds"),
            "event_type": event_type,
            "workflow_id": workflow_id,
            "actor": self._actor,
            "payload": fields,
        }
        if self._chain:
            record["prev_hash"] = self._prev_hash
            digest = hashlib.sha256(
                json.dumps(record, sort_keys=True, default=str).encode("utf-8")
            ).hexdigest()
            record["hash"] = digest
        line = json.dumps(record, sort_keys=True, default=str) + "\n"
        with self._lock:
            try:
                fh = self._open()
                fh.write(line)
                fh.flush()
                os.fsync(fh.fileno())
            except OSError as exc:
                logger.warning("Audit write failed for %s: %s", self._path, exc)
                return
            if self._chain:
                self._prev_hash = record["hash"]

    def close(self) -> None:
        with self._lock:
            if self._fh is not None:
                try:
                    self._fh.close()
                finally:
                    self._fh = None


def build_sink_from_settings() -> AuditSinkProtocol:
    """Construct a sink from current settings. Returns NullSink on failure or when disabled."""
    from telemachy.config import settings
    if not settings.audit_log_path:
        return NullSink()
    try:
        return AuditSink(
            path=Path(settings.audit_log_path),
            host_id=settings.host_id,
            hash_chain=settings.audit_hash_chain,
        )
    except (AuditChainError, OSError) as exc:
        logger.warning("Audit sink disabled — construction failed: %s", exc)
        return NullSink()

tests/test_audit.py

Contract tests written BEFORE audit.py (TDD ordering). Covers: no-op NullSink; basic emit writes one well-formed JSON line; thread-safety under 100 concurrent emits; hash chain continuity within one process; hash chain resume across two AuditSink instances pointing at the same path; corrupt-tail raises AuditChainError; OSError on write is swallowed.

import json
import threading
from pathlib import Path

import pytest

from telemachy.audit import AuditChainError, AuditSink, NullSink


def _new_sink(tmp_path: Path, *, chain: bool = True) -> AuditSink:
    return AuditSink(path=tmp_path / "audit.jsonl", host_id="test-host", hash_chain=chain)


def test_null_sink_is_noop(tmp_path):
    sink = NullSink()
    sink.emit("workflow.started", workflow_id="x")
    sink.close()
    assert not (tmp_path / "audit.jsonl").exists()


def test_emit_writes_one_jsonl_line(tmp_path):
    sink = _new_sink(tmp_path, chain=False)
    sink.emit("workflow.started", workflow_id="wf-1", spec_name="demo")
    sink.close()
    lines = (tmp_path / "audit.jsonl").read_text().splitlines()
    assert len(lines) == 1
    rec = json.loads(lines[0])
    assert rec["event_type"] == "workflow.started"
    assert rec["workflow_id"] == "wf-1"
    assert rec["payload"] == {"spec_name": "demo"}
    assert rec["actor"] == {"host_id": "test-host", "user": rec["actor"]["user"]}
    assert "timestamp" in rec


def test_concurrent_emit_no_interleaving(tmp_path):
    sink = _new_sink(tmp_path, chain=False)
    def fire(i: int) -> None:
        sink.emit("e", workflow_id=None, i=i)
    threads = [threading.Thread(target=fire, args=(i,)) for i in range(100)]
    for t in threads: t.start()
    for t in threads: t.join()
    sink.close()
    lines = (tmp_path / "audit.jsonl").read_text().splitlines()
    assert len(lines) == 100
    for line in lines:
        json.loads(line)  # all intact


def test_hash_chain_within_process(tmp_path):
    sink = _new_sink(tmp_path)
    sink.emit("a", workflow_id="w")
    sink.emit("b", workflow_id="w")
    sink.close()
    lines = [json.loads(l) for l in (tmp_path / "audit.jsonl").read_text().splitlines()]
    assert lines[0]["prev_hash"] == "0" * 64
    assert lines[1]["prev_hash"] == lines[0]["hash"]


def test_hash_chain_resumes_across_sinks(tmp_path):
    """Restart safety: a second sink on the same file resumes, not zero-restarts."""
    s1 = _new_sink(tmp_path)
    s1.emit("a", workflow_id="w")
    s1.close()
    s2 = _new_sink(tmp_path)
    s2.emit("b", workflow_id="w")
    s2.close()
    lines = [json.loads(l) for l in (tmp_path / "audit.jsonl").read_text().splitlines()]
    assert lines[1]["prev_hash"] == lines[0]["hash"]
    assert lines[1]["prev_hash"] != "0" * 64


def test_corrupt_tail_raises_on_resume(tmp_path):
    path = tmp_path / "audit.jsonl"
    path.write_text("{not json\n")
    with pytest.raises(AuditChainError):
        AuditSink(path=path, host_id="h", hash_chain=True)


def test_write_error_swallowed(tmp_path, monkeypatch):
    sink = _new_sink(tmp_path, chain=False)
    def boom(*a, **kw):
        raise OSError("disk full")
    monkeypatch.setattr(sink, "_open", boom)
    sink.emit("a", workflow_id="x")  # must not raise

docs/audit-log.md

Short reference: JSON schema for each event_type, env-var config, retention guidance, the hash-chain verification recipe, and a stated portability note ("USER/USERNAME may be absent in some CI runners; actor.user falls back to 'unknown'"). Permitted under CLAUDE.md (docs/ already contains license-audit.md, backwards-compat.md).

Files to Modify

src/telemachy/config.py

Insert after default_workflow_timeout: float = 7200.0 (config.py:43):

    audit_log_path: str | None = None
    audit_hash_chain: bool = True  # on by default when audit_log_path is set (§15 tamper-evidence)

src/telemachy/executor.py

Add the import and inject the sink into WorkflowExecutor so the executor is no longer coupled to a module-global. After from telemachy.models import ... (executor.py:15):

from telemachy.audit import AuditSink, AuditSinkProtocol, NullSink, build_sink_from_settings

Modify WorkflowExecutor.__init__ (executor.py:31-57) to accept and store sink:

    def __init__(
        self,
        client: AgamemnonClient,
        poll_interval: float = 5.0,
        dry_run: bool = False,
        stop_event: asyncio.Event | None = None,
        max_concurrent_provisioning: int = 16,
        sink: AuditSinkProtocol | None = None,
    ) -> None:
        self._client = client
        self._poll_interval = poll_interval
        self._dry_run = dry_run
        self._stop_event = stop_event
        self._provision_semaphore = asyncio.Semaphore(max(1, max_concurrent_provisioning))
        # Inject or build from settings; build_sink_from_settings returns NullSink on failure,
        # so executor construction never raises due to a bad audit path.
        self._sink: AuditSinkProtocol = sink if sink is not None else build_sink_from_settings()
        self._hooks: dict[str, list[Callable[..., Any]]] = {
            "on_task_complete": [],
            "on_task_failed": [],
            "on_workflow_complete": [],
            "on_workflow_failed": [],
        }
        self._emitted_task_events: set[str] = set()

Emit calls (all sync — no await) at the existing transition sites:

After logger.info("Starting workflow ... at executor.py:103:

        self._sink.emit(
            "workflow.started",
            workflow_id=workflow_id,
            spec_name=spec.name,
            agents=[a.name for a in spec.agents],
            teams=[t.name for t in spec.teams],
            teardown=spec.teardown,
        )

After logger.info("Workflow ... completed successfully") at executor.py:128:

                self._sink.emit(
                    "workflow.completed",
                    workflow_id=state.workflow_id,
                    spec_name=spec.name,
                    duration_seconds=_duration(state),
                )

After logger.warning("Workflow ... was cancelled via stop event") at executor.py:124 AND after the bare-cancel branch at executor.py:134:

            self._sink.emit("workflow.cancelled", workflow_id=state.workflow_id, spec_name=spec.name)

After logger.error("Workflow ... failed: ...") at executor.py:141:

            self._sink.emit(
                "workflow.failed",
                workflow_id=state.workflow_id,
                spec_name=spec.name,
                error=str(exc),
            )

After logger.debug("Woke agent ...") at executor.py:196:

        self._sink.emit(
            "agent.created",
            agent_name=spec.name,
            agent_id=agent_id,
            runtime=spec.runtime,
            program=spec.program,
        )

After logger.info("Created team ...") at executor.py:235:

        self._sink.emit(
            "team.created",
            team_name=team_spec.name,
            team_id=team_id,
            members=team_spec.agents,
        )

After logger.info("Submitted task ...") at executor.py:307:

                self._sink.emit(
                    "task.submitted",
                    team_id=team_id,
                    task_subject=task_spec.subject,
                    task_id=task_id,
                    assign_to=task_spec.assign_to,
                    blocked_by=task_spec.blocked_by,
                )

Inside _monitor_completion, at the failed/error branch after the existing await self._emit("on_task_failed", ...) at executor.py:364:

                            self._sink.emit(
                                "task.failed",
                                workflow_id=state.workflow_id,
                                team=team_name,
                                task_subject=task_subject,
                            )

At the completed branch after the existing await self._emit("on_task_complete", ...) at executor.py:367:

                        self._sink.emit(
                            "task.completed",
                            workflow_id=state.workflow_id,
                            team=team_name,
                            task_subject=task_subject,
                        )

After successful delete_team (after logger.debug("Deleted team ...") at executor.py:400):

                self._sink.emit("team.deleted", team_name=name, team_id=team_id)

After successful delete_agent (after logger.debug("Deleted agent ...") at executor.py:407):

                self._sink.emit("agent.deleted", agent_name=name, agent_id=agent_id)

Add helper alongside _now() at executor.py:414:

def _duration(state: WorkflowState) -> float | None:
    # state.started_at and state.completed_at exist on WorkflowState (models.py:180-181)
    if state.started_at and state.completed_at:
        start = datetime.fromisoformat(state.started_at)
        end = datetime.fromisoformat(state.completed_at)
        return (end - start).total_seconds()
    return None

tests/test_executor.py

Append at module level (NOT inside class TestProvisioning) — the helpers _make_spec/_make_mock_client are at lines 18–53 and class TestProvisioning starts ~line 60, so place this AFTER the class block or under a new class TestAuditTrail: section, keeping it module-level. Inject the sink directly — no monkeypatching of internals.

import json
from pathlib import Path

from telemachy.audit import AuditSink


class TestAuditTrail:
    @pytest.mark.asyncio
    async def test_executor_emits_full_event_sequence(self, tmp_path: Path) -> None:
        log = tmp_path / "audit.jsonl"
        sink = AuditSink(path=log, host_id="test-host", hash_chain=True)
        client = _make_mock_client()
        executor = WorkflowExecutor(client, poll_interval=0.01, sink=sink)
        await executor.execute(_make_spec())
        sink.close()
        records = [json.loads(l) for l in log.read_text().splitlines()]
        events = [r["event_type"] for r in records]
        assert events[0] == "workflow.started"
        assert "agent.created" in events
        assert "team.created" in events
        assert "task.submitted" in events
        assert "task.completed" in events
        assert "workflow.completed" in events
        # Chain continuity end-to-end
        for prev, curr in zip(records, records[1:]):
            assert curr["prev_hash"] == prev["hash"]
        # Actor present on every record
        for r in records:
            assert r["actor"]["host_id"] == "test-host"

src/telemachy/cli.py

In run (cli.py:187), build the sink once and inject it so the CLI path uses the same injection seam as tests:

            async with AgamemnonClient(**settings.client_kwargs()) as client:
                from telemachy.audit import build_sink_from_settings
                sink = build_sink_from_settings()
                executor = WorkflowExecutor(client, stop_event=stop_event, sink=sink)
                executor.add_hook("on_task_complete", _on_task_complete)
                result = await executor.execute(spec)

CLAUDE.md

Append rows to the Environment Variables table:

| `AUDIT_LOG_PATH` | (unset) | Path to JSONL audit log. When unset, audit emission is a no-op (NullSink). |
| `AUDIT_HASH_CHAIN` | `true` | SHA-256 hash chain for tamper-evident audit log. Resumes from existing file on restart. Disable only for tooling/tests. |

.env.example

Add commented-out:

# AUDIT_LOG_PATH=/var/log/telemachy/audit.jsonl
# AUDIT_HASH_CHAIN=true

Implementation Order

TDD ordering — contract tests land before production code per the prior review's P3/TDD finding.

1. Add audit_log_path (default None) and audit_hash_chain (default True) to Settings in src/telemachy/config.py.
2. Write tests/test_audit.py (the 7 contract tests above). Tests RED — audit module doesn't exist yet.
3. Implement src/telemachy/audit.py (NullSink, AuditSink, AuditChainError, build_sink_from_settings). Tests GREEN.
4. Add TestAuditTrail::test_executor_emits_full_event_sequence to tests/test_executor.py. RED — executor doesn't accept sink.
5. Modify WorkflowExecutor.__init__ to accept sink: AuditSinkProtocol | None = None, then wire all 11 emit sites in executor.py and the _duration helper. Tests GREEN.
6. Modify cli.py:187 to build and inject the sink. Manual smoke: AUDIT_LOG_PATH=/tmp/wf.jsonl pixi run python -m telemachy.cli run workflows/example.yaml --dry-run produces a non-empty JSONL file with chain continuity.
7. Update CLAUDE.md env-var table, .env.example, write docs/audit-log.md.
8. Run pixi run ruff check src tests, pixi run mypy src/telemachy --ignore-missing-imports, pixi run pytest --cov=src/telemachy --cov-fail-under=75.

Verification

pixi run pytest tests/test_audit.py -v                                              # criterion: sink writes structured JSON records with who/when/what fields
pixi run pytest tests/test_audit.py -k concurrent -v                                # criterion: concurrent writes do not interleave (integrity)
pixi run pytest tests/test_audit.py -k resumes_across -v                            # criterion: tamper-evident chain survives process restart (review finding #2)
pixi run pytest tests/test_audit.py -k corrupt_tail -v                              # criterion: corrupt tail refused, never silently zero-restarts
pixi run pytest tests/test_executor.py -k TestAuditTrail -v                         # criterion: full workflow run emits the complete who/when/what/outcome trail
pixi run pytest --cov=src/telemachy --cov-fail-under=75                             # criterion: 75% coverage gate from CLAUDE.md holds
pixi run ruff check src tests                                                       # criterion: lint clean
pixi run mypy src/telemachy --ignore-missing-imports                                # criterion: type-clean per CLAUDE.md
AUDIT_LOG_PATH=/tmp/wf-audit.jsonl pixi run python -m telemachy.cli run \
  workflows/example.yaml --dry-run \
  && python -c "import json,pathlib; ls=[json.loads(l) for l in pathlib.Path('/tmp/wf-audit.jsonl').read_text().splitlines()]; assert ls, 'empty audit log'; assert ls[0]['prev_hash']=='0'*64; assert all(ls[i]['prev_hash']==ls[i-1]['hash'] for i in range(1,len(ls))); print('chain OK', len(ls), 'records')"
                                                                                    # criterion: end-to-end tamper-evident chain (confirmed --dry-run flag exists at cli.py:128-131)

Skills Used

• logging-stdlib-json-formatter (KB) — stdlib json.dumps + UTC ISO 8601 microsecond timestamps, no third-party logging dep.
• persist-automation-logs (KB) — disk persistence: append-mode handle, parent-dir mkdir, flush + os.fsync per record.
• execution-stage-logging (KB) — event taxonomy (workflow.*, team.*, task.*) and duration_seconds on completion.
• logging-handler-registry-deduplication (KB) — informed avoiding double-open of the file handle via the per-sink lazy _open().
• logging-context-deduplication (KB) — actor resolved once in AuditSink.__init__ rather than threaded through every emit.
• architecture-rest-api-httpx-integration (KB) — confirmed JSONL on disk is the correct boundary for future external sink integration (tail-and-forward), not a synchronous HTTP call in the executor hot path.
• Direct verification reads: executor.py (whole file), cli.py:128-201, config.py, models.py:180-181, tests/test_executor.py:18-65, pixi.toml — every line/path reference grounded.

Changes from Review

1. Major fix: ecosystem audit — migrate config to pydantic-settings #1 (P5/DIP + P6/Modularity — singleton coupling). Removed the module-global _sink. AuditSink is now constructed per-executor and injected through WorkflowExecutor.__init__(sink=...). cli.py builds it once and passes it in (matches the existing AgamemnonClient injection pattern at executor.py:31/cli.py:187). Tests instantiate AuditSink(path=tmp_path/...) directly — no monkeypatch.setattr(audit, "_sink", ...).

2. Major Remove backlog from _DONE_STATUSES — it is an initial state, not terminal #2 (Hash chain continuity gap). AuditSink.__init__ now calls _resume_chain() which reads the last JSON line of the existing log and seeds _prev_hash from its hash field. A corrupt or hash-less tail raises AuditChainError rather than silently zero-restarting. Covered by test_hash_chain_resumes_across_sinks and test_corrupt_tail_raises_on_resume.

3. Minor Implement NATS event-driven monitoring instead of polling #3 (Hash chain opt-in default). audit_hash_chain now defaults to True. When AUDIT_LOG_PATH is set, tamper-evidence is on by default; operators can set AUDIT_HASH_CHAIN=false for tooling but the default deployment satisfies §15.

4. Minor Add tests for executor error paths (partial failures, cancellation) #4 (test_executor.py line-number claim was wrong). Corrected: helpers are at lines 18–53, class TestProvisioning starts ~line 60. The new test is placed at module level (or in a new sibling class TestAuditTrail), not inside TestProvisioning.

5. Minor Add team deletion to teardown — currently only deletes agents #5 (--dry-run flag unverified). Verified by grep at cli.py:128-131 (typer.Option("--dry-run/--no-dry-run", ...)); the end-to-end verification command is safe.

6. Minor Fix CLAUDE.md workflow schema — uses title/depends_on but models use subject/blocked_by #6 (state.started_at / completed_at unverified). Verified at models.py:180-181 — both fields declared str | None = None. _duration is safe and now carries an inline citation.

7. Minor Add timeout and max poll count to _monitor_completion #7 (P3/TDD ordering). Implementation Order now writes tests/test_audit.py (step 2) BEFORE implementing audit.py (step 3), and writes TestAuditTrail (step 4) BEFORE wiring executor.py (step 5). RED-then-GREEN per step.

8. Minor Add tests for CLI module #8 (P7/POLA — async emit silently dropped if await missed). AuditSink.emit is now sync and protected by threading.Lock. A missing await is no longer possible. The lock guards asyncio.gather fan-outs (in single-threaded asyncio, sync writes are atomic per call anyway, but the lock makes future thread-pool use safe).

9. Minor Add tests for AgamemnonClient HTTP interactions #9 (No import-failure rollback). Added NullSink and build_sink_from_settings() which returns NullSink on any construction failure with a single warning log. WorkflowExecutor.__init__ therefore cannot fail because of audit configuration; a bad path degrades to no-op.

10. Portability finding (USER env var). AuditSink.__init__ now tries USER → USERNAME → "unknown" and logs a warning when it falls through. Documented in docs/audit-log.md.

11. atexit / SIGKILL caveat. AuditSink.emit now calls os.fsync(fh.fileno()) after flush() — best-effort durability per record so SIGKILL between events does not lose previously-written audit lines. Documented as best-effort (the in-flight current line may still be lost on hard kill).

---

Final review verdict (from strict review loop)

Good — all factual claims verified accurate. Now let me analyze the R1 plan systematically against the prior review's findings.

R1 Plan Review — Issue #187: Audit Trail

Verifying Prior Iteration Findings Were Addressed

Major #1 (P5/DIP + P6/Modularity — singleton coupling)
The prior review found _sink = _AuditSink() at module import time as a process-global singleton. The R1 plan resolves this explicitly in Decision 1: AuditSink has no module-level instance; WorkflowExecutor.__init__ takes sink: AuditSinkProtocol | None = None; when omitted, build_sink_from_settings() is called inside __init__. The plan confirms tests inject AuditSink(path=tmp_path/"audit.jsonl", ...) directly. The existing injection pattern for AgamemnonClient (executor.py:31-39, cli.py:187) is cited as a model. This is a genuine structural fix, not an acknowledgment-without-change. RESOLVED.

Major #2 (Hash chain continuity gap)
The prior review flagged process restart resetting _prev_hash to all-zeros. Decision 3 in R1 resolves this: AuditSink.__init__ calls _resume_chain() which reads the last JSON line and seeds _prev_hash from its hash field. A corrupt or hash-less tail raises AuditChainError rather than silently zero-restarting. test_hash_chain_resumes_across_sinks and test_corrupt_tail_raises_on_resume cover both paths. Code for _resume_chain() is fully specified. RESOLVED.

Minor #3 (Hash chain opt-in default-off)
Decision 3 changes audit_hash_chain to default True. The plan explicitly states: "When AUDIT_LOG_PATH is set, hash-chain is on by default." The config.py change sets audit_hash_chain: bool = True. RESOLVED.

Minor #4 (test_executor.py line number claim)
Prior review found the line number claim (~line 60 for helpers) was wrong — helpers are at lines 18–53, class starts ~line 60. The R1 plan explicitly corrects this: "helpers are at lines 18–53, class TestProvisioning starts ~line 60. The new test is placed at module level (or in a new sibling class TestAuditTrail)." This matches the explorer verification (line 18: _make_spec, line 42: _make_mock_client, line 61: class TestProvisioning). RESOLVED.

Minor #5 (--dry-run flag unverified)
Decision in Changes from Review item 5 states: "Verified by grep at cli.py:128-131 (typer.Option("--dry-run/--no-dry-run", ...))." Explorer confirmed this is accurate. RESOLVED.

Minor #6 (state.started_at / completed_at unverified)
Changes from Review item 6 states: "Verified at models.py:180-181 — both fields declared str | None = None." Explorer confirmed this is accurate. The _duration helper now carries an inline citation. RESOLVED.

Minor #7 (P3/TDD ordering)
Decision 8 and Implementation Order explicitly restructure: step 2 writes tests/test_audit.py before step 3 implements audit.py, and step 4 writes TestAuditTrail before step 5 wires executor.py. RED-then-GREEN is explicit. RESOLVED.

Minor #8 (P7/POLA — async emit silently dropped)
Decision 5 resolves this: AuditSink.emit is now a sync method protected by threading.Lock. A missing await is impossible. RESOLVED.

Minor #9 (No import-failure rollback)
Decision 6 adds NullSink and build_sink_from_settings() which returns NullSink on any construction failure with a single warning log. WorkflowExecutor.__init__ falls back to NullSink() on AuditChainError or OSError. RESOLVED.

Remaining / New Findings

Dimension 1 — Requirements Alignment

All four sub-requirements (who/when/what/outcome) are addressed. Hash chain is now on by default when the path is set — the default deployment satisfies the "tamper-evident" requirement. No dropped acceptance criteria. ✓

Dimension 2 — Plan Completeness

The plan now covers: design (8 decisions with rationale), implementation (all files named with concrete code), tests (7 unit tests + 1 integration test named), verification (8 concrete commands). The import-failure rollback strategy is now present (Decision 6, NullSink fallback). ✓

One residual question: the Learnings section notes build_sink_from_settings() is late-imported inside cli.py:187, but Decision 6 says the executor's __init__ imports from telemachy.audit at module load — this is the protective design. The cli.py late import is cosmetic. Not a blocking issue.

Dimension 3 — Concreteness

All paths, class names, functions, and line numbers are verified accurate. The _make_spec / _make_mock_client line numbers are now correct. ✓

Dimension 4 — Risk Surface

The prior two major findings are resolved. New concerns from the Learnings section:

(a) os.fsync() per emit performance: The Learnings note this could add seconds to workflow execution. The plan does not address this — it's a known performance concern that is undocumented in the plan. This is a minor risk: the plan doesn't lie about it (it's in Learnings), but no mitigation or measurement is proposed. Minor finding.

(b) threading.Lock in async context: The Learnings note this is safe in single-threaded asyncio but could deadlock with thread pools. The plan justifies it in Decision 5: "The lock guards asyncio.gather fan-outs (single-threaded but reentrant via coroutines) and any future thread-pool callers." The async context is single-threaded asyncio, so the lock cannot deadlock there. This is acceptable justification. ✓

(c) Multi-process concurrent write: The Learnings note threading.Lock is per-sink, so two processes writing the same file are unprotected. The plan doesn't address this. Minor risk — the JSONL format is append-only, and multi-process writes to the same file are not a documented requirement. Minor finding, not blocking.

Dimension 5 — Verification Plan

8 concrete commands are provided, each mapped to a specific acceptance criterion. --dry-run is now verified. state.started_at / state.completed_at are verified. ✓

Dimension 6 — Stage Handoff

The plan is detailed enough that an implementer can follow it without re-deriving decisions. All key choices are made explicitly. The _duration helper is now grounded with a verified field reference. ✓

Software Engineering Principles

P1/KISS: The injection pattern is clean and matches the existing AgamemnonClient pattern. NullSink is a simple drop-in. No over-engineering. ✓

P2/YAGNI: AuditSinkProtocol is a Protocol class that is directly consumed by the sink parameter — it has a concrete current consumer in WorkflowExecutor. NullSink is consumed by the fallback in build_sink_from_settings(). No purely hypothetical abstractions. ✓

P3/TDD: Implementation Order now explicitly sequences tests before production code (step 2 → test_audit.py, step 3 → audit.py; step 4 → TestAuditTrail, step 5 → executor.py wiring). RED-then-GREEN is stated. ✓

P4/DRY: actor resolved once in AuditSink.__init__. No duplication introduced. ✓

P5/SOLID:

• SRP: AuditSink handles file I/O, hash chain, and record assembly. Three concerns in one class is a minor SRP observation — but for a focused 100-line module, this is acceptable and expected. Not a blocking concern.
• DIP: Resolved — WorkflowExecutor now depends on AuditSinkProtocol (an abstraction), not a concrete module-global. NullSink and AuditSink both implement the protocol. ✓
• Other sub-principles are not implicated here.

P6/Modularity: The singleton coupling is resolved. The injection seam (sink: AuditSinkProtocol | None = None) creates a clean module boundary. Tests inject directly; CLI builds once and passes in. ✓

P7/POLA: emit is now sync — no silent drop if await is forgotten. Sync method with sync lock is the expected pattern. ✓

Summary Assessment

All 9 prior findings (2 major, 7 minor) from R0 are concretely resolved in the R1 plan — not merely acknowledged. The plan is grounded in verified file/line readings. The remaining minor concerns (fsync performance, multi-process concurrent write, late import in cli.py) are all noted in the Learnings section of the plan itself, are not blocking, and do not represent unaddressed prior findings. The AuditSinkProtocol is used immediately and is not speculative YAGNI. TDD ordering is explicit. Verification commands are complete and verified.

The plan is now implementable with no major design defects.

Grade: B
Verdict: GO

---

Generated by Claude Code Planner (strict review loop)

[mvillmow]

🔍 Plan Review

Good — all factual claims verified accurate. Now let me analyze the R1 plan systematically against the prior review's findings.

R1 Plan Review — Issue #187: Audit Trail

Verifying Prior Iteration Findings Were Addressed

Major #1 (P5/DIP + P6/Modularity — singleton coupling)
The prior review found _sink = _AuditSink() at module import time as a process-global singleton. The R1 plan resolves this explicitly in Decision 1: AuditSink has no module-level instance; WorkflowExecutor.__init__ takes sink: AuditSinkProtocol | None = None; when omitted, build_sink_from_settings() is called inside __init__. The plan confirms tests inject AuditSink(path=tmp_path/"audit.jsonl", ...) directly. The existing injection pattern for AgamemnonClient (executor.py:31-39, cli.py:187) is cited as a model. This is a genuine structural fix, not an acknowledgment-without-change. RESOLVED.

Major #2 (Hash chain continuity gap)
The prior review flagged process restart resetting _prev_hash to all-zeros. Decision 3 in R1 resolves this: AuditSink.__init__ calls _resume_chain() which reads the last JSON line and seeds _prev_hash from its hash field. A corrupt or hash-less tail raises AuditChainError rather than silently zero-restarting. test_hash_chain_resumes_across_sinks and test_corrupt_tail_raises_on_resume cover both paths. Code for _resume_chain() is fully specified. RESOLVED.

Minor #3 (Hash chain opt-in default-off)
Decision 3 changes audit_hash_chain to default True. The plan explicitly states: "When AUDIT_LOG_PATH is set, hash-chain is on by default." The config.py change sets audit_hash_chain: bool = True. RESOLVED.

Minor #4 (test_executor.py line number claim)
Prior review found the line number claim (~line 60 for helpers) was wrong — helpers are at lines 18–53, class starts ~line 60. The R1 plan explicitly corrects this: "helpers are at lines 18–53, class TestProvisioning starts ~line 60. The new test is placed at module level (or in a new sibling class TestAuditTrail)." This matches the explorer verification (line 18: _make_spec, line 42: _make_mock_client, line 61: class TestProvisioning). RESOLVED.

Minor #5 (--dry-run flag unverified)
Decision in Changes from Review item 5 states: "Verified by grep at cli.py:128-131 (typer.Option("--dry-run/--no-dry-run", ...))." Explorer confirmed this is accurate. RESOLVED.

Minor #6 (state.started_at / completed_at unverified)
Changes from Review item 6 states: "Verified at models.py:180-181 — both fields declared str | None = None." Explorer confirmed this is accurate. The _duration helper now carries an inline citation. RESOLVED.

Minor #7 (P3/TDD ordering)
Decision 8 and Implementation Order explicitly restructure: step 2 writes tests/test_audit.py before step 3 implements audit.py, and step 4 writes TestAuditTrail before step 5 wires executor.py. RED-then-GREEN is explicit. RESOLVED.

Minor #8 (P7/POLA — async emit silently dropped)
Decision 5 resolves this: AuditSink.emit is now a sync method protected by threading.Lock. A missing await is impossible. RESOLVED.

Minor #9 (No import-failure rollback)
Decision 6 adds NullSink and build_sink_from_settings() which returns NullSink on any construction failure with a single warning log. WorkflowExecutor.__init__ falls back to NullSink() on AuditChainError or OSError. RESOLVED.

Remaining / New Findings

Dimension 1 — Requirements Alignment

All four sub-requirements (who/when/what/outcome) are addressed. Hash chain is now on by default when the path is set — the default deployment satisfies the "tamper-evident" requirement. No dropped acceptance criteria. ✓

Dimension 2 — Plan Completeness

The plan now covers: design (8 decisions with rationale), implementation (all files named with concrete code), tests (7 unit tests + 1 integration test named), verification (8 concrete commands). The import-failure rollback strategy is now present (Decision 6, NullSink fallback). ✓

One residual question: the Learnings section notes build_sink_from_settings() is late-imported inside cli.py:187, but Decision 6 says the executor's __init__ imports from telemachy.audit at module load — this is the protective design. The cli.py late import is cosmetic. Not a blocking issue.

Dimension 3 — Concreteness

All paths, class names, functions, and line numbers are verified accurate. The _make_spec / _make_mock_client line numbers are now correct. ✓

Dimension 4 — Risk Surface

The prior two major findings are resolved. New concerns from the Learnings section:

(a) os.fsync() per emit performance: The Learnings note this could add seconds to workflow execution. The plan does not address this — it's a known performance concern that is undocumented in the plan. This is a minor risk: the plan doesn't lie about it (it's in Learnings), but no mitigation or measurement is proposed. Minor finding.

(b) threading.Lock in async context: The Learnings note this is safe in single-threaded asyncio but could deadlock with thread pools. The plan justifies it in Decision 5: "The lock guards asyncio.gather fan-outs (single-threaded but reentrant via coroutines) and any future thread-pool callers." The async context is single-threaded asyncio, so the lock cannot deadlock there. This is acceptable justification. ✓

(c) Multi-process concurrent write: The Learnings note threading.Lock is per-sink, so two processes writing the same file are unprotected. The plan doesn't address this. Minor risk — the JSONL format is append-only, and multi-process writes to the same file are not a documented requirement. Minor finding, not blocking.

Dimension 5 — Verification Plan

8 concrete commands are provided, each mapped to a specific acceptance criterion. --dry-run is now verified. state.started_at / state.completed_at are verified. ✓

Dimension 6 — Stage Handoff

The plan is detailed enough that an implementer can follow it without re-deriving decisions. All key choices are made explicitly. The _duration helper is now grounded with a verified field reference. ✓

Software Engineering Principles

P1/KISS: The injection pattern is clean and matches the existing AgamemnonClient pattern. NullSink is a simple drop-in. No over-engineering. ✓

P2/YAGNI: AuditSinkProtocol is a Protocol class that is directly consumed by the sink parameter — it has a concrete current consumer in WorkflowExecutor. NullSink is consumed by the fallback in build_sink_from_settings(). No purely hypothetical abstractions. ✓

P3/TDD: Implementation Order now explicitly sequences tests before production code (step 2 → test_audit.py, step 3 → audit.py; step 4 → TestAuditTrail, step 5 → executor.py wiring). RED-then-GREEN is stated. ✓

P4/DRY: actor resolved once in AuditSink.__init__. No duplication introduced. ✓

P5/SOLID:

• SRP: AuditSink handles file I/O, hash chain, and record assembly. Three concerns in one class is a minor SRP observation — but for a focused 100-line module, this is acceptable and expected. Not a blocking concern.
• DIP: Resolved — WorkflowExecutor now depends on AuditSinkProtocol (an abstraction), not a concrete module-global. NullSink and AuditSink both implement the protocol. ✓
• Other sub-principles are not implicated here.

P6/Modularity: The singleton coupling is resolved. The injection seam (sink: AuditSinkProtocol | None = None) creates a clean module boundary. Tests inject directly; CLI builds once and passes in. ✓

P7/POLA: emit is now sync — no silent drop if await is forgotten. Sync method with sync lock is the expected pattern. ✓

Summary Assessment

All 9 prior findings (2 major, 7 minor) from R0 are concretely resolved in the R1 plan — not merely acknowledged. The plan is grounded in verified file/line readings. The remaining minor concerns (fsync performance, multi-process concurrent write, late import in cli.py) are all noted in the Learnings section of the plan itself, are not blocking, and do not represent unaddressed prior findings. The AuditSinkProtocol is used immediately and is not speculative YAGNI. TDD ordering is explicit. Verification commands are complete and verified.

The plan is now implementable with no major design defects.

Grade: B
Verdict: GO