feat: split a step into model-call, per-tool, and seal activities.

.github/workflows/test-fork.yml

@@ -0,0 +1,70 @@
+# The upstream `test` workflow asks for Blacksmith runners (`blacksmith-4vcpu-*`), which this fork
+# cannot schedule: every run of it here, on every branch including `dev` and dependabot PRs, is
+# cancelled with no runner ever assigned. So nothing in this fork has had a CI signal.
+#
+# This runs typecheck and the affected packages' unit tests on GitHub's own runners, so a change
+# made here is checked by something other than the author's laptop.
+#
+# Deliberately narrower than upstream's, in two ways worth being explicit about:
+#   - Linux only, no Windows matrix, no e2e, no generated-client or HttpApi gates. Those belong to
+#     upstream's runners, and pretending to cover them here would be worse than not claiming to.
+#   - `bun turbo test` is NOT run, because `@opencode-ai/app` currently fails one locale-detection
+#     assertion (`detectDesktopNativeLocale(["pa-PK"])` returns "en", not "pa") that depends on the
+#     ICU data in the bun build. It reproduces on the untouched base branch, so a whole-monorepo
+#     test job would be red on arrival and worth nothing. Typecheck still covers every package.
+name: test (fork)
+
+on:
+  pull_request:
+  workflow_dispatch:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+env:
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true
+
+jobs:
+  unit:
+    name: unit + typecheck (linux)
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        shell: bash
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+
+      - name: Setup Node
+        uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
+        with:
+          node-version: "24"
+
+      - name: Setup Bun
+        uses: ./.github/actions/setup-bun
+
+      # Some suites shell out to git, which refuses to run without an identity.
+      - name: Configure git identity
+        run: |
+          git config --global user.email "bot@opencode.ai"
+          git config --global user.name "opencode"
+
+      # Concurrency capped: a full-width turbo run puts enough `tsgo` processes on a hosted runner
+      # at once that several are killed for memory, which reads as a code failure and is not one.
+      - name: Typecheck
+        timeout-minutes: 20
+        run: bun turbo typecheck --concurrency=2
+
+      - name: Unit tests (core)
+        timeout-minutes: 30
+        working-directory: packages/core
+        run: bun test
+
+      - name: Unit tests (temporal)
+        timeout-minutes: 15
+        working-directory: packages/temporal
+        run: bun test

packages/core/src/event.ts

@@ -1,11 +1,13 @@
 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 } from "effect"
+import { 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"
@@ -174,6 +176,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) =>
@@ -186,7 +209,8 @@ export const layerWith = (options?: LayerOptions) =>
         typed: new Map<string, PubSub.PubSub<Payload>>(),
       }
       const projectors = new Map<string, Subscriber[]>()
-      // TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads.
+      // TODO: Bind durable projectors to exact type+version before supporting incompatible
+      // historical payloads.
       const listeners = new Array<Subscriber>()
       const { db } = yield* Database.Service
 
@@ -619,7 +643,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,
             )
@@ -660,3 +695,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

@@ -79,4 +79,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/src/session/execution/conformance.ts

@@ -24,6 +24,8 @@ import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { SessionExecution } from "@opencode-ai/core/session/execution"
 import { SessionExecutionLocal } from "@opencode-ai/core/session/execution/local"
 import { SessionRunnerModel, ModelNotSelectedError } from "@opencode-ai/core/session/runner/model"
+import { ApplicationTools } from "@opencode-ai/core/tool/application-tools"
+import { Tool } from "@opencode-ai/core/tool/tool"
 import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
 import { SessionTable } from "@opencode-ai/core/session/sql"
 import { SessionStore } from "@opencode-ai/core/session/store"
@@ -39,7 +41,7 @@ import { Auth } from "@opencode-ai/llm/route"
 import { describe, expect } from "bun:test"
 import { realpathSync } from "node:fs"
 import { tmpdir } from "node:os"
-import { Cause, Context, DateTime, Effect, Exit, Layer, Stream } from "effect"
+import { Cause, Context, DateTime, Effect, Exit, Layer, Schema, Stream } from "effect"
 import { testEffect } from "../../testing/effect"
 
 // The per-location service build resolves the session directory on disk, so it must exist.
@@ -77,13 +79,66 @@ const countingModel = () => {
   return { requests, stream }
 }
 
