feat(core): Let a durable event tail re-read on a tick.

packages/core/src/event.ts

@@ -1,11 +1,12 @@
 export * as EventV2 from "./event"
 
-import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect"
+import { Cause, Context, Duration, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
 import { and, asc, eq, gt, inArray } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
+import { Flag } from "./flag/flag"
 import { Location } from "./location"
 import { makeGlobalNode } from "./effect/app-node"
 import { isDeepStrictEqual } from "node:util"
@@ -165,6 +166,27 @@ export const allBounded = (events: Interface, capacity: number) =>
 
 export interface LayerOptions {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
+  /** How often a durable tail re-reads on its own, on top of the in-process wake. The wake only
+   * fires for commits made in THIS process, so without a tick a subscriber cannot see events
+   * another process appended: a standalone worker's whole turn is invisible to a tail on the
+   * HTTP server.
+   * Zero (the default) relies on the wake alone, which is the whole story for a deployment that
+   * runs in one process. */
+  readonly livePollInterval?: Duration.Input
+}
+
+/** Chosen to be well under what a person notices in a transcript while staying one cheap indexed
+ * read per subscribed session. In-process commits still wake instantly; this only catches what the
+ * wake cannot see, so it is worth its cost only where another process writes: see `pollingNode`. */
+const DEFAULT_LIVE_POLL = Duration.seconds(1)
+
+// An operator's override, in milliseconds, for either node. Read at layer build rather than at
+// import, so a test or a CLI that sets it late still gets it.
+const configuredPoll = (fallback: Duration.Input): Duration.Input => {
+  const raw = Flag.OPENCODE_EVENT_POLL_MS
+  if (raw === undefined) return fallback
+  const millis = Number(raw)
+  return Number.isFinite(millis) && millis >= 0 ? millis : fallback
 }
 
 export const layerWith = (options?: LayerOptions) =>
@@ -595,7 +617,18 @@ export const layerWith = (options?: LayerOptions) =>
               ),
             )
             const historical = yield* read
-            const live = Stream.fromSubscription(wakes).pipe(
+            // Wake on either an in-process commit or the tick. A tick that finds nothing new reads
+            // zero rows and emits nothing, so an idle subscriber costs one indexed query per
+            // period.
+            const pollInterval = configuredPoll(options?.livePollInterval ?? Duration.zero)
+            const woken = Stream.fromSubscription(wakes)
+            // haltStrategy "left" keeps the wake stream as what ends the tail. The tick never ends,
+            // so the default ("both") would leave a subscriber hanging past the layer's own
+            // PubSub.shutdown, still reading from a database being torn down.
+            const source = Duration.isZero(Duration.fromInputUnsafe(pollInterval))
+              ? woken
+              : woken.pipe(Stream.merge(Stream.tick(pollInterval), { haltStrategy: "left" }))
+            const live = source.pipe(
               Stream.mapEffect(() => read),
               Stream.flattenIterable,
             )
@@ -636,3 +669,13 @@ export const layerWith = (options?: LayerOptions) =>
 
 const layer = layerWith()
 export const node = makeGlobalNode({ service: Service, layer: layer, deps: [Database.node] })
+
+// For a deployment where the process serving subscribers is not the only one appending: a serve
+// process driving sessions on standalone workers. The wake is published in-process, so without the
+// tick a worker's whole turn is invisible to a tail on the server. Composition roots that know they
+// are in that shape swap this in for `node`; everything else keeps the wake alone.
+export const pollingNode = makeGlobalNode({
+  service: Service,
+  layer: layerWith({ livePollInterval: DEFAULT_LIVE_POLL }),
+  deps: [Database.node],
+})

packages/core/src/flag/flag.ts

@@ -75,4 +75,10 @@ export const Flag = {
   get OPENCODE_CLIENT() {
     return process.env["OPENCODE_CLIENT"] ?? "cli"
   },
+  // Milliseconds between a live event tail's own re-reads, `0` to rely on the in-process wake
+  // alone. Only a deployment where another process appends to the log needs it, so it is off unless
+  // the composition root asks for it; this overrides either way.
+  get OPENCODE_EVENT_POLL_MS() {
+    return process.env["OPENCODE_EVENT_POLL_MS"]
+  },
 }

packages/core/test/event.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect"
+import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Scope, Stream } from "effect"
 import { EventV2 } from "@opencode-ai/core/event"
 import { Event } from "@opencode-ai/schema/event"
 import { Session } from "@opencode-ai/schema/session"
@@ -1097,6 +1097,55 @@ describe("EventV2", () => {
     }),
   )
 
+  it.live("tails another writer's events only when the layer asked for a tick", () =>
+    Effect.gen(function* () {
+      // A second service over the SAME database with its OWN pubsub: exactly what a standalone
+      // worker is to the HTTP server. Its commits cannot reach the reader's wake map, so a tail
+      // that only listens for wakes sits there forever.
+      const tailSees = (options: EventV2.LayerOptions) =>
+        Effect.gen(function* () {
+          const reader = yield* EventV2.Service.pipe(Effect.provide(EventV2.layerWith(options)))
+          const writer = yield* EventV2.Service.pipe(Effect.provide(EventV2.layerWith()))
+          const aggregateID = Session.ID.create()
+          const tail = yield* reader
+            .durable({ aggregateID, after: -1 })
+            .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+          yield* Effect.sleep("50 millis")
+          yield* writer.publish(DurableMessage, durableData(aggregateID, "from-elsewhere"))
+          const collected = yield* Fiber.join(tail).pipe(Effect.timeoutOption("1 second"))
+          return Option.isSome(collected)
+        })
+
+      expect(yield* tailSees({ livePollInterval: "50 millis" })).toBe(true)
+      // Off unless asked for. One process wakes its own subscribers, so a tick there is a query per
+      // second per subscribed session for events that cannot exist.
+      expect(yield* tailSees({})).toBe(false)
+    }),
+  )
+
+  it.live("ends a durable tail when the event layer is released under it", () =>
+    Effect.gen(function* () {
+      const aggregateID = Session.ID.create()
+      // The layer needs its own closeable scope, and the consumer has to be forked into a scope that
+      // OUTLIVES it. Fork into the test's own scope and closing that scope interrupts the consumer
+      // before the finalizer runs, so a hung tail and a killed one look the same.
+      const outer = yield* Effect.scope
+      const layerScope = yield* Scope.make()
+      const ctx = yield* Layer.buildWithScope(EventV2.layerWith(), layerScope)
+      const events = yield* EventV2.Service.pipe(Effect.provideContext(ctx as never))
+      yield* events.publish(DurableMessage, durableData(aggregateID, "seed"))
+
+      const tail = yield* events.durable({ aggregateID, after: -1 }).pipe(Stream.runDrain, Effect.forkIn(outer))
+      yield* Effect.sleep("150 millis")
+      yield* Scope.close(layerScope, Exit.void)
+
+      // The tick never ends on its own, so it must not become what holds the tail open: a subscriber
+      // would outlive the layer and keep reading a database being torn down.
+      const settled = yield* Fiber.await(tail).pipe(Effect.timeout("5 seconds"), Effect.exit)
+      expect(Exit.isSuccess(settled)).toBe(true)
+    }),
+  )
+
   it.effect("remove clears durable event sequence", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service