zbeam-pipeline-auditor

name: zbeam-pipeline-auditor description: "Manual pipeline health check — loop coherence, pipeline logic, corpus scan, manifest integrity, gap-signals refresh. Run before any bulk content session."

Z-Beam Pipeline Auditor

Invoke before any bulk content session — ensures the pipeline is coherent, the manifest matches disk, and gap-signals are fresh before generation-loop work begins.

What this skill does:

• Validates loop architecture coherence (Step 0b)
• Validates skill handoff contracts (Step 0c)
• Checks manifest vs filesystem (Step 0e)
• Runs structural corpus scan → improvement backlog (Step 1b)
• Runs dedup scan → improvement backlog (Step 1c)
• Refreshes gap-signals.json for batch work (Step 4)
• Updates campaignFocus in campaign.json (Step 4)

What was removed (2026-06-26): Steps 1 and 2 called zbeam-signal-monitor and zbeam-intelligence-diff — both deleted. Step 5 (compile-signals.py) depended on their output and is also removed. Adaptive learning is now zbeam-quality-correlator + evalScore-correlator.py. Auto-apply patches is zbeam-improvement-loop (manual).

Todd's approval required before any skill or threshold change.

Step 0: Load protected files config

Read skills/visibility/zbeam-pipeline-auditor/references/protected-files.json before any auto-apply step. This file defines which file paths and patch types always require Todd's explicit approval regardless of confidence level.

import json
protected = json.load(open('skills/visibility/zbeam-pipeline-auditor/references/protected-files.json'))
protected_paths = protected.get('protectedFilePaths', [])
protected_patch_types = protected.get('protectedPatchTypes', [])
# Any proposed edit whose target path matches protected_paths, or whose type matches
# protected_patch_types, is routed to human-review — never auto-applied.

Step 0b: Load session state

Read references/startup-checks.md (Step 0b section). Loads data/pipeline/session-state.json, clears stale in-flight slugs, and runs the oscillation check.

Step 0c: Pipeline logic check

Run immediately after the loop coherence check. Audits the pipeline's own logic for handoff contract gaps, broken feedback loops, and quality gate leakage.

python3 skills/shared/check-pipeline-logic.py --json

Read data/audit/pipeline-logic-[today].json:

logic_path = f'data/audit/pipeline-logic-{date.today().isoformat()}.json'
logic = json.load(open(logic_path)) if os.path.exists(logic_path) else {}
logic_high = logic.get('highGapCount', 0)

if logic_high > 0:
    print(f'⚠ PIPELINE LOGIC GAPS — {logic_high} HIGH gap(s). Auto-apply blocked this run.')
    for g in logic.get('gaps', []):
        if g['severity'] == 'HIGH':
            print(f'  [{g["check"]}] {g["description"]}')
    block_auto_apply = True  # merged with coherence block flag
else:
    print(f'✓ Pipeline logic PASS')

MEDIUM gaps are logged but do not block auto-apply.

Step 0b: Loop coherence check

Run before any other pipeline work. If violations are found, log them and block auto-apply for this run (proposals are still generated, but none are applied autonomously).

python3 skills/shared/check-loop-coherence.py --json

Read data/audit/loop-coherence-[today].json:

import json, os
from datetime import date

coherence_path = f'data/audit/loop-coherence-{date.today().isoformat()}.json'
coherence = json.load(open(coherence_path)) if os.path.exists(coherence_path) else {}
coherence_status = coherence.get('status', 'UNKNOWN')
coherence_violations = coherence.get('violations', [])

if coherence_status == 'FAIL':
    print(f'⚠ LOOP COHERENCE FAIL — {len(coherence_violations)} violation(s). Auto-apply blocked this run.')
    for v in coherence_violations:
        print(f'  [{v["rule"]}] {v["description"]} — {v["path"]}')
    # Set flag to block Step 3 auto-apply
    block_auto_apply = True
else:
    print(f'✓ Loop coherence PASS ({coherence.get("passCount", 0)} rules checked)')
    block_auto_apply = False

In Step 3, check block_auto_apply before running auto-apply.py — if True, skip auto-apply and add coherence violations to the needs_approval list in the review queue instead.

Coherence failures are reported in the weekly audit report under loopCoherence.

Step 0d: llms.txt freshness check

Read references/startup-checks.md (Step 0d section). Counts frontmatter files and compares to declared counts in public/llms.txt. Reports stale entries — never auto-writes the file.

Step 0e: Governance — manifest vs reality check

Read references/startup-checks.md (Step 0e section). Verifies every sharedScripts entry and dataDirs entry in manifest.json exists on disk; flags unregistered files in skills/shared/. MISSING FILE items block auto-apply.

Steps 1b + 1c: Corpus structural scan + dedup scan

Read references/corpus-scan-protocol.md. Run both scans to populate the improvement backlog and set block_auto_apply on any FAIL-severity findings.

Plan

Print all proposals, confidence levels, and files they touch, then proceed immediately — no approval required.

Step 2: Auto-apply high-confidence patches

Read skills/visibility/zbeam-pipeline-auditor/references/auto-apply.py and execute it.

Requires proposals (from Step 2) and today_str = date.today().isoformat(). Produces auto_applied and needs_approval lists; writes review queue to data/review-queue/[date]-summary.md and updates data/audit/improvement-backlog.json.

Adversarial challenges (implemented in auto-apply.py):

• Conflict with a patch applied in the last 7 days → escalate to approval
• Safety gate removal (removes if len( / assert / threshold checks) → escalate
• Numeric threshold increase >25% → escalate

Step 4: Update campaignFocus

Read references/campaign-focus.md and execute the script. Updates data/marketing/campaign.json with fresh intelligence signals and uncovered content gaps from gap-signals.

Step 5: Refresh gap-signals

python3 skills/shared/gap-signals-generator.py

Produces data/audit/gap-signals-[date].json — required by zbeam-weekly-batch trigger gate (must be < 24h old) and by zbeam-gap-researcher. Run this before any manual batch session.

Step 6: Weekly Owner Review Questions

Surface these questions at the end of every pipeline-auditor run. They are not automated checks — they require a human answer before the session's batch work begins.

1. What moved this week and why? — find the most recent GSC daily snapshot, compare to the snapshot 7 days prior, identify the 3 largest impression changes. For each: is there a change-log hypothesis that predicted this? If yes, resolve it. If no, note it as an unexplained signal.

2. Is there a recorded hypothesis for every item the pipeline is proposing to change? Check data/audit/gap-signals-[latest].json → queue. For each slug: does data/audit/change-log.json have an open entry? If any queued slug has no hypothesis on record, surface it — that slug should not be processed until a hypothesis is written.

These questions are from skills/PIPELINE_CONSTITUTION.md §Weekly. They are the human gate the Constitution requires but cannot enforce automatically.

Steps 7–8 + constraints + output format

Read references/execution-steps.md. Covers: data retention cleanup (Step 7), session state write (Step 8), what this skill never does, and gap-signals output format.