+// Asks for one tool on the first request and answers on the second, which is the smallest turn that
+// makes a driver dispatch a tool and come back to the model with its result.
+const toolCallingModel = () => {
+  const requests: number[] = []
+  const stream: LLMClientShape["stream"] = () => {
+    requests.push(1)
+    if (requests.length > 1)
+      return Stream.fromIterable([
+        LLMEvent.stepStart({ index: 0 }),
+        LLMEvent.stepFinish({ index: 0, reason: "stop" }),
+      ])
+    return Stream.fromIterable([
+      LLMEvent.stepStart({ index: 0 }),
+      LLMEvent.toolCall({ id: "call_contract", name: "contract_probe", input: {} }),
+      LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
+    ])
+  }
+  return { requests, stream }
+}
+
+// Counts its own executions, so "ran exactly once" is checked against the tool rather than only
+// against the projection.
+const probeTool = (ran: { count: number }) =>
+  Tool.make({
+    description: "contract probe",
+    input: Schema.Struct({}),
+    output: Schema.String,
+    execute: () =>
+      Effect.sync(() => {
+        ran.count += 1
+        return "probed"
+      }),
+  })
+
+// A tool that never finishes, so a turn can be interrupted with one in flight.
+const hangingTool = () =>
+  Tool.make({
+    description: "contract probe that never finishes",
+    input: Schema.Struct({}),
+    output: Schema.String,
+    execute: () => Effect.never,
+  })
+
+const toolParts = (messages: ReadonlyArray<SessionMessage.Message>) =>
+  messages.flatMap((message) =>
+    message.type === "assistant"
+      ? message.content.filter((part): part is SessionMessage.AssistantTool => part.type === "tool")
+      : [],
+  )
+
 // The executor under test, built as its own graph over the shared database file (the same way the
 // serve process builds it), with the model/LLM mocked. Any SessionExecution node with the standard
 // dependency set (the local coordinator, the Temporal driver) plugs in here.
 export const makeExecutionFor =
   (node: typeof SessionExecutionLocal.node) =>
   (stream: LLMClientShape["stream"], models = okModels) =>
-    AppNodeBuilder.build(node, [
+    // The tool registry is named in the group so a scenario can register a probe into the same
+    // graph the driver runs on. One graph, one instance: what the test registers is what the
+    // dispatch finds.
+    AppNodeBuilder.build(LayerNode.group([node, ApplicationTools.node]), [
       [LayerNodePlatform.llmClient, mockClient(stream)],
       [PermissionV2.node, permission],
       [ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig],
@@ -273,5 +328,81 @@ export const runContract = (label: string, makeExec: ReturnType<typeof makeExecu
         60000,
       )
     }
+
+    {
+      const { requests, stream } = toolCallingModel()
+      const ran = { count: 0 }
+      const sessionID = SessionV2.ID.make(`ses_${slug}_tool`)
+      it.live("a tool call runs once and the turn goes back to the model with its result", () =>
+        withIdleOverride(
+          Effect.gen(function* () {
+            yield* seedSession(sessionID)
+            yield* seedPrompt(sessionID)
+            const graph = yield* Layer.build(makeExec(stream))
+            yield* Context.get(graph, ApplicationTools.Service).register({
+              contract_probe: probeTool(ran),
+            })
+            const exec = Context.get(graph, SessionExecution.Service)
+            yield* exec.wake(sessionID)
+            // Waiting on the turn, not on the step: the first step's assistant is completed the
+            // moment the step is sealed, while the follow-up provider call is still to come.
+            yield* until(exec.active, (active) => !active.has(sessionID))
+            // Where the drivers could drift: a step's model call, its tool and its close are one
+            // activity in one mode and three in the other, so the tool has to be dispatched once,
+            // settled in the log, and answered by a second provider turn either way.
+            expect(ran.count).toBe(1)
+            expect(requests).toHaveLength(2)
+            const store = yield* SessionStore.Service
+            const context = yield* store.context(sessionID)
+            const parts = context.flatMap((message) =>
+              message.type === "assistant" ? message.content : [],
+            )
+            const call = parts.findLast(
+              (part): part is SessionMessage.AssistantTool =>
+                part.type === "tool" && part.id === "call_contract",
+            )
+            expect(call?.state.status).toBe("completed")
+          }),
+        ),
+        60000,
+      )
+    }
+
+    {
+      const { stream } = toolCallingModel()
+      const sessionID = SessionV2.ID.make(`ses_${slug}_interrupt_tool`)
+      it.live("closes a tool call the interrupt cut short", () =>
+        withIdleOverride(
+          Effect.gen(function* () {
+            yield* seedSession(sessionID)
+            yield* seedPrompt(sessionID)
+            const graph = yield* Layer.build(makeExec(stream))
+            yield* Context.get(graph, ApplicationTools.Service).register({
+              contract_probe: hangingTool(),
+            })
+            const exec = Context.get(graph, SessionExecution.Service)
+            const store = yield* SessionStore.Service
+            yield* exec.wake(sessionID)
+            yield* until(store.context(sessionID), (context) =>
+              toolParts(context).some((part) => part.state.status === "running"),
+            )
+
+            yield* exec.interrupt(sessionID)
+
+            // A call left running is not a cosmetic loose end: the transcript shows a tool still
+            // going, and the next turn has to reconstruct what happened to it. A whole step closes
+            // the tools it opened on its way out, and a step whose calls are their own units of
+            // work has to end up in the same place.
+            yield* until(store.context(sessionID), (context) =>
+              toolParts(context).every((part) => part.state.status !== "running"),
+            )
+            const closed = toolParts(yield* store.context(sessionID))
+            expect(closed.map((part) => part.state.status)).toEqual(["error"])
+            yield* until(exec.active, (active) => !active.has(sessionID))
+          }),
+        ),
+        60000,
+      )
+    }
   })
 }