Summary

KR-P2-I-integration ST2 of 5. Adds the listener that writes a chain event for every transition through OperationalStateHolder. Stacked on top of #32 (ST1) — base is feat/kora-KR-P2-I-integration-st1-holder. Retarget to main after ST1 merges.

Event vocabulary (PM-verified against substrate)

Verified against packages/db/migrations/foundation/0159_kora_r41_operational_state_event_vocabulary.sql on rafe-walker/isokron (bd165eb HEAD):

• Always emit kora.operational_state.transitioned — substrate-team's documented design: one generic edge event per transition; payload carries the from/to state, claim_permission, sorted degradation_reasons, and trigger.

• Additionally emit a per-trigger informational literal when one exists in the vocab:

  Transition	Trigger match	Extra literal
  BOOTING → READY	"all §9.2 gates pass" in trigger	kora.boot.ready
  BOOTING → STOPPED	"invariant gate failure" in trigger	kora.boot.failed
  * → PAUSED	"cost 100%" in trigger	kora.paused.cost_limit
  Operator-pause, substrate-pause	—	generic only (no vocab literal yet)

  BOOTING → STOPPED is intentionally disambiguated: STOP-KORA L4/L5 from BOOTING reaches the same arrow as invariant-gate-failure but is operator action, not a boot failure — kora.boot.failed only fires when the trigger string contains "invariant gate failure".

  Trigger matching is substring, so caller wording like "cost 100% — auto-pause" or "STOP-KORA L2: cost 100% triggered budget watcher" all land on kora.paused.cost_limit.

Failure mode — fail-LOUD

Mirrors agent/constitution_audit.py. emit_state_transition raises OperationalStateEmitError on:

• provider is None
• provider._connection is None
• _resolve_workspace_id() raises or returns empty
• The substrate-side kora__append_event call raises (Sea MCP unavailable, CHECK violation, chain lock failure, etc.)

The holder's listener-exception handler (ST1) catches and logs so one broken emit doesn't deadlock the state machine — but the raise + log is the observability path. No silent swallow.

agent/operational_state_emit.py (new, 273 LOC)

• Constants — GENERIC_TRANSITION_EVENT, BOOT_READY_EVENT, BOOT_FAILED_EVENT, PAUSED_COST_LIMIT_EVENT (named so the literal value is greppable from logs)
• _select_extra_literal(from, to, trigger) — returns the supplementary literal or None. Pure function — no I/O.
• _build_payload(from, to, trigger) — builds the payload dict. Pure function. Cockpit relies on stable sort for degradation_reasons so we sorted(...) rather than list(...).
• emit_state_transition(provider, from, to, trigger) — async. Always emits generic; conditionally emits extra. Raises on any failure.
• make_emit_listener(provider) — factory returning a StateTransitionListener bound to the provider. ST3 calls this once and holder.add_listener(...) registers it.

Tests — tests/test_operational_state_emit.py (399 LOC)

16 cases:

• Payload — shape matches spec, post-transition values (not pre)
• Per-trigger selection — boot.ready / boot.failed / paused.cost_limit positive paths; the operator/substrate negative paths; the BOOTING → STOPPED disambiguation; substring leniency across three trigger variants
• Fail-LOUD preflight — provider=None, missing connection, workspace_id raise, workspace_id empty
• Happy path — generic-only for ACTIVE-claim, generic+boot.ready for boot ready, both events carry identical payload
• Substrate failure — Sea MCP unavailable raises bubble as OperationalStateEmitError with the underlying cause in .cause
• Listener factory — make_emit_listener returns a usable listener

Tests use a stubbed IsoKron connection that runs the coro on the test's event loop and wraps in a concurrent.futures.Future so asyncio.wrap_future works — no real MCP transport in the unit tests. Integration test against a live provider lands in ST3.

§1 verifications — all green

• ✅ Event vocab — see foundation/0159 references in PR description and module docstring
• ✅ Skeleton API surface (ST1 feat(KR-P2-I-integration ST1): OperationalStateHolder singleton + transition broker #32 referenced)
• ✅ Endpoint shape (ST5 will preserve)

Sub-task chain

Test plan

• CI green on pytest tests/test_operational_state_emit.py
• No production wire-in yet — ST3 is the first real caller. If anything imports make_emit_listener outside this PR's tests, that's an early wire-in to flag

🤖 Generated with Claude Code