From e13e2dbe0afca9c194aad9de8b3066b3cf13eff0 Mon Sep 17 00:00:00 2001 From: Anmolpreet Kandola Date: Mon, 6 Jul 2026 20:07:41 -0400 Subject: [PATCH] feat: emit turn_start event carrying provider and model Adds an additive turn_start WireEvent emitted once per turn after the model is resolved, so consumers can attribute a turn to its provider and model without inferring from a later provider_switched/provider_error (which only fire on a switch or failure). Model resolution degrades to blank rather than blocking the turn. Handles the new arm in the runtime-client snapshot reducer's metadata group. Adds a protocol type test. --- packages/protocol/src/wire.test.ts | 22 ++++++++++++++++++++++ packages/protocol/src/wire.ts | 13 +++++++++++++ packages/runtime-client/src/snapshot.ts | 8 +++++--- packages/runtime/src/session/chat.ts | 21 +++++++++++++++++++++ packages/runtime/src/turn/turn-session.ts | 4 ++++ 5 files changed, 65 insertions(+), 3 deletions(-) create mode 100644 packages/protocol/src/wire.test.ts diff --git a/packages/protocol/src/wire.test.ts b/packages/protocol/src/wire.test.ts new file mode 100644 index 000000000..42e5aaffc --- /dev/null +++ b/packages/protocol/src/wire.test.ts @@ -0,0 +1,22 @@ +import { expectTypeOf, test } from "vitest"; +import type { WireEvent } from "./index"; + +test("the protocol index re-exports the turn_start WireEvent arm", () => { + const turnStart: WireEvent = { + type: "turn_start", + data: { provider: "anthropic", model: "claude-sonnet-4-6" }, + }; + + expectTypeOf(turnStart).toMatchTypeOf(); + // `type` is the discriminant → narrows `data` to the turn_start shape. + if (turnStart.type === "turn_start") { + expectTypeOf(turnStart.data).toEqualTypeOf<{ + provider: string; + model: string; + }>(); + } + + // @ts-expect-error — `type` is the discriminant; unknown values are not assignable + const bad: WireEvent = { type: "not_an_event", data: null }; + void bad; +}); diff --git a/packages/protocol/src/wire.ts b/packages/protocol/src/wire.ts index d7990405e..b857c1a82 100644 --- a/packages/protocol/src/wire.ts +++ b/packages/protocol/src/wire.ts @@ -23,6 +23,7 @@ import type { ProviderError } from "./provider-error"; * be served — the client must treat the stream as fresh (refetch history, * rebuild from this snapshot) instead of splicing. * - `user` — a user message was added (by any client); `nonce` echoes the sender's. + * - `turn_start` — the provider + resolved model this turn runs against. * - `text` / `thinking` — assistant output deltas. * - `tool_start` / `tool_end` — tool activity within the turn. * - `usage` — normalized token usage for the turn (when the provider reports it), @@ -100,6 +101,18 @@ export type WireEvent = mentions?: { userId: string; name?: string }[]; }; } + | { + /** + * Turn metadata, emitted once at turn start: the provider and the + * concrete model this turn resolved to after applying any per-turn pin + * (not the raw pin, which may be absent = inherit). Lets a client + * attribute a turn's output to a specific provider + model without + * inferring it from a later `provider_switched` / `provider_error` (which + * only fire on a switch or a failure). + */ + type: "turn_start"; + data: { provider: string; model: string }; + } | { type: "text"; data: string } | { type: "thinking"; data: string } | { type: "tool_start"; data: { name: string; args: unknown } } diff --git a/packages/runtime-client/src/snapshot.ts b/packages/runtime-client/src/snapshot.ts index 74533cce7..61cc3b871 100644 --- a/packages/runtime-client/src/snapshot.ts +++ b/packages/runtime-client/src/snapshot.ts @@ -146,11 +146,13 @@ export function reduceSnapshot( // clears the in-flight snapshot — otherwise a late subscriber's `sync` // would report the turn as still running forever. return { running: false, partial: "", seq }; + case "turn_start": case "provider_switched": case "context_compacted": - // Boundary markers (a mid-session provider switch / a proactive context - // compaction), not turn progress — published while a turn is live, so - // leave running/partial untouched. + // Boundary / metadata markers (the turn's provider+model announcement, a + // mid-session provider switch, a proactive context compaction), not turn + // progress — published while a turn is live, so leave running/partial + // untouched. return { running: prev.running, partial: prev.partial, diff --git a/packages/runtime/src/session/chat.ts b/packages/runtime/src/session/chat.ts index a957fa15e..f5c0dcd1b 100644 --- a/packages/runtime/src/session/chat.ts +++ b/packages/runtime/src/session/chat.ts @@ -149,6 +149,19 @@ export async function runTurn( return; } + // turn_start metadata: the provider + concrete model this turn resolves to + // under its pin, for observers that attribute a turn's output to a model. + // Best-effort — getConversation already validated the pin, so this resolves; + // a resolve failure must never block the turn, so it degrades to blank. + const startProvider = pin?.provider ?? activeProvider(); + let startModel = ""; + try { + startModel = + (resolveModel(pin?.model, pin?.provider) as { id?: string }).id ?? ""; + } catch { + /* leave blank — the turn still runs; provider still rides later frames */ + } + // Two layers of serialization: per-conversation ordering (conv.queue) AND // the per-workdir lock — every conversation in this runtime shares ONE // workspaceDir, so a routine's turn and a user chat queue instead of @@ -177,6 +190,14 @@ export async function runTurn( displayText, mentions, ); + // turn_start rides right after the user frame (metadata for the live turn), + // matching the cloud executor where the concrete model is only known then. + if (startProvider) + publish(id, { + type: "turn_start", + data: { provider: startProvider, model: startModel }, + turnId, + }); return withWorkdirLock(config.workspaceDir, () => execTurn(conv, id, turnId, text, recorded, pin, acting), ); diff --git a/packages/runtime/src/turn/turn-session.ts b/packages/runtime/src/turn/turn-session.ts index 3c97f1fbe..18a781788 100644 --- a/packages/runtime/src/turn/turn-session.ts +++ b/packages/runtime/src/turn/turn-session.ts @@ -189,6 +189,10 @@ export async function runPiTurn( console.log( `[turn] provider=${provider} model=${m.id} baseUrl=${m.baseUrl}`, ); + // Announce the turn's resolved provider + model to observers (turn_start). + // Emitted here — the first point the concrete model is known (after applying + // the pin) — so it carries the real model id, not the raw pin. + emit({ type: "turn_start", data: { provider, model: m.id ?? "" } }); // Effort → pi's thinking level. The turn's pin (the host bakes the agent's // saved effort into it) wins; if none and the model can reason, default to // medium so a "thinking" model actually reasons (pi enables reasoning only