Event feed connector: the run loop and tier-2 driver (2/3)

CONTRIBUTING.md

@@ -227,6 +227,12 @@ All SDKs are generated from a single Smithy specification. When adding support f
    - **Kotlin:** `make kt-generate-services`
    - **Python:** `make py-generate`
 
+   `go/pkg/basecamp/eventfeed/` is outside that drift check by design: it is
+   hand-written §23 infrastructure rather than a wrapper over generated
+   operations, so nothing about it is derivable from the spec. It is verified
+   instead by the tier-2 conformance driver in the package, which replays every
+   fixture under `conformance/event-feed/fixtures/`.
+
 4. **Add tests** for each SDK
 
 5. **Add conformance tests** (`conformance/tests/`) covering the new operations

SPEC.md

@@ -2498,7 +2498,24 @@ is shared:
   which keeps hosts and tests deterministic.
 - **`close()`** is idempotent and callable from any context: it abandons, never drains.
   Undelivered buffered events are abandoned — the next run re-serves from the last **usable**
-  checkpoint (see the exclusion under Entry Boundary).
+  checkpoint (see the exclusion under Entry Boundary). Cancellation is visible before it
+  returns; it does **not** wait for the run to unwind, because every consumer callback runs
+  on the run's own execution context and waiting there would deadlock on the caller.
+- **`close()` does not order a second connector over the same checkpoint store, and cannot.**
+  A save decided just before the close is still written after it. That is intended: the
+  position was accepted and its events delivered before the close, so dropping the write
+  would silently re-deliver them. **The save therefore runs under a context detached from
+  the run's cancellation** — carrying the run's context values, but not its cancellation.
+  Passing the live run context instead makes the guarantee conditional on the store: one
+  that ignores its context writes anyway, and one that honors it — which this contract
+  permits, and says nothing against — sees a cancelled context and drops the position. The
+  trade is explicit: a store that blocks indefinitely delays the run's exit, and therefore
+  `wait()`, rather than being released by `close()`. That is bounded by the store's own
+  behavior, where a dropped position is unbounded re-delivery with nothing recording it. **`wait()`** is the quiescence point — it blocks
+  until the run has exited, after which no save can be in flight. A consumer that owns the
+  iteration needs nothing extra, since the iteration terminating is the same guarantee.
+  Await termination — or `wait()` — before opening a second connector over the same store.
+  `wait()` is not callable from a consumer callback, for the reason `close()` does not wait.
 - A consumer break takes the identical teardown path; the in-flight page's checkpoint is
   **not** saved.
 - All Observer callbacks fire on the consumer's execution context, never concurrently with a
@@ -2681,6 +2698,59 @@ Draining is Terminal(`protocol_fatal`) immediately (the state-generic rule under
 Disconnect Dispatch) — the drain is not completed, the held entry position is NOT saved,
 and no `caught_up` is announced; only recoverable failures defer.
 
+**A socket outcome observed while a poll seam call is in flight is deferred to the page
+boundary, and the wait for that call is bounded by DETECTION WINDOW + GRACE PHASE.** This
+is transition 21 in CatchingUp, the only state that holds a wire call open across a socket
+event, and it is normative for every SDK.
+
+The deferral itself is the finish-the-page ordering the rest of §23 already states: the
+in-flight page is accepted, delivered and saved, and only then is the socket's outcome
+dispatched. It applies to **every** socket outcome, and that expressly includes a staleness
+expiry — a socket that goes silently half-open produces no frame and no read error, so its
+expiry is the only evidence there will ever be, and an implementation whose in-flight-poll
+wait does not observe `staleness` cannot detect that socket at all. Disposing the attempt
+where the expiry is observed is NOT conformant: it strands the deliveries and the save of a
+page the server had already served.
+
+The two intervals bound the whole sequence, and their sum is the worst case:
+
+- **Detection window** — `EVENT_FEED_STALE_AFTER`, the staleness window, measured from the
+  last inbound frame. It is what decides the socket is dead, under the ordinary evaluation
+  rule: a firing superseded by a frame the reader took first, or one whose window
+  overlapped a blocked hand-off, is not evidence and must not be deferred.
+- **Grace phase** — one further `EVENT_FEED_STALE_AFTER`, measured from the deferral, after
+  which the seam call is abandoned to the deferred outcome's teardown (the teardown
+  cancels it — Seam-Call Semantics). It is a **deadline read from the clock at the instant
+  of deferral**, not a window, and it carries two immunities that are the point of naming
+  it:
+  - **Immune to frame resets.** A frame arriving inside the phase re-arms `staleness` in
+    the ordinary way and MUST NOT move the deadline. The phase bounds how long the
+    connector waits for an abandoned call, which is unrelated to whether the peer is still
+    talking.
+  - **Immune to suspension.** The full-queue suspension rule below rests on a full queue
+    proving the peer is outrunning a connector that is still consuming. This is the one
+    wait that deliberately stops consuming — an outcome already occupies the single
+    deferral slot — so the premise is false here by construction, and a suspended
+    evaluation MUST NOT extend the phase.
+
+An implementation may wake as often as it likes and from whatever source it has; wakes may
+be early or late, and the deadline is what decides. In particular this introduces **no new
+timer kind**: the six kinds and the per-state exact timer sets are both unchanged, and an
+implementation that re-arms `staleness` purely to obtain a wake keeps CatchingUp's set at
+{`staleness`}.
+
+**"Observed" means handed to the state machine, and the boundary is normative.** A frame
+the transport reader has taken off the socket but not yet handed over is not observed, and
+no implementation is required to find it. That is not a tolerance granted for convenience:
+making it observable requires the read to complete inside a critical section the scan can
+enter, and the read blocks indefinitely on a quiet socket, so the lock deadlocks the drain
+against a peer that simply stopped talking. Sampling a flag the reader sets after its read
+does not close it either — the flag is published after the read returns, and the scan reads
+it after its own check of the queue, so a frame can arrive and the flag clear between the
+two. What every implementation MUST cover is everything handed over, **plus the one frame a
+blocked hand-off is holding** — the reader is a single goroutine, so there is exactly one,
+and it is why the scan's budget is pump depth + 1 rather than pump depth.
+
 ### Disconnect Dispatch `[conformance]`
 
 Action Cable's `disconnect` is a **text frame**, not a WebSocket close frame, and stock
@@ -2993,6 +3063,9 @@ INTERFACE PollSource
   -- triggered on close(), caller cancellation, AND any teardown of the attempt the call
   -- belongs to (mid-walk socket failure, staleness, a terminal): a superseded poll must
   -- not stall reconnection or return into a disposed attempt. Prompt return required.
+  -- The connector's own side of that is bounded rather than trusting: an outcome observed
+  -- while the call is in flight is awaited for the grace phase and then abandoned to the
+  -- teardown (detection window + grace phase, under the state machine above).
 END
 
 RECORD Cursor           -- exactly one field set; the zero Cursor is the bare present entry
@@ -3288,6 +3361,13 @@ options object with optional fields; Python and Ruby keyword arguments (Ruby wit
 | Signal handler (none ⇒ default-terminal) | `WithSignalHandler` | `signalHandler?` | `signal_handler` | `signalHandler` |
 | Observer (none) | `WithObserver` | `observer?` | `observer` | `observer` |
 
+The two lifecycle methods take each language's native spelling of the same two acts: Go
+`Close()` / `Wait()`, TypeScript `close()` / `wait()`, Python and Ruby `close` / `wait`,
+Kotlin `close()` / `join()`, Swift `close()` / `wait()`. Where the language's streaming
+idiom already exposes the run's completion — a Kotlin `Job`, a Swift `Task` — that handle
+IS `wait()` and no second method is added; what must exist is a way to observe the run's
+exit that is not `close()`.
+
 The Observer is a struct of optional callbacks in the `httptrace.ClientTrace` style —
 extensible without breaking implementers: `connecting(attempt, delay)`, `connected()`,
 `confirmed()`, `disconnected(reason, error)`, `catch_up_started(cursor)`,

conformance/event-feed/README.md

@@ -45,6 +45,26 @@ The strict-match rule splits by action class; every driver implements exactly th
   unmatched when the script ends fails the scenario, and the seam-call counts in
   `finally` count it either way.
 
+A driver's step pointer moves in two ways, and both are needed for the atomic
+handoff to hold. It advances inside the critical section that satisfies a
+rendezvous, which covers a driver blocked ON one; and where the pointer has not
+yet reached a step the recorded history has already satisfied, the arrival rule
+reads THROUGH that step to the one after it — a poll's page is delivered and
+then checkpointed by one causal chain in the connector, and the driver need not
+have reached its `expectDelivered` step by the time the save lands. A step the
+DRIVER performs rather than waits for (`serve`, `advance`, `fireTimer`,
+`sever`, `serverClose`) is not read through: the connector cannot react before
+the driver acts, so the pointer is handed to the next step before such a step
+acts, and an action arriving under one the driver has not performed is early.
+
+What a driver can hold an action to is when it OBSERVED it. A save is observed
+where the connector makes it, but a driver running a real socket observes an
+outbound frame only when it reads one, which lags the write by a hop — so a
+frame written a step early can still be read after the driver has stepped on.
+The rule is not weaker for it; the observation is. A driver that wants the
+write's own instant pinned needs a witness at the write, not a stricter reading
+of this rule.
+
 ## Validation
 
 `schema.json` is the contract. `make event-feed-fixtures-check` validates the schema

go/README.md

@@ -599,6 +599,142 @@ webhooks, err := account.Webhooks().List(ctx, bucketID, nil)
 err = account.Webhooks().Delete(ctx, webhookID)
 ```
 
+## Event Feed (experimental)
+
+The `eventfeed` package is the account-wide event feed connector: an Action Cable
+subscription for live push, plus a poll lane that catches up on entry, repairs on a
+timer, and resumes after a disconnect. You consume it as one serial, deduplicated
+stream of events; the connector owns reconnection, backoff, staleness detection, and
+the durable position.
+
+**Experimental: the Layer-1 seam adapters have not landed yet.** The connector performs
+no HTTP API I/O of its own: every HTTP exchange reaches the wire through a seam backed by
+a generated operation. Its one direct wire act is the Action Cable dial above — the
+connector connects verbatim to the URL a generated `CreateStreamTicket` call returned,
+which is the sanctioned non-HTTP wire act. The adapters that build those seams over the
+generated `CreateStreamTicket` and `PollEvents` operations are still to come. Until they
+do, a consumer must supply the `TicketMinter` and `PollSource` implementations itself, and
+the exported surface may still change as they land.
+
+```go
+import (
+    "context"
+    "errors"
+    "fmt"
+    "log"
+
+    "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed"
+)
+
+// The two seams the host supplies until the Layer-1 adapters land. Each call is
+// exactly one generated operation:
+//
+//	MintStreamTicket(ctx) (eventfeed.StreamTicket, error)   // CreateStreamTicket
+//	Poll(ctx, cursor, filters) (eventfeed.PollPage, error)  // PollEvents
+var minter eventfeed.TicketMinter
+var polls eventfeed.PollSource
+
+ctx := context.Background()
+
+feed, err := eventfeed.New("https://3.basecampapi.com", "5951425", minter, polls,
+    eventfeed.WithFilters(eventfeed.Filters{Types: []string{"message.created"}}),
+    eventfeed.WithCheckpointStore(eventfeed.NewFileCheckpointStore("/var/lib/myapp/feed.json")),
+    eventfeed.WithConsumerNamespace("myapp"),
+    eventfeed.WithSignalHandler(func(sig eventfeed.Signal) eventfeed.Disposition {
+        switch s := sig.(type) {
+        case eventfeed.FeedGap:
+            // History before this id is gone. Accept resumes at the server's
+            // resume URL, having acknowledged what it skips.
+            log.Printf("feed gap: history before %d is gone", s.EpochAfterID)
+            return eventfeed.Accept
+        case eventfeed.BufferOverflow:
+            log.Printf("live buffer dropped %d events: %v", s.DroppedCount, s.DroppedIDs)
+            return eventfeed.Terminate
+        }
+        return eventfeed.Terminate
+    }),
+)
+if err != nil {
+    return err // *eventfeed.TerminalError: a usage-coded construction error, zero wire attempts
+}
+defer feed.Close()
+
+for ev, err := range feed.Events(ctx) {
+    if err != nil {
+        var te *eventfeed.TerminalError
+        if errors.As(err, &te) {
+            return fmt.Errorf("event feed terminated (%s): %w", te.Reason, te)
+        }
+        return err
+    }
+    // A feed row is a wake-up signal — enough to route, not enough to act on.
+    // Refetch the recording through the canonical resource API before acting.
+    handle(ev.BucketID, ev.RecordingID, ev.EventType)
+}
+```
+
+Construction validates and does no I/O. The base origin must be `https://` — cleartext
+`http://` is accepted only for localhost/loopback, the same carve-out the client's base
+URL and the connector's cable URL make — because that origin is the trust anchor every
+continuation and resume URL is validated against before an authenticated poll follows
+it. The checkpoint identity's text inputs (origin, account id, consumer namespace,
+filter types) must be valid UTF-8, since the identity encoding is one-to-one only over
+valid UTF-8. Either violation is a `ReasonUsage` construction error with zero wire
+attempts.
+
+`Events` is single-shot: consuming it twice yields one `ReasonUsage` error element.
+`Close` stops the feed without draining, and cancelling the context, calling `Close`, or
+breaking out of the loop all end iteration with **no** error element — a clean stop, and
+the feed is resumable by design.
+
+### Checkpointing
+
+`FileCheckpointStore` is the built-in `CheckpointStore`: one JSON file holding every
+lineage's position, keyed by the four-part checkpoint identity (origin, account,
+consumer namespace, filter key). It writes temp-file-plus-rename at 0600, and it is safe
+for concurrent use within one process but deliberately not across processes. A store
+requires `WithConsumerNamespace` — two independent consumers in one account must not
+share a lineage — and changing filters starts a new lineage, because positions are
+filter-bound.
+
+Only poll pages ever advance the durable position; live event ids never do. What the
+connector publishes is SPEC.md §23's conjunctive save-ordering invariant, and nothing
+stronger: a position is saved only after the retained events it covers have been
+delivered **and** every loss condition in that window has been explicitly accepted.
+Terminate — or no handler — means no save. A load failure is terminal
+(`ReasonCheckpointLoad`, before any wire attempt: silently starting at the present would
+skip history), while a save failure is reported through `Observer.CheckpointSaveFailed`
+and the feed continues.
+
+### Semantic signals
+
+A semantic signal is a condition that changes what the feed can promise, and there are
+exactly two: `BufferOverflow` (the live buffer dropped events, naming the exact ids) and
+`FeedGap` (a 410 — history before `EpochAfterID` is gone). The handler registered with
+`WithSignalHandler` is invoked exactly once per signal, synchronously, on your own
+execution context, and returns `Accept` or `Terminate`.
+
+**With no handler registered, every semantic signal is terminal** — `ReasonBufferOverflow`
+or `ReasonFeedGap` — so an unhandled signal cannot disappear, and a 410 never silently
+auto-continues. `Accept` on a `FeedGap` resumes via the server's resume URL; `Accept` on a
+`BufferOverflow` means you own the acknowledged incompleteness. The `Observer.Gap` and
+`Observer.BufferOverflow` callbacks see the same conditions but are observability only:
+the disposition lives exclusively in the handler.
+
+### Terminal versus continuable
+
+Continuable failures — a dropped socket, a staleness expiry, a throttled or transient
+mint or poll, a rejected position (400-position or 409) — never reach the consumer as
+errors. They ride the reconnect cycle (full-jitter backoff) or the poll-retry timer, and
+you see them through `Observer` callbacks if you register any.
+
+A terminal condition ends the iteration with exactly one final `*eventfeed.TerminalError`
+element carrying a `TerminalReason`: `subscription_rejected`, `protocol_fatal`,
+`filter_invalid`, `authorization_failed`, `checkpoint_load`, `usage`, `buffer_overflow`,
+`feed_gap`, `invalid_continuation`, `poll_failed`, `mint_failed`, or `invalid_cable_url`.
+Switch on `te.Reason` rather than on message text; `errors.Unwrap` reaches the generated
+error behind `mint_failed` and `poll_failed`.
+
 ## Pagination
 
 List methods auto-paginate: they follow the API's `Link: rel="next"` headers and

go/pkg/basecamp/eventfeed/buffer_test.go

@@ -0,0 +1,36 @@
+package eventfeed
+
+import "testing"
+
+// TestLiveBufferAddClearsEvictedSlots pins the eviction half of the live
+// buffer's memory ceiling. SPEC.md §23 publishes the connector's worst case
+// as (pump depth + EVENT_FEED_LIVE_BUFFER_CAPACITY) × EVENT_FEED_MAX_FRAME_BYTES;
+// a reslice alone removes the evicted event LOGICALLY while the slice that
+// results still points into the same backing array, whose prefix keeps that
+// event's strings reachable until a later reallocation. Under sustained
+// overflow — the one condition eviction happens under — that is a second
+// buffer's worth of payload held by events which no longer count toward
+// occupancy. shift already zeroes for exactly this reason.
+func TestLiveBufferAddClearsEvictedSlots(t *testing.T) {
+	b := newLiveBuffer(2, nil)
+	b.add(Event{ID: 1, Kind: "message", EventType: "message.created"})
+	b.add(Event{ID: 2, Kind: "message", EventType: "message.created"})
+	// Captured before the eviction: base still spans the backing array the
+	// live slice is resliced from, so base[0] IS the vacated slot.
+	base := b.events
+
+	dropped := b.add(Event{ID: 3, Kind: "message", EventType: "message.created"})
+	if len(dropped) != 1 || dropped[0] != 1 {
+		t.Fatalf("dropped = %v, want [1]", dropped)
+	}
+	if base[0] != (Event{}) {
+		t.Errorf("evicted slot = %+v, want the zero Event — a bare reslice keeps its payload reachable", base[0])
+	}
+	// The eviction itself is unchanged.
+	if got := len(b.events); got != 2 {
+		t.Fatalf("occupancy = %d, want 2", got)
+	}
+	if b.events[0].ID != 2 || b.events[1].ID != 3 {
+		t.Fatalf("buffer = %d,%d, want 2,3", b.events[0].ID, b.events[1].ID)
+	}
+}