gascity/engdocs/architecture/event-bus.md at main · gastownhall/gascity

title	Event Bus

Last verified against code: 2026-04-25

Summary

The Event Bus is Gas City's Layer 0-1 primitive providing an append-only pub/sub log of all system activity -- the universal observation substrate. Every state change in the system (agent started, bead created, order fired, controller lifecycle) is recorded as an immutable event with a monotonically increasing sequence number. The event bus enables infrastructure mechanisms like order trigger evaluation, CLI event tailing, and audit logging without coupling producers to consumers.

Key Concepts

Event: A single immutable record of something that happened. Struct with Seq (monotonically increasing uint64), Type (dotted string like bead.created), Ts (time.Time), Actor (who did it), Subject (what was affected), Message (human-readable description), and Payload (optional json.RawMessage for structured data). Defined in internal/events/events.go.
Provider: The full read/write interface for event backends. Embeds Recorder for writing and adds List, LatestSeq, Watch, and Close. Implementations: FileRecorder (built-in JSONL file), Fake (in-memory test double), FailFake (error-returning test double), and exec.Provider (user-supplied script). Defined in internal/events/events.go.
Recorder: The write-only sub-interface. Contains a single method Record(Event) that is best-effort: errors are logged to stderr, never returned to callers. Used by subsystems that only need to emit events. Defined in internal/events/events.go.
Watcher: A cursor that yields events one at a time. Created by Provider.Watch(ctx, afterSeq). Blocks on Next() until a new event arrives, the context is canceled, or the watcher is closed. Defined in internal/events/events.go.
Filter: A query predicate for List and ReadFiltered. Supports filtering by Type (exact match), Actor (exact match), Since (time.Time lower bound), and AfterSeq (uint64 sequence cursor). Zero-valued fields are ignored. Multiple non-zero fields are ANDed. Defined in internal/events/reader.go.
Discard: A sentinel Recorder that silently drops all events. Used when event recording is unwanted (e.g., certain test scenarios). Defined in internal/events/events.go.

Architecture

The event bus is a single interface with three implementations, selected at startup by the [events].provider config key or GC_EVENTS env var.

                     events.Provider (interface)
                    /        |          \
                   /         |           \
           FileRecorder    Fake        exec.Provider
          (JSONL file)   (in-memory)  (user script)

  Recorder (sub-interface: write-only)
      |
  Discard (no-op sentinel)

Provider resolution (in cmd/gc/providers.go:newEventsProvider):

GC_EVENTS env var (highest priority)
[events].provider in city.toml
Default: file-backed JSONL at .gc/events.jsonl

Valid provider values: "" (default FileRecorder), "fake" (in-memory), "fail" (broken test double), "exec:<script-path>" (user-supplied script).

Data Flow

The most common operation: recording an event and reading it back.

Record(Event{Type, Actor, Subject, Message, Payload})
  --> provider assigns Seq (monotonically increasing)
  --> provider fills Ts if zero (time.Now())
  --> provider writes one JSON line to .gc/events.jsonl (FileRecorder)
  --> errors logged to stderr, never returned (best-effort)

List(Filter{Type, Actor, Since, AfterSeq})
  --> reads all events from .gc/events.jsonl
  --> applies filter predicates (AND semantics)
  --> returns matching events as []Event

LatestSeq()
  --> scans file for highest Seq value
  --> returns 0 if empty or missing

Watch(ctx, afterSeq)
  --> returns a Watcher positioned after afterSeq
  --> Watcher.Next() blocks until new event arrives
  --> context cancellation unblocks Next() with ctx.Err()

Watch lifecycle for FileRecorder:

Watch(ctx, afterSeq=5)
  --> creates fileWatcher{path, afterSeq=5, poll=250ms}

Next() loop:
  1. Drain internal buffer (previously fetched events)
  2. Check context (return ctx.Err() if canceled)
  3. ReadFrom(path, byteOffset) to get new events since last read
  4. Filter to events with Seq > afterSeq
  5. Buffer matching events, drain on next iteration
  6. If no new events, sleep 250ms and retry

Watch lifecycle for Fake:

Watch(ctx, afterSeq=5)
  --> creates fakeWatcher{fake, afterSeq=5, ctx}

Next() loop:
  1. Scan in-memory Events slice under mutex
  2. Return first event with Seq > afterSeq
  3. If none found, block on select:
     - ctx.Done() --> return ctx.Err()
     - fake.notify channel --> new event recorded, retry

Key Types

Event (internal/events/events.go) -- The immutable event record. JSON-tagged for JSONL serialization. Payload uses json.RawMessage for arbitrary structured data and is omitted from JSON output when nil.
Provider (internal/events/events.go) -- The full read/write interface. Embeds Recorder and adds List, LatestSeq, Watch, Close.
Recorder (internal/events/events.go) -- The write-only sub-interface. Single method Record(Event) with best-effort semantics.
Filter (internal/events/reader.go) -- Query predicate for List and ReadFiltered. Zero values are ignored; non-zero fields are ANDed.
FileRecorder (internal/events/recorder.go) -- Production implementation. Appends JSONL to .gc/events.jsonl with O_APPEND for cross-process safety and a sync.Mutex for in-process serialization.

Invariants

These properties must hold for any correct Provider implementation. They are enforced by the conformance suite in internal/events/eventstest/conformance.go.

Seq is monotonically increasing. For any two events recorded by the same provider, the later event has a strictly greater Seq.
Seq is unique. No two events share the same Seq value, even under concurrent recording.
Seq is auto-filled by the provider. Callers do not set Seq; the provider assigns it on Record.
Ts is auto-filled when zero. If the caller provides a zero Ts, the provider fills it with the current time. An explicit non-zero Ts is preserved.
Record is best-effort. Recording errors are logged to stderr but never returned to callers. The caller's operation must not fail because event recording failed.
Events are immutable once recorded. There is no Update or Delete operation. The append-only log only grows.
List with empty Filter returns all events. A zero-valued Filter matches everything.
Filter fields are ANDed. When multiple Filter fields are non-zero, an event must match all of them to be included.
LatestSeq returns 0 for an empty provider. Missing file, empty file, or no events all return (0, nil).
Watch(ctx, afterSeq) yields only events with Seq > afterSeq. Existing events at or before afterSeq are never returned.
Watch.Next() blocks until an event arrives or the context is canceled. Context cancellation returns context.Canceled or context.DeadlineExceeded.
Malformed lines are skipped. ReadAll, ReadFiltered, and ReadFrom silently skip lines that fail JSON unmarshalling. This handles partial writes from crashes.
Missing file returns nil, not error. ReadAll and ReadLatestSeq return (nil, nil) and (0, nil) respectively for nonexistent files.
FileRecorder resumes Seq across restarts. NewFileRecorder scans the existing file to find the maximum Seq, so new events continue monotonically even after a process restart.
Payload is omitted from JSON when nil. The omitempty tag ensures events without payloads produce compact JSON lines.

Interactions

Depends on	How
`encoding/json`	All serialization uses standard library JSON
`context`	Watch and Watcher use contexts for cancellation
(no internal Gas City dependencies)	Event Bus is a pure Layer 0-1 primitive with no upward dependencies

Depended on by	How
`cmd/gc/controller.go`	Records `controller.started` and `controller.stopped` events at lifecycle boundaries; passes `Recorder` to reconciliation and shutdown
`cmd/gc/session_lifecycle_parallel.go`	Records `session.woke` and parallel lifecycle `session.stopped` events (renamed from `agent.*` by `be8debd8`)
`cmd/gc/session_reconciler.go`	Records `session.crashed`, `session.draining`, `session.idle_killed`, `session.stopped`, and `session.updated` while reconciling session beads
`cmd/gc/cmd_runtime_drain.go`	Records manual `session.draining` and `session.undrained` events
`cmd/gc/cmd_handoff.go`	Records handoff-related `session.draining` and `session.stopped` events
`cmd/gc/order_dispatch.go`	Records `order.fired`, `order.completed`, `order.failed` events during order dispatch
`cmd/gc/cmd_events.go`	CLI `gc events` command: reads and displays events with filtering (`--type`, `--since`), watch mode (`--watch`), and sequence query (`--seq`)
`cmd/gc/cmd_event_emit.go`	CLI `gc event emit` command: records custom events from scripts and bd hooks (best-effort, always exits 0)
`cmd/gc/cmd_agent.go`	Records session lifecycle events during start/stop/restart operations
`cmd/gc/cmd_suspend.go`	Records `city.suspended` and `city.resumed` events
`cmd/gc/cmd_mail.go`	Records CLI `mail.*` events for send, read, archive, reply, mark-read/unread, and delete operations
`cmd/gc/cmd_convoy.go`	Records `convoy.created` and `convoy.closed` events
`internal/orders/triggers.go`	Event triggers query the Provider via `List(Filter{Type, AfterSeq})` to check if matching events exist since the last cursor position

Code Map

Path	Description
`internal/events/events.go`	Event struct, Recorder interface, Provider interface, Watcher interface, event type constants, Discard sentinel
`internal/events/recorder.go`	FileRecorder: JSONL file-backed Provider with O_APPEND + mutex; fileWatcher with 250ms polling
`internal/events/reader.go`	Filter struct, ReadAll, ReadFiltered, ReadLatestSeq, ReadFrom (byte-offset incremental reading)
`internal/events/fake.go`	Fake: in-memory Provider for testing with channel-based watcher notification; FailFake: error-returning variant
`internal/events/exec/exec.go`	exec.Provider: delegates all operations to a user-supplied script via fork/exec with JSON wire protocol
`internal/events/exec/exec_test.go`	exec.Provider tests including stateful mock script, conformance suite, timeout, and error handling
`internal/events/eventstest/conformance.go`	RunProviderTests: 20+ subtests that any Provider must pass; RunConcurrencyTests: concurrent recording safety
`internal/events/conformance_test.go`	Wires FileRecorder and Fake into the conformance suite
`internal/events/events_test.go`	FileRecorder-specific tests: write, payload round-trip, monotonic seq, concurrent safety, seq resume, timestamp handling
`cmd/gc/providers.go`	eventsProviderName: resolution logic (GC_EVENTS env -> city.toml -> default); newEventsProvider: factory function
`cmd/gc/cmd_events.go`	`gc events` CLI: list, filter, watch, payload-match, seq query
`cmd/gc/cmd_event_emit.go`	`gc event emit` CLI: best-effort custom event recording

Event Type Constants

All event type constants in events.KnownEventTypes are defined in internal/events/events.go and must have a registered payload for the API/SSE projection:

Constant	Value	Emitted by
`SessionWoke`	`session.woke`	`cmd/gc/session_lifecycle_parallel.go` when a reconciler start succeeds
`SessionStopped`	`session.stopped`	`cmd/gc/session_lifecycle_parallel.go`, `cmd/gc/session_reconciler.go`, `cmd/gc/controller.go`, `cmd/gc/cmd_handoff.go`, `cmd/gc/cmd_session.go`
`SessionCrashed`	`session.crashed`	`cmd/gc/session_reconciler.go` when a runtime exists but the expected child process is gone
`SessionDraining`	`session.draining`	`cmd/gc/session_reconciler.go`, `cmd/gc/cmd_runtime_drain.go`, `cmd/gc/cmd_handoff.go`
`SessionUndrained`	`session.undrained`	`cmd/gc/cmd_runtime_drain.go`
`SessionQuarantined`	`session.quarantined`	Registered/reserved; no production emitter today
`SessionIdleKilled`	`session.idle_killed`	`cmd/gc/session_reconciler.go` when idle timeout handling stops a session
`SessionSuspended`	`session.suspended`	Registered/reserved; no production emitter today
`SessionUpdated`	`session.updated`	`cmd/gc/session_reconciler.go` on live-only config drift repair
`BeadCreated`	`bead.created`	Bead creation hooks
`BeadClosed`	`bead.closed`	Bead close hooks
`BeadUpdated`	`bead.updated`	Bead update hooks
`MailSent`	`mail.sent`	Mail send/API handlers and handoff command
`MailRead`	`mail.read`	Mail read command
`MailArchived`	`mail.archived`	Mail archive command and API handler
`MailMarkedRead`	`mail.marked_read`	Mail mark-read command and API handler
`MailMarkedUnread`	`mail.marked_unread`	Mail mark-unread command and API handler
`MailReplied`	`mail.replied`	Mail reply command and API handler
`MailDeleted`	`mail.deleted`	Mail delete command and API handler
`ConvoyCreated`	`convoy.created`	Convoy creation
`ConvoyClosed`	`convoy.closed`	Convoy close
`ControllerStarted`	`controller.started`	Per-city controller startup
`ControllerStopped`	`controller.stopped`	Per-city controller shutdown
`SupervisorStarted`	`supervisor.started`	Machine-wide supervisor process: emitted once per startup, classifying how the previous supervisor instance exited (`clean` — it completed its STOPPING path; `crash` — a prior instance ran but did not complete STOPPING; `unknown` — no evidence of a prior instance) from the clean-shutdown handoff token. Attribution is best-effort across binary up/downgrades: a mixed-version window can misattribute one start, self-correcting on the next cycle.
`SupervisorShutdownRequested`	`supervisor.shutdown_requested`	Machine-wide supervisor process: emitted when a shutdown trigger is observed (SIGINT/SIGTERM or socket `stop`), before the cascade of per-city `controller.stopped` events. Carries trigger attribution (source, signal, client addr, mode).
`SupervisorRequest`	`supervisor.request`	Machine-wide supervisor API bounded request audit. Omits request bodies, raw origins, raw remote addresses, and query strings.
`CitySuspended`	`city.suspended`	City suspend command
`CityResumed`	`city.resumed`	City resume command
`RequestResultCityCreate`	`request.result.city.create`	Supervisor/API city create completion
`RequestResultCityUnregister`	`request.result.city.unregister`	Supervisor city unregister completion
`RequestResultSessionCreate`	`request.result.session.create`	API async session create completion
`RequestResultSessionMessage`	`request.result.session.message`	API async session message completion
`RequestResultSessionSubmit`	`request.result.session.submit`	API async session submit completion
`RequestFailed`	`request.failed`	Supervisor/API async request failure handlers
`CityCreated`	`city.created`	City init lifecycle diagnostics
`CityUnregisterRequested`	`city.unregister_requested`	City unregister lifecycle diagnostics
`OrderFired`	`order.fired`	Order dispatch when a trigger is due
`OrderCompleted`	`order.completed`	Order dispatch on successful completion
`OrderFailed`	`order.failed`	Order dispatch on failure
`ProviderSwapped`	`provider.swapped`	Controller provider-swap reload path
`WorkerOperation`	`worker.operation`	Worker session handle and runtime handle operation tracing
`ProjectIdentityStamped`	`project.identity.stamped`	Project identity writer when a scope receives or reconciles project identity metadata
`SupervisorFSPressureSkippedTick`	`supervisor.fs_pressure.skipped_tick`	Supervisor filesystem-pressure guard when a scan tick is skipped under pressure
`ExtMsgBound`	`extmsg.bound`	External messaging bind handler
`ExtMsgUnbound`	`extmsg.unbound`	External messaging unbind handler
`ExtMsgGroupCreated`	`extmsg.group_created`	External messaging group ensure handler
`ExtMsgAdapterAdded`	`extmsg.adapter_added`	External messaging adapter registration handler
`ExtMsgAdapterRemoved`	`extmsg.adapter_removed`	External messaging adapter unregister handler
`ExtMsgInbound`	`extmsg.inbound`	External messaging inbound adapter pipeline
`ExtMsgOutbound`	`extmsg.outbound`	External messaging outbound adapter pipeline
`EventsRotated`	`events.rotated`	File event recorder after rotating an active log, carrying the archived seq range

Configuration

The event bus backend is selected via the [events] section in city.toml:

[events]
provider = ""   # "" (default: file JSONL), "fake", "fail", or "exec:/path/to/script"

The GC_EVENTS environment variable overrides the config file. This is used primarily in tests (GC_EVENTS=fake for in-memory, GC_EVENTS=fail for error path testing).

The default FileRecorder stores events at .gc/events.jsonl relative to the city directory. The file is created automatically on first write.

Storage Format

Events are stored as newline-delimited JSON (JSONL / NDJSON). Each line is a complete, self-contained JSON object:

{"seq":1,"type":"controller.started","ts":"2026-03-01T10:00:00Z","actor":"gc"}
{"seq":2,"type":"session.woke","ts":"2026-03-01T10:00:01Z","actor":"gc","subject":"worker-1","message":"session woke successfully"}
{"seq":3,"type":"bead.created","ts":"2026-03-01T10:00:05Z","actor":"human","subject":"gc-42","payload":{"title":"Fix bug","labels":["urgent"]}}

The JSONL format provides:

Append-only writes -- new events are appended without reading or rewriting the file
Crash resilience -- partial writes (truncated last line) are skipped by readers
Incremental reads -- ReadFrom(path, byteOffset) reads only new data from a known position
Cross-process safety -- O_APPEND flag ensures atomic appends at the OS level

Exec Provider Wire Protocol

The exec provider (exec:<script>) delegates operations to a user-supplied script. The script receives the operation name as its first argument:

Operation	Script invocation	Stdin	Stdout
`ensure-running`	`script ensure-running`	(none)	(ignored)
`record`	`script record`	JSON event	(ignored)
`list`	`script list`	JSON filter	JSON array of events
`latest-seq`	`script latest-seq`	(none)	Integer
`watch`	`script watch <afterSeq>`	(none)	NDJSON stream

Exit code 2 means "unknown operation" and is treated as success (forward compatible). ensure-running is called once per provider lifetime via sync.Once.

Testing

The event bus has a layered testing strategy aligned with TESTING.md:

Conformance suite (internal/events/eventstest/conformance.go): RunProviderTests runs 20+ subtests against any Provider implementation, covering Record+List round-trip, auto-fill of Seq and Ts, field preservation, List filtering (by type, actor, afterSeq, since, combined), no-match and empty cases, LatestSeq (empty, after records, monotonic), Watch (existing events, new events, afterSeq cursor, context cancellation), and Close. RunConcurrencyTests verifies concurrent Record safety with unique Seq values.

FileRecorder-specific tests (internal/events/events_test.go): Tests for JSONL writing, payload round-trip, payload omission when nil, monotonic Seq, concurrent safety (10 goroutines x 10 events), Seq resume across process restarts, timestamp auto-fill and explicit preservation.

Reader tests (internal/events/events_test.go): Tests for ReadAll (missing file, empty file), ReadFiltered (by type, actor, since, combined, AfterSeq, no match), ReadLatestSeq (missing, empty, after writes), ReadFrom (full read, incremental from mid-file, missing file, no new data).

Fake tests (internal/events/events_test.go): Record and List for in-memory provider, LatestSeq, Watch with goroutine recording, FailFake error paths.

Conformance wiring (internal/events/conformance_test.go): Runs both RunProviderTests and RunConcurrencyTests against FileRecorder and Fake.

exec.Provider tests (internal/events/exec/exec_test.go): Record via stdin capture, List and LatestSeq with mock scripts, Watch with NDJSON streaming, ensure-running called once, exit 2 handling, error propagation, timeout enforcement, and full conformance suite against a stateful jq-based mock script.

Compile-time interface checks: Both FileRecorder and Fake have var _ Provider = (*T)(nil) compile-time assertions in events_test.go. The exec Provider has its own in exec.go.

Known Limitations

FileRecorder Watch uses polling, not inotify. The fileWatcher polls the JSONL file every 250ms via ReadFrom. This adds up to 250ms latency for event delivery and uses CPU for polling. A future optimization could use fsnotify to wake on file changes. The Fake provider uses channel-based notification for zero-latency delivery in tests.
No event retention or rotation. The JSONL file grows without bound. There is no built-in log rotation, retention policy, or compaction. For long-running cities, manual truncation or external log rotation is needed.
ReadFiltered streams without indexes. ReadFiltered scans the JSONL file once, applies Filter as it reads, and stops early when a positive direct-provider Limit is reached. There are still no indexes, so broad time/type/actor/subject queries remain linear in the event log until their limit is satisfied. ReadFrom with byte offsets provides incremental reading for the Watch path.
No event schema validation. Event types are string constants with no runtime validation. Recording an event with a misspelled type succeeds silently.
Multiplexer limits are global post-merge caps. The multiplexer clears per-provider Filter.Limit, merges and sorts provider results, then applies the global limit so cross-city ordering stays correct. This means a multiplexer Limit does not cap work inside each provider.
Exec provider Watch is subprocess-lifetime-bound. The exec watcher reads from a long-running subprocess's stdout. If the subprocess exits, the watcher reports an error rather than reconnecting.

Navigation Menu

Summary

Key Concepts

Architecture

Data Flow

Key Types

Invariants

Interactions

Code Map

Event Type Constants

Configuration

Storage Format

Exec Provider Wire Protocol

Testing

Known Limitations

See Also

FilesExpand file tree

event-bus.md

Latest commit

History

event-bus.md

File metadata and controls

Summary

Key Concepts

Architecture

Data Flow

Key Types

Invariants

Interactions

Code Map

Event Type Constants

Configuration

Storage Format

Exec Provider Wire Protocol

Testing

Known Limitations

See Also