Skip to content

feat(core): epoch:seq engine-side sequencing — the P1a port of #1803 (#1806) #1807

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
aarontrowbridge merged 1 commit into from
Oct 17, 2026
Merged

feat(core): epoch:seq engine-side sequencing — the P1a port of #1803 (#1806) #1807

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 23 additions & 5 deletions packages/app-bundle/manifest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -338,6 +338,9 @@
"packages/llm/src/protocols/bedrock-converse.ts": "834a415fa8ddf9ad1224133f8352d0a926694f19f0954201d182111c171ec03b",
"packages/llm/test/provider/bedrock-converse.test.ts": "e8e14d60e8502e928f170f8ac5b406c44bf16d2dd34996ddc2fb7b9261decf85",
"packages/opencode/src/agent/agent.ts": "512372a4c62997e9565266638ada19b9dc0a3685b0c0483fd28fede2abdb5c9b",
"packages/opencode/src/bus/global-replay.ts": "d4d64e027ab136655db088419b33e3ae524c1d3d8f892b5bb8697050ddc03d33",
"packages/opencode/src/bus/global.ts": "65b3b380936fc34b2c5ef9f3ab419a6c8e6d4cfc696302a266188f26ccdc320a",
"packages/opencode/src/control-plane/workspace.ts": "ab986282d1bd15b7f4473e33766884e17c97684ef0778cd8d84fd173604175cc",
"packages/opencode/src/effect/runner.ts": "b660b5b22df810b16293c23c6d832e5e8dcfb12b9a8a858c875d2c051c855b81",
"packages/opencode/src/mcp/browser.ts": "ecdda42cbd55a97b8d14050aa15c6e6450a1d63b0a641c8c8e5b81e50f957c87",
"packages/opencode/src/mcp/index.ts": "da45165e6791a5b30c8962c7fd0ea391bb8e1931cc2357c6b90e9fd090cdeb0d",
Expand Down Expand Up @@ -415,11 +418,15 @@
"packages/opencode/test/server/amicode-widgets.test.ts": "14fc6ccff4e83a5b005a724936e9f0d7ce411fab74109f319e8bc748665f7ab8",
"packages/opencode/test/server/auth-accept-set.test.ts": "8fe698812ed9a809d21ace1a9751bd3f1976ab1e58ad5bda65dbfff58df8e3bc",
"packages/opencode/test/server/gateway-secret.test.ts": "8d38ec187098be99f698c5e795b867a0e2fe5e2ef2aa3a9e2c03a8b0147d4473",
"packages/opencode/test/server/global-bus-sequencing.test.ts": "db0998c0f4715bc5d600b006c18c37133e5405e735853d9b31346501da31367f",
"packages/opencode/test/server/global-event-replay.test.ts": "c6c1fdba829c7b9a82976a90d71073086078d438f678c11f8577b21044cf3f56",
"packages/opencode/test/server/httpapi-compression.test.ts": "5f7603d7018e0b95f1c2f36c01dd297c01770bfcdd91ce29778e03b7d6143f75",
"packages/opencode/test/server/httpapi-event-replay.test.ts": "c94c13dbfb4ca5d5e9b80f4c1a17955d1530a8bf9141b9a83a810ee56743269f",
"packages/opencode/test/server/httpapi-event-sequencing.test.ts": "a2ec57f999b26b8ac9e4dfd6208c8028264cb0b267f4e6383429df62738967fa",
"packages/opencode/test/server/httpapi-event.test.ts": "36263284302434461c4c6cf3e6652e094f4a02782acbe5de26c489748fe4cb50",
"packages/opencode/test/server/httpapi-file-write-gate.test.ts": "0c91b7417e615850edfde3f2878207dab6b6624d4cfbb6e6783b2e6d9aa4dff0",
"packages/opencode/test/server/httpapi-file.test.ts": "d3386129bc81467ecbcc57a3861f53767ee56ac41b4ff6ad58d8b168fb3b3373",
"packages/opencode/test/server/httpapi-global-event.test.ts": "6359f12508e6a34bfd9e0abe72b03113f933e1c83b9c55aae2a84c4bc5222bad",
"packages/opencode/test/server/httpapi-instance-route-auth.test.ts": "4b73a89d2431c02bc3b665d93f9002914e51eb1add979dc1867efbcbc0c36e25",
"packages/opencode/test/server/httpapi-provider.test.ts": "9e8b3e085b478e3f99b689ffff600e10c0b4b50b3cabdbd749f88dae19bf5904",
"packages/opencode/test/server/httpapi-session.test.ts": "eda6fd763ac4c17891cedaa4ccd49597c03e7227fcfdcf0f94a1b4545ba623ed",
Expand Down Expand Up @@ -842,6 +849,7 @@
"packages/opencode/src/server/amicode/widget-runtime.ts": "a5fe95eba46df87c4b4c54bc07bb91587b73e88eb90db889bc87358ece4b3f27",
"packages/opencode/src/server/amicode/widgets.ts": "d15481086c41b365c28937f002bfcdeb8c2a9f68066413da695fbac8b7acdb61",
"packages/opencode/src/server/shared/public-ui.ts": "a843b4e84673d9a5e34f36eb7cffbb499850f7588154c23061fc8a81cd1369ef",
"packages/opencode/src/server/shared/sse-bounded-writer.ts": "ba2b6cd332ae6a1aa8a0fe72af0f9faabccc28cc5ab5b66b45fa6bd7ee08310e",
"packages/opencode/src/server/shared/sse-tick-telemetry.ts": "66ec419c2c32584c1f546b63fe93dbd7f7061d45be3dd37e9a261aa129027ec2",
"packages/opencode/src/server/shared/ui.ts": "c66ef686ee090f442ee69b6a5f3c985ff005dc135b8265eaf5955f83ecfb86ea",
"packages/opencode/src/server/shared/workspace-routing.ts": "3704b6cf218fc165ce9c1cdad0ee42fc9cf604976e4a552478200f85db80876a",
Expand Down Expand Up @@ -970,9 +978,10 @@
"packages/opencode/src/server/routes/instance/httpapi/groups/mcp.ts": "048efa7b74d43c4ad4a75d0f8421c808760c8c0dcd593f0051a87044e40487d6",
"packages/opencode/src/server/routes/instance/httpapi/groups/provider.ts": "34c6f1ab1a085d1b047c220bbff32a36032dc5e924784e9272cda166cf0ec84b",
"packages/opencode/src/server/routes/instance/httpapi/groups/session.ts": "1421f8c07948b121c7e2005bd9b0fcd6b2c456565ddd1da32d8e6c35d03fac9c",
"packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "5824d237bedb868d97ca24f68da6b137b7c2d84e42111d7c17637619ce87c866",
"packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "494b471bd797a0b20abdb94417b1479c14460c33347800e7caa435c4959b8e41",
"packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "69ca3dd48d95157fc822e1f76eb42ac873df50f22616b85194619ca658613867",
"packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "10bcb79b117beebee04184fb48c1ad789e782e2f7ae4b0c68603edc00f03aee9",
"packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts": "97f8e20d4f1425152a9719ed77af9c42163bbcd34f2cc7ae5ede991bf27430fd",
"packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "7da08d58765cdf2b6ce4648fc70de371402949f5bb7b75696f674d43f2d5365b",
"packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts": "8baf16cad0992c5351b61b0c0fb27903bb25b17e1b25fc7c5d77f807c24413a6",
"packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts": "b37ff4136c4aea85b23083bf441f03e09820e3e4cdd95e4118ca570d8435b94f",
Expand All @@ -993,7 +1002,7 @@
"packages/ui/src/v2/components/text-shimmer-v2.tsx"
],
"exceptions": [],
"extracted_at": "2026-10-15T07:32:24.625Z",
"extracted_at": "2026-10-15T12:25:06.525Z",
"per_package": {
"packages/app": {
"A": 251,
Expand All @@ -1011,8 +1020,8 @@
"D": 0
},
"packages/opencode": {
"A": 87,
"M": 94,
"A": 93,
"M": 97,
"D": 0
},
"packages/schema": {
Expand Down Expand Up @@ -1042,7 +1051,7 @@
}
},
"counts": {
"overlay_total": 971,
"overlay_total": 980,
"deletions": 7,
"server_coupled": 0
},
Expand Down Expand Up @@ -1377,6 +1386,9 @@
"packages/llm/src/protocols/bedrock-converse.ts": "M",
"packages/llm/test/provider/bedrock-converse.test.ts": "M",
"packages/opencode/src/agent/agent.ts": "M",
"packages/opencode/src/bus/global-replay.ts": "A",
"packages/opencode/src/bus/global.ts": "M",
"packages/opencode/src/control-plane/workspace.ts": "M",
"packages/opencode/src/effect/runner.ts": "M",
"packages/opencode/src/mcp/browser.ts": "M",
"packages/opencode/src/mcp/index.ts": "M",
Expand Down Expand Up @@ -1454,11 +1466,15 @@
"packages/opencode/test/server/amicode-widgets.test.ts": "A",
"packages/opencode/test/server/auth-accept-set.test.ts": "A",
"packages/opencode/test/server/gateway-secret.test.ts": "A",
"packages/opencode/test/server/global-bus-sequencing.test.ts": "A",
"packages/opencode/test/server/global-event-replay.test.ts": "A",
"packages/opencode/test/server/httpapi-compression.test.ts": "M",
"packages/opencode/test/server/httpapi-event-replay.test.ts": "A",
"packages/opencode/test/server/httpapi-event-sequencing.test.ts": "A",
"packages/opencode/test/server/httpapi-event.test.ts": "M",
"packages/opencode/test/server/httpapi-file-write-gate.test.ts": "A",
"packages/opencode/test/server/httpapi-file.test.ts": "M",
"packages/opencode/test/server/httpapi-global-event.test.ts": "A",
"packages/opencode/test/server/httpapi-instance-route-auth.test.ts": "M",
"packages/opencode/test/server/httpapi-provider.test.ts": "M",
"packages/opencode/test/server/httpapi-session.test.ts": "M",
Expand Down Expand Up @@ -1881,6 +1897,7 @@
"packages/opencode/src/server/amicode/widget-runtime.ts": "A",
"packages/opencode/src/server/amicode/widgets.ts": "A",
"packages/opencode/src/server/shared/public-ui.ts": "M",
"packages/opencode/src/server/shared/sse-bounded-writer.ts": "A",
"packages/opencode/src/server/shared/sse-tick-telemetry.ts": "A",
"packages/opencode/src/server/shared/ui.ts": "M",
"packages/opencode/src/server/shared/workspace-routing.ts": "M",
Expand Down Expand Up @@ -2012,6 +2029,7 @@
"packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "M",
"packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "M",
"packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "M",
"packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts": "M",
"packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "M",
"packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts": "M",
"packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts": "M",
Expand Down
277 changes: 277 additions & 0 deletions packages/app-bundle/overlay/packages/opencode/src/bus/global-replay.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,277 @@
// The global event stream's replay buffer (#1776, ported onto the lane for
// #1806 P1a). One tap on the global bus serializes each event once, at emit,
// into a frame stamped with its `<epoch>:<seq>` id; the buffer keeps recent
// frames (one buffer shared by every subscriber, bounded by bytes and age) and
// fans each frame out to the live subscriptions. A connection that resumes
// from a servable cursor gets the frames after it, then live frames; any other
// cursor gets an explicit reset.
//
// Lane adaptations vs main (#1806):
// - a frame keeps a reference to the emitted event (`event`): this engine
// serves TWO wire shapes off the one buffer (the global route serializes
// `data` verbatim; the instance /event route projects the payload view from
// the live object), so subscribers never re-parse the wire serialization;
// - a cursor that is not `<epoch>:<seq>` resolves against retained frames by
// payload id instead of resetting outright: the lane's client tracks
// payload ids (#1264), and wire compatibility requires its cursors to keep
// being served — resolved when the retained history covers them, reset as
// evicted when it does not;
// - main's patch-body wire cut (#1778's wireEvent, 06eac851) is P3 and is
// deliberately NOT ported — the serialized frame is the whole event.
import { GlobalBus, type GlobalEvent } from "./global"

export type Frame = {
/** The SSE id: `<epoch>:<seq>`. */
readonly id: string
readonly seq: number
/** The serialized global event — the SSE `data`. */
readonly data: string
/** The emitted event itself — the buffer's in-process view (see header). */
readonly event: GlobalEvent
/** UTF-8 size of `data`, charged against the byte bound. */
readonly bytes: number
/** Push time (ms), checked against the age bound. */
readonly time: number
/** Push order; unlike `seq` it is unique per frame (a sync copy shares its source's seq). */
readonly ordinal: number
/**
* A sync copy sharing the sequence of the frame it mirrors (#1778). Sent only
* to subscribers that request sync copies; dropping it leaves no hole in a
* subscriber's sequence, because its source carries the same one.
*/
readonly syncCopy: boolean
}

export type Bounds = {
/** Retained frame bytes; the oldest frames are evicted past it. */
readonly maxBytes: number
/** Retained frame age in milliseconds; older frames are evicted. */
readonly maxAgeMs: number
}

/** About 32 MB and 5 minutes of history: covers link blips and window reloads. */
export const DEFAULT_BOUNDS: Bounds = { maxBytes: 32 * 1024 * 1024, maxAgeMs: 5 * 60_000 }

export const BOUNDS_ENV = {
maxBytes: "OPENCODE_GLOBAL_EVENT_REPLAY_MAX_BYTES",
maxAgeMs: "OPENCODE_GLOBAL_EVENT_REPLAY_MAX_AGE_MS",
} as const

/** The buffer bounds from the environment; an unset or invalid value keeps its default. */
export function boundsFromEnv(env: Record<string, string | undefined>): Bounds {
const read = (key: string, fallback: number) => {
const raw = env[key]?.trim()
if (!raw || !/^\d+$/.test(raw)) return fallback
return Number(raw)
}
return {
maxBytes: read(BOUNDS_ENV.maxBytes, DEFAULT_BOUNDS.maxBytes),
maxAgeMs: read(BOUNDS_ENV.maxAgeMs, DEFAULT_BOUNDS.maxAgeMs),
}
}

export function formatID(epoch: string, seq: number) {
return `${epoch}:${seq}`
}

/** Parse an `<epoch>:<seq>` cursor; `undefined` when malformed. */
export function parseCursor(raw: string): { epoch: string; seq: number } | undefined {
const match = /^([A-Za-z0-9_-]+):(\d+)$/.exec(raw.trim())
if (!match) return undefined
return { epoch: match[1], seq: Number(match[2]) }
}

/** Why a cursor cannot be served. */
export type ResetReason = "malformed" | "epoch" | "ahead" | "evicted"

export type Subscription = {
/** The connection's starting position: the sequence its first frame follows. */
readonly position: number
/** Set when the cursor could not be served: the connection must start with a reset frame. */
readonly reset: ResetReason | undefined
/** Frames after the cursor, in order — delivered before any live frame. */
readonly replay: readonly Frame[]
/** Begin delivering live frames (any that arrived since subscribing first, minus duplicates of `replay`). */
start(deliver: (frame: Frame) => void): void
/** Unregister. Idempotent. */
close(): void
}

export class ReplayBuffer {
readonly #listeners = new Set<(frame: Frame) => void>()
// Retained frames are #frames[#head..]; evicted slots before #head are
// compacted away in batches so eviction stays O(1) amortized.
#frames: (Frame | undefined)[] = []
#head = 0
#bytes = 0
#ordinal = 0
#current: number
// The highest sequence whose frames are no longer all retained (evicted, or
// emitted before this buffer existed). A cursor below it cannot be served.
#floor: number

readonly epoch: string
readonly bounds: Bounds
readonly #now: () => number

constructor(input: {
epoch: string
bounds?: Bounds
/** The sequence the buffer starts at (events before it were never recorded). */
start?: number
now?: () => number
}) {
this.epoch = input.epoch
this.bounds = input.bounds ?? DEFAULT_BOUNDS
this.#now = input.now ?? Date.now
this.#current = input.start ?? 0
this.#floor = this.#current
}

/** The current sequence: the last one pushed (or skipped). */
get current() {
return this.#current
}

/** Live subscriptions currently registered. */
get subscribers() {
return this.#listeners.size
}

push(seq: number, event: GlobalEvent, data: string, options: { syncCopy?: boolean } = {}): Frame {
const frame: Frame = {
id: formatID(this.epoch, seq),
seq,
data,
event,
bytes: Buffer.byteLength(data),
time: this.#now(),
ordinal: ++this.#ordinal,
syncCopy: options.syncCopy === true,
}
this.#current = Math.max(this.#current, seq)
this.#frames.push(frame)
this.#bytes += frame.bytes
for (const listener of this.#listeners) listener(frame)
this.#evict()
return frame
}

/** Advance past a sequence that produced no frame (the hole stays visible to clients). */
skip(seq: number) {
this.#current = Math.max(this.#current, seq)
}

/**
* Subscribe, optionally resuming after `cursor`. The live listener is
* registered BEFORE the buffer is read, and live frames already covered by
* the replay are dropped, so nothing published around the reconnect is
* missed or delivered twice. Sync copies reach the subscription only when it
* asks for them (`sync`), in its replay and its live frames alike.
*/
subscribe(cursor?: string, options: { sync?: boolean } = {}): Subscription {
const wanted = options.sync === true ? () => true : (frame: Frame) => !frame.syncCopy
let deliver: ((frame: Frame) => void) | undefined
let pending: Frame[] = []
const listener = (frame: Frame) => {
if (!wanted(frame)) return
if (deliver) deliver(frame)
else pending.push(frame)
}
this.#listeners.add(listener)

this.#evict()
const watermark = this.#ordinal
const resolved = this.#resolve(cursor)
const replay = resolved.reset
? []
: this.#retained().filter((frame) => frame.seq > resolved.position && wanted(frame))

return {
...resolved,
replay,
start: (fn) => {
const queued = pending
pending = []
for (const frame of queued) if (frame.ordinal > watermark) fn(frame)
deliver = fn
},
close: () => {
this.#listeners.delete(listener)
pending = []
},
}
}

#resolve(cursor: string | undefined): { position: number; reset: ResetReason | undefined } {
if (cursor === undefined) return { position: this.#current, reset: undefined }
const parsed = parseCursor(cursor)
if (parsed) {
const reset: ResetReason | undefined =
parsed.epoch !== this.epoch
? "epoch"
: parsed.seq > this.#current
? "ahead"
: parsed.seq < this.#floor
? "evicted"
: undefined
return reset ? { position: this.#current, reset } : { position: parsed.seq, reset: undefined }
}
// Legacy vocabulary (the lane's client sends the last payload id it
// applied, #1264): resolve to the newest retained frame at or behind the
// cursor and serve the frames after it; a cursor no retained frame covers
// predates retention and resets.
let resolved: Frame | undefined
for (const frame of this.#retained()) {
const id = frame.event.payload?.id
if (typeof id === "string" && id <= cursor) resolved = frame
}
if (resolved) return { position: resolved.seq, reset: undefined }
return { position: this.#current, reset: "evicted" }
}

#retained(): Frame[] {
return this.#frames.slice(this.#head) as Frame[]
}

// Oldest-first, until retained bytes and the oldest frame's age are both within bounds.
#evict() {
const oldest = this.#now() - this.bounds.maxAgeMs
while (this.#head < this.#frames.length) {
const frame = this.#frames[this.#head]!
if (this.#bytes <= this.bounds.maxBytes && frame.time >= oldest) break
this.#frames[this.#head++] = undefined
this.#bytes -= frame.bytes
this.#floor = Math.max(this.#floor, frame.seq)
}
if (this.#head > 1024 && this.#head * 2 > this.#frames.length) {
this.#frames = this.#frames.slice(this.#head)
this.#head = 0
}
}
}

export const GlobalEventReplay = new ReplayBuffer({
epoch: GlobalBus.epoch,
bounds: boundsFromEnv(process.env),
start: GlobalBus.seq,
})

GlobalBus.tap((event, seq) => {
// A sync copy whose sequence was already handed out mirrors its source. One
// that consumed a fresh sequence (no source framed nearby) is a frame of its
// own and reaches every subscriber, so filtering never opens a hole.
const syncCopy = event.payload?.type === "sync" && seq <= GlobalEventReplay.current
let data: string
try {
// serialized once: every subscriber, live or replayed, gets this same frame
data = JSON.stringify(event)
} catch (error) {
// An unserializable event cannot be framed; its sequence becomes a hole the
// client detects (heartbeats carry the current sequence), never a dead stream.
console.error("global event not serializable", { seq, type: event.payload?.type, error })
GlobalEventReplay.skip(seq)
return
}
GlobalEventReplay.push(seq, event, data, { syncCopy })
})
Loading
Loading