Pipeline observability: quota events, wall-clock timing, drift fix

[Trecek]

Summary

Adds three pipeline observability capabilities: a new get_quota_events MCP tool surfacing quota guard decisions from quota_events.jsonl, wall_clock_seconds merged into get_token_summary output for per-step wall-clock visibility, and a .telemetry_cleared_at replay fence preventing token accounting drift when the MCP server restarts after a clear=True call. Includes a follow-up refactor extracting _get_log_root() in tools_status.py to eliminate three identical inline log-root expressions.

Individual Plan Details

Group 1: Pipeline Observability — Quota Guard Logging and Per-Step Elapsed Time

Three related pipeline observability improvements, tracked as GitHub issue #302 (collapsing #218, #65, and the #304/#148 token accounting item):

1. Quota guard MCP tool (feat: Add quota guard observability to diagnostic logging system #218): The quota_check.py hook already writes quota_events.jsonl with approved/blocked/cache-miss events. Add a new ungated get_quota_events tool to tools_status.py to surface those decisions through the MCP API.

2. Wall-clock time in token summary (feat: report per-step elapsed time in token summary #65): Merge total_seconds from the timing log into each step's get_token_summary output as wall_clock_seconds, so operators see wall-clock duration alongside token counts in one call. Updates _format_token_summary and write_telemetry_files.

3. Token accounting drift fix (Combined: Pre-release readiness — stability fixes #304/Stability and correctness fixes for public release #148): Persist a .telemetry_cleared_at timestamp when any log is cleared. _state._initialize reads this on startup and uses max(now - 24h, marker_ts) as the effective replay lower bound, excluding already-cleared sessions.

Group 2: Remediation — Extract _get_log_root() helper in tools_status.py

The audit identified that tools_status.py had three identical inline expressions — resolve_log_dir(_get_ctx().config.linux_tracing.log_dir) — repeated in get_pipeline_report, get_token_summary, and get_timing_summary. This remediation adds _get_log_root() to centralize that computation and replaces all three inline call sites. No behavioral change.

Architecture Impact

Operational Diagram

%%{init: {'flowchart': {'nodeSpacing': 50, 'rankSpacing': 65, 'curve': 'basis'}}}%%
flowchart TB
    %% CLASS DEFINITIONS %%
    classDef cli fill:#1a237e,stroke:#7986cb,stroke-width:2px,color:#fff;
    classDef stateNode fill:#004d40,stroke:#4db6ac,stroke-width:2px,color:#fff;
    classDef handler fill:#e65100,stroke:#ffb74d,stroke-width:2px,color:#fff;
    classDef phase fill:#6a1b9a,stroke:#ba68c8,stroke-width:2px,color:#fff;
    classDef newComponent fill:#2e7d32,stroke:#81c784,stroke-width:2px,color:#fff;
    classDef output fill:#00695c,stroke:#4db6ac,stroke-width:2px,color:#fff;
    classDef detector fill:#b71c1c,stroke:#ef5350,stroke-width:2px,color:#fff;

    subgraph UngatedTools ["UNGATED MCP TOOLS (tools_status.py)"]
        GTS["● get_token_summary<br/>━━━━━━━━━━<br/>clear=False<br/>+ ● wall_clock_seconds<br/>from timing_log"]
        GTIM["● get_timing_summary<br/>━━━━━━━━━━<br/>clear=False<br/>total_seconds per step"]
        GPR["● get_pipeline_report<br/>━━━━━━━━━━<br/>clear=False<br/>audit failures"]
        GQE["★ get_quota_events<br/>━━━━━━━━━━<br/>n=50<br/>quota guard decisions"]
    end

    subgraph LogRoot ["★ _get_log_root() helper (tools_status.py)"]
        LR["★ _get_log_root()<br/>━━━━━━━━━━<br/>resolve_log_dir(ctx.config<br/>.linux_tracing.log_dir)"]
    end

    subgraph InMemory ["IN-MEMORY PIPELINE LOGS"]
        TK["token_log<br/>━━━━━━━━━━<br/>step_name → tokens<br/>elapsed_seconds"]
        TI["timing_log<br/>━━━━━━━━━━<br/>step_name → total_seconds<br/>(monotonic clock)"]
        AU["audit_log<br/>━━━━━━━━━━<br/>list FailureRecord"]
    end

    subgraph DiskLogs ["PERSISTENT LOG FILES (~/.local/share/autoskillit/logs/)"]
        QE["quota_events.jsonl<br/>━━━━━━━━━━<br/>approved / blocked<br/>cache_miss / parse_error"]
        CM["★ .telemetry_cleared_at<br/>━━━━━━━━━━<br/>UTC ISO timestamp fence<br/>written on clear=True"]
    end

    subgraph Startup ["SERVER STARTUP (_state._initialize)"]
        INIT["● _state._initialize<br/>━━━━━━━━━━<br/>since = max(now−24h, marker)<br/>load_from_log_dir × 3"]
    end

    subgraph Hook ["HOOK (quota_check.py)"]
        QH["quota_check.py<br/>━━━━━━━━━━<br/>PreToolUse: approve/block<br/>_write_quota_log_event"]
    end

    GTS -->|"clear=True"| LR
    GTIM -->|"clear=True"| LR
    GPR -->|"clear=True"| LR
    LR -->|"write_telemetry_clear_marker"| CM
    GQE -->|"_read_quota_events(n)"| QE
    QH -->|"append event"| QE
    GTS -->|"get_report()"| TK
    GTS -->|"● merge wall_clock_seconds"| TI
    GTIM -->|"get_report()"| TI
    GPR -->|"get_report()"| AU
    CM -->|"read marker → since bound"| INIT
    INIT -->|"load_from_log_dir since=effective"| TK
    INIT -->|"load_from_log_dir since=effective"| TI
    INIT -->|"load_from_log_dir since=effective"| AU

    class GTS,GTIM,GPR cli;
    class GQE newComponent;
    class LR newComponent;
    class TK,TI,AU stateNode;
    class QE stateNode;
    class CM newComponent;
    class INIT phase;
    class QH detector;

Loading

Color Legend: Dark Blue = MCP query tools | Green = New components (get_quota_events, _get_log_root, .telemetry_cleared_at) | Teal = State/logs | Purple = Server startup | Dark Red = PreToolUse hook

State Lifecycle Diagram

%%{init: {'flowchart': {'nodeSpacing': 50, 'rankSpacing': 65, 'curve': 'basis'}}}%%
flowchart TB
    %% CLASS DEFINITIONS %%
    classDef cli fill:#1a237e,stroke:#7986cb,stroke-width:2px,color:#fff;
    classDef stateNode fill:#004d40,stroke:#4db6ac,stroke-width:2px,color:#fff;
    classDef handler fill:#e65100,stroke:#ffb74d,stroke-width:2px,color:#fff;
    classDef phase fill:#6a1b9a,stroke:#ba68c8,stroke-width:2px,color:#fff;
    classDef newComponent fill:#2e7d32,stroke:#81c784,stroke-width:2px,color:#fff;
    classDef output fill:#00695c,stroke:#4db6ac,stroke-width:2px,color:#fff;
    classDef detector fill:#b71c1c,stroke:#ef5350,stroke-width:2px,color:#fff;
    classDef gap fill:#ff6f00,stroke:#ffa726,stroke-width:2px,color:#000;

    subgraph InMem ["IN-MEMORY (clearable — MUTABLE)"]
        TK["● token_log<br/>━━━━━━━━━━<br/>MUTABLE<br/>step_name → tokens + elapsed_seconds<br/>cleared on clear=True"]
        TI["timing_log<br/>━━━━━━━━━━<br/>MUTABLE<br/>step_name → total_seconds<br/>cleared on clear=True"]
        AU["audit_log<br/>━━━━━━━━━━<br/>MUTABLE<br/>list FailureRecord<br/>cleared on clear=True"]
    end

    subgraph Derived ["DERIVED (computed per query)"]
        WC["★ wall_clock_seconds<br/>━━━━━━━━━━<br/>DERIVED<br/>timing_log.get_report()<br/>merged into token summary response<br/>never persisted"]
    end

    subgraph ClearFence ["★ CLEAR FENCE (write-then-read across restarts)"]
        CM["★ .telemetry_cleared_at<br/>━━━━━━━━━━<br/>WRITE-FENCE<br/>UTC ISO timestamp<br/>written atomically by write_telemetry_clear_marker<br/>read exactly once by _initialize"]
    end

    subgraph DiskReplay ["DISK REPLAY (bounded by clear fence)"]
        SJ["sessions.jsonl + session/<br/>━━━━━━━━━━<br/>REPLAY-SOURCE<br/>historical token + timing + audit data<br/>replayed with since= lower bound"]
        QE["quota_events.jsonl<br/>━━━━━━━━━━<br/>APPEND-ONLY<br/>quota hook writes, never rewrites<br/>read by ★ get_quota_events"]
    end

    subgraph ClearGate ["CLEAR GATE (state mutation trigger)"]
        ClearTrue["clear=True in<br/>● get_token_summary /<br/>● get_timing_summary /<br/>● get_pipeline_report<br/>━━━━━━━━━━<br/>1. Clear in-memory log<br/>2. ★ Write .telemetry_cleared_at"]
    end

    subgraph StartupGate ["★ STARTUP REPLAY GATE (_state._initialize)"]
        INIT["● _state._initialize<br/>━━━━━━━━━━<br/>1. Read .telemetry_cleared_at<br/>2. since = max(now−24h, marker)<br/>3. load_from_log_dir × 3<br/>Guards: no double-counting"]
    end

    ClearTrue -->|"1. in_memory.clear()"| TK
    ClearTrue -->|"1. in_memory.clear()"| TI
    ClearTrue -->|"1. in_memory.clear()"| AU
    ClearTrue -->|"2. write_telemetry_clear_marker()"| CM
    TI -->|"get_report() per query"| WC
    WC -->|"★ merged into response"| TK
    CM -->|"read → since bound"| INIT
    SJ -->|"load_from_log_dir since=effective"| INIT
    INIT -->|"populate (bounded)"| TK
    INIT -->|"populate (bounded)"| TI
    INIT -->|"populate (bounded)"| AU

    class TK,TI,AU handler;
    class WC newComponent;
    class CM newComponent;
    class ClearTrue detector;
    class INIT phase;
    class SJ,QE stateNode;

Loading

Color Legend: Orange = MUTABLE in-memory logs | Green = New (wall_clock_seconds, .telemetry_cleared_at fence) | Dark Red = clear=True trigger | Purple = startup replay gate | Teal = persistent disk state

Module Dependency Diagram

%%{init: {'flowchart': {'nodeSpacing': 50, 'rankSpacing': 70, 'curve': 'basis'}}}%%
graph TB
    %% CLASS DEFINITIONS %%
    classDef cli fill:#1a237e,stroke:#7986cb,stroke-width:2px,color:#fff;
    classDef phase fill:#6a1b9a,stroke:#ba68c8,stroke-width:2px,color:#fff;
    classDef handler fill:#e65100,stroke:#ffb74d,stroke-width:2px,color:#fff;
    classDef stateNode fill:#004d40,stroke:#4db6ac,stroke-width:2px,color:#fff;
    classDef newComponent fill:#2e7d32,stroke:#81c784,stroke-width:2px,color:#fff;
    classDef output fill:#00695c,stroke:#4db6ac,stroke-width:2px,color:#fff;
    classDef integration fill:#c62828,stroke:#ef9a9a,stroke-width:2px,color:#fff;

    subgraph L3 ["L3 — SERVER (tools_status.py, _state.py, helpers.py)"]
        direction LR
        TS["● tools_status.py<br/>━━━━━━━━━━<br/>★ _get_log_root()<br/>★ get_quota_events<br/>● 3 clear=True paths"]
        ST["● _state.py<br/>━━━━━━━━━━<br/>● _initialize<br/>reads clear marker"]
        HLP["● helpers.py<br/>━━━━━━━━━━<br/>re-exports<br/>write/read_telemetry_clear_marker<br/>resolve_log_dir"]
    end

    subgraph L1 ["L1 — EXECUTION (execution/__init__.py, session_log.py)"]
        direction LR
        EINIT["● execution/__init__.py<br/>━━━━━━━━━━<br/>★ exports write/read_telemetry_clear_marker<br/>public API surface"]
        SL["● session_log.py<br/>━━━━━━━━━━<br/>★ write_telemetry_clear_marker()<br/>★ read_telemetry_clear_marker()<br/>_CLEAR_MARKER_FILENAME"]
    end

    subgraph L0 ["L0 — CORE (core/types.py)"]
        TY["● core/types.py<br/>━━━━━━━━━━<br/>● UNGATED_TOOLS frozenset<br/>+ get_quota_events"]
    end

    TS -->|"import resolve_log_dir<br/>write/read_telemetry_clear_marker<br/>(via helpers shim)"| HLP
    ST -->|"★ import read_telemetry_clear_marker<br/>(direct from execution)"| EINIT
    HLP -->|"re-export from execution"| EINIT
    EINIT -->|"defined in"| SL
    TS -.->|"UNGATED_TOOLS check<br/>(via pipeline.gate)"| TY

    class TS,ST,HLP cli;
    class EINIT,SL handler;
    class TY stateNode;

Loading

Color Legend: Dark Blue = L3 server layer | Orange = L1 execution layer | Teal = L0 core types | Dashed = indirect (via pipeline.gate) | All imports flow downward (no violations)

Closes #302

Implementation Plans

Plan files:

• temp/make-plan/302_pipeline_observability_plan_2026-03-10_204500.md
• temp/make-plan/302_remediation_get_log_root_plan_2026-03-10_210500.md

🤖 Generated with Claude Code via AutoSkillit

[Trecek]

AutoSkillit PR Review — Verdict: changes_requested

[Trecek]

[warning] tests: test_limits_to_n_events does not assert total_count equals 10 (the full log size). Only checks len(result["events"]) == 3 but omits verifying total_count reflects the full dataset.

[Trecek]

[warning] tests: test_limits_to_n_events does not verify ordering of returned events (most-recent-first). The 10-event pagination test never checks which 3 events are returned.

[Trecek]

[warning] tests: test_wall_clock_falls_back_to_elapsed_when_no_timing never verifies timing_log has no step-b entry. Parallel test pollution could silently bypass the fallback path.

[Trecek]

[warning] tests: test_initialize_uses_clear_marker_as_since_bound: monkeypatching _state._ctx=None conflicts with tool_ctx fixture which already patches it. Interleaved None-then-reinit may leave _ctx in unexpected state under xdist teardown.

[Trecek]

[warning] tests: Same xdist/fixture-teardown concern as L661. Also missing boundary condition: marker timestamp == session timestamp (boundary for <= vs < in since_dt logic).

[Trecek]

[warning] slop: noqa comment claims read_telemetry_clear_marker is used by tools_status.py but it is not called anywhere in tools_status.py. The re-export and comment are misleading dead weight.

[Trecek]

[warning] defense: _merge_wall_clock_seconds parameter timing_log is typed as Any, bypassing static type checking. Should be typed as DefaultTimingLog or its protocol to enable mypy to catch misuse at call sites.

[Trecek]

[warning] fidelity: write_telemetry_clear_marker is called when get_timing_summary(clear=True) fires, advancing the shared fence even when token_log and audit are NOT cleared. On next restart, _state._initialize skips sessions for all three log types — may under-count token and audit data that was never cleared. Issue #302 does not describe this shared-fence side effect.

[Trecek]

AutoSkillit Review Findings

Verdict: changes_requested

8 actionable findings (all warning severity). Implementation is correct and addresses all three requirements from issue #302 (quota events tool, wall-clock seconds in token summary, drift prevention fence). Inline comments posted above.

src/autoskillit/server/helpers.py

• L14 [warning/slop]: noqa comment claims read_telemetry_clear_marker is 'used by tools_status.py' but it is not called anywhere in tools_status.py — unused re-export with misleading comment.

src/autoskillit/server/tools_status.py

• L37 [warning/defense]: _merge_wall_clock_seconds parameter timing_log typed as Any — bypasses static type checking; use DefaultTimingLog or its protocol.
• L191 [warning/fidelity]: Shared-fence side effect — get_timing_summary(clear=True) advances the fence for all three log types (token_log, timing_log, audit), not just timing_log. On next restart, _state._initialize may skip token/audit sessions that were never explicitly cleared. Not described in issue #302.

tests/server/test_tools_status.py

• L627 [warning/tests]: test_limits_to_n_events omits assert result["total_count"] == 10 — total_count goes untested for the n-limiting case.
• L636 [warning/tests]: test_limits_to_n_events never asserts which 3 events are returned (oldest or newest) — ordering unverified in the paginated case.
• L695 [warning/tests]: test_wall_clock_falls_back_to_elapsed_when_no_timing never asserts timing_log has no 'step-b' entry — parallel pollution could silently bypass the fallback.

tests/server/test_server_init.py

• L661 [warning/tests]: test_initialize_uses_clear_marker_as_since_bound uses bare _state._ctx = None assignment conflicting with fixture monkeypatch — xdist teardown ordering may leave _ctx in unexpected state.
• L705 [warning/tests]: Same xdist concern as L661; missing boundary condition test (marker ts == session ts).