PR D: typed + signed event bus + events table + SSE (arch step 5) #8

2026-04-22T09:38:12-07:00

nsatoshi commented

2026-04-22 09:38:12 -07:00

Implements step 5 from the architecture gap-analysis — replaces the in-memory EventEmitter pattern with a typed, signed, hash-chained, persisted event journal + SSE endpoint. Stacks on PR B.

What lands

`db/migrations/003_events.ts` — append-only journal

column	type	note
id	UUID	pk
plan_id	UUID	FK → plans(plan_id), cascade delete
type	VARCHAR(128)	one of arch §7.2
actor	VARCHAR(255)	nullable (system events)
payload	JSONB	arbitrary event body
payload_hash	CHAR(64)	sha256(JSON(payload))
prev_hash	CHAR(64)	signature of previous event for same plan
signature	CHAR(64)	hmac_sha256(secret, plan_id\|type\|payload_hash\|prev_hash)
created_at	TIMESTAMPTZ	default now()

Indexed on (plan_id, created_at) and (type).

`services/eventBus.ts`

EVENT_TYPES — all 15 normalised categories from arch §7.2. Single source of truth for subscribers.
publish({ planId, type, actor?, payload? }) — hashes the payload, reads the last signature for the plan, signs the new event, INSERTs, emits to the in-process EventEmitter.
verifyChain(planId) — replays the chain and returns { ok: true } or { ok: false, brokenAt, reason } on the first payload / signature / prev_hash mismatch.
getEventsForPlan(planId) — chronological replay.
subscribe(planId, cb) — returns an unsubscribe fn. Used by the SSE route.

HMAC secret precedence: EVENT_BUS_HMAC_SECRET → SESSION_SECRET → dev default.

`api/plans.ts`

GET /api/plans/:planId/events — full trail. ?verify=1 runs verifyChain and returns { chain_valid, broken_at?, broken_reason? } so UIs can flag tampered audit logs.
GET /api/plans/:planId/events/stream — SSE. On connect we replay history then subscribe to live events; 15s keep-alive; clean unsubscribe on client disconnect.

Tests

tests/unit/eventBus.test.ts — 9 cases: enum completeness, first-event null prev_hash, multi-event chain, per-plan isolation, untampered chain passes, three tamper-detection cases (payload / signature / prev_hash mutation).

Verification

$ npx tsc --noEmit     # clean
$ npx jest             # 60 passed, 5 suites

Not in this PR

ExecutionCoordinator still uses its own in-process EventEmitter for status events. PR A's state machine already records source_event_id on every transition; PR E (next) will wire ExecutionCoordinator to eventBus.publish() so every §7.2 event flows through the signed journal and every state transition can be joined back to its source event.
The in-process subscriber does not fan out across replicas — for multi-replica orchestrators the publish() call would additionally broadcast to NATS/Kafka, but the persistence layer and hash chain stay unchanged.

Series order

A → B → C → D → E → F → G → H.

Base: devin/1776875351-validating-exception-manager (PR B). The diff shown here is D-only.

Implements **step 5** from the architecture gap-analysis — replaces the in-memory `EventEmitter` pattern with a typed, signed, hash-chained, persisted event journal + SSE endpoint. Stacks on PR B. ## What lands ### `db/migrations/003_events.ts` — append-only journal | column | type | note | | --- | --- | --- | | id | UUID | pk | | plan_id | UUID | FK → plans(plan_id), cascade delete | | type | VARCHAR(128) | one of arch §7.2 | | actor | VARCHAR(255) | nullable (system events) | | payload | JSONB | arbitrary event body | | payload_hash | CHAR(64) | sha256(JSON(payload)) | | prev_hash | CHAR(64) | signature of previous event for same plan | | signature | CHAR(64) | hmac_sha256(secret, plan_id\|type\|payload_hash\|prev_hash) | | created_at | TIMESTAMPTZ | default now() | Indexed on `(plan_id, created_at)` and `(type)`. ### `services/eventBus.ts` - `EVENT_TYPES` — **all 15** normalised categories from arch §7.2. Single source of truth for subscribers. - `publish({ planId, type, actor?, payload? })` — hashes the payload, reads the last signature for the plan, signs the new event, INSERTs, emits to the in-process `EventEmitter`. - `verifyChain(planId)` — replays the chain and returns `{ ok: true }` or `{ ok: false, brokenAt, reason }` on the first payload / signature / prev_hash mismatch. - `getEventsForPlan(planId)` — chronological replay. - `subscribe(planId, cb)` — returns an unsubscribe fn. Used by the SSE route. HMAC secret precedence: `EVENT_BUS_HMAC_SECRET` → `SESSION_SECRET` → dev default. ### `api/plans.ts` - `GET /api/plans/:planId/events` — full trail. `?verify=1` runs `verifyChain` and returns `{ chain_valid, broken_at?, broken_reason? }` so UIs can flag tampered audit logs. - `GET /api/plans/:planId/events/stream` — SSE. On connect we replay history then subscribe to live events; 15s keep-alive; clean unsubscribe on client disconnect. ### Tests `tests/unit/eventBus.test.ts` — 9 cases: enum completeness, first-event null prev_hash, multi-event chain, per-plan isolation, untampered chain passes, three tamper-detection cases (payload / signature / prev_hash mutation). ## Verification ``` $ npx tsc --noEmit # clean $ npx jest # 60 passed, 5 suites ``` ## Not in this PR - ExecutionCoordinator still uses its own in-process `EventEmitter` for `status` events. PR A's state machine already records `source_event_id` on every transition; PR E (next) will wire `ExecutionCoordinator` to `eventBus.publish()` so every §7.2 event flows through the signed journal and every state transition can be joined back to its source event. - The in-process subscriber does not fan out across replicas — for multi-replica orchestrators the `publish()` call would additionally broadcast to NATS/Kafka, but the persistence layer and hash chain stay unchanged. ## Series order A → B → C → **D** → E → F → G → H. Base: `devin/1776875351-validating-exception-manager` (PR B). The diff shown here is D-only.

nsatoshi changed target branch from devin/1776875351-validating-exception-manager to main

2026-04-22 10:16:07 -07:00

nsatoshi force-pushed devin/1776875718-event-bus-sse from 18bdaf61d5 to 59e1a85267

2026-04-22 10:17:21 -07:00

Compare

nsatoshi merged commit cb376eda31 into main

2026-04-22 10:17:41 -07:00

nsatoshi referenced this issue from a commit

2026-04-22 10:17:42 -07:00

PR D: typed + signed event bus + events table + SSE (arch step 5) (#8)

nsatoshi deleted branch devin/1776875718-event-bus-sse

2026-04-22 10:17:44 -07:00

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: d-bis/CurrenciCombo#8