/event-log-and-reducer

name: event-log-and-reducer
description: The append-only event log + deterministic reducer pattern John uses to coordinate parallel subagent work on shared state. Each subagent emits its own event files; one reducer folds all events into canonical state. Beats file locks, scales to thousands of work units.
metadata:
  triggers:
    - event log
    - reduce events
    - coordinate subagents
    - shared state
    - reducer

event-log-and-reducer

When N subagents are working in parallel on shared state, the naive approach (each subagent writes to a shared catalog file with a lock) is what KC (a sibling verification harness) learned the hard way is fragile at scale. John uses **event log + reducer** instead — same shape that React/Redux and event-sourced systems use, ported to filesystem.

The pattern in one diagram

Subagent A ── events/extract/chunks/A-001.json (append-only, A's own file) ─┐
Subagent B ── events/extract/chunks/B-002.json                              │
Subagent C ── events/extract/chunks/C-003.json                              ├──► reducer ──► .john/checkpoints/extract/state.json
...                                                                         │              (canonical)
Subagent N ── events/extract/chunks/N-200.json                             ─┘

• Each subagent invokes `emit_event.py`; the writer assigns a unique filename and atomic envelope. Zero contention and retries never overwrite history.
• The reducer reads all event files for a phase, in deterministic order, and produces canonical state.
• Running the reducer twice yields the same output (idempotent).
• Canonical state is the read source for the next phase.

Where files live

In the user's project, under `<project>/.john/`:

• `<project>/.john/events/<phase-name>/<work-unit-type>/<subagent-id>-<sequence>.json` — events. One file per subagent emission.
• `<project>/.john/checkpoints/<phase-name>/state.json` — canonical reduced state. Written by reducer.

Conventions:

• Phase, work-unit, agent, and audit-run IDs use letters, digits, `_`, and `-`, beginning with a letter or digit.
• Event files are JSON; canonical state is JSON. Markdown if the canonical state is human-facing.
• Producers do not choose filenames. Pipe one JSON object through the shipped atomic writer:

printf '%s\n' '{"event_type":"chunk_complete","chunk_id":"chunk-042"}' | \
  python3 "${CLAUDE_PLUGIN_ROOT}/scripts/emit_event.py" \
    --phase extract --work-unit-id chunk-042 \
    --agent-id extractor-7 --audit-run-id run-20260709

The writer injects a UUID `event_id`, UTC `timestamp`, `agent_id`, and `audit_run_id`. Raw events are append-only.

Event shape — one valid approach

The only hard requirements (everything else is taste):

1. One event = one self-contained record. No references that require another event to interpret. 2. JSON-parseable. 3. Enough metadata for the reducer to order and deduplicate (timestamp + a sender id is the usual minimum).

A **minimal** schema that satisfies the rules:

{ "timestamp": "ISO 8601", "subagent_id": "string", "payload": {} }

A **richer** schema (used in many knowledge-extraction templates) — useful when you need to query the event log by type:

{
  "event_type": "entry_extracted",
  "work_unit_id": "chunk_042",
  "timestamp": "2026-05-21T10:42:33Z",
  "subagent_id": "sub-xx7f3a",
  "payload": { "entry_ids": ["e_001", "e_002"], "notes": "..." }
}

Templates and phase-specific skills define their own schemas freely, as long as the three requirements above hold. Neither shape above is canonical — wide tunnel.

How to write the reducer

The reducer is a script (Python — John ships `scripts/reduce_events.py`) that:

1. **Reads all event files** under `<project>/.john/events/<phase>/`. 2. **Sorts them deterministically** (timestamp + subagent_id is a safe primary key). At thousands-of-events scale, clock skew or identical timestamps happen — `${CLAUDE_PLUGIN_ROOT}/scripts/reduce_events.py` handles the tiebreaker. If your fold function depends on strict ordering, review the tiebreaker before trusting the result. 3. **Folds them into canonical state** using a per-phase fold function. The fold function's exact shape depends on what the phase is producing — for extraction, it concatenates entry lists and indexes by ID; for review, it tallies pass/fail; etc. 4. **Writes canonical state** to `<project>/.john/checkpoints/<phase>/state.json`. 5. **Returns idempotently**: running it twice with the same event set produces the same output, bit-for-bit.

Idempotency matters because the reducer may be invoked multiple times during a phase (e.g., after each wave of subagents) without state corruption.

Phase-boundary checks: count gate + disk reconciliation

`reduce_events.py` ships two deterministic checks for the end of a phase — zero tokens, pure file walking:

• **Count gate** — `--expect-entries N` or `--expect-entries MIN-MAX`. Counts unique entry ids claimed in the phase's events (`payload.entry_id` / `payload.entry_ids`, deduplicated, so corrective re-emits don't inflate it) against the expectation from PLAN.md. The *caller* supplies the number — the script never parses PLAN.md. Below ~90% of the minimum → **exit 3**: do not advance the phase. Small drift or overage → warning, exit 0. The checkpoint is still written on failure — the gate blocks *advancement*, not state derivation. Always prints actual-vs-expected so the number lands in the transcript.
• **Disk reconciliation** — `--verify-knowledge [--knowledge-dir PATH]`. Cross-checks knowledge entries on disk against claimed entry ids: **orphans** (on disk, no claiming event) and **missing-on-disk** (claimed, no entry dir). Strictly report-only — it warns and never mutates or deletes. The orphan policy is *warn, never fix*: a hand-added entry is legitimate; flag it, let a human decide. And missing-on-disk after the rewrite phase is often legitimate dedup, not corruption — the warning