diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts index 1bc7d3938..770f936bc 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.test.ts @@ -136,6 +136,7 @@ interface FakeCaptures { readonly mimeType: string; }>; readonly signal: AbortSignal | undefined; + readonly rlmQuiescenceToken: string | undefined; }>; readonly followUps: Array<{ readonly text: string; readonly imageCount: number }>; followUpFailure: boolean; @@ -271,6 +272,13 @@ interface FakeCaptures { | undefined; sideQuestionFailure: boolean; promptObserved: Queue.Queue | undefined; + rlmQuiescenceAvailable: boolean; + rlmQuiescenceRelease: Deferred.Deferred | undefined; + readonly rlmQuiescenceCalls: Array; + rlmQuiescenceObserved: Queue.Queue | undefined; + rlmQuiescenceFailure: boolean; + rlmConnectionGeneration: number; + rlmQuiescenceUsage: typeof usage | undefined; queue: Queue.Queue | undefined; startupEvents: Array; } @@ -385,6 +393,13 @@ function makeCaptures(): FakeCaptures { sideQuestionRelease: undefined, sideQuestionFailure: false, promptObserved: undefined, + rlmQuiescenceAvailable: false, + rlmQuiescenceRelease: undefined, + rlmQuiescenceCalls: [], + rlmQuiescenceObserved: undefined, + rlmQuiescenceFailure: false, + rlmConnectionGeneration: 0, + rlmQuiescenceUsage: undefined, queue: undefined, startupEvents: [], }; @@ -698,15 +713,45 @@ function fakeRuntimeFactory( return captures.agentDepth; }), events: Stream.fromQueue(queue), + rlmQuiescenceAvailable: captures.rlmQuiescenceAvailable, + waitForRlmQuiescence: (token) => + Effect.gen(function* () { + captures.rlmQuiescenceCalls.push(token); + if (captures.rlmQuiescenceObserved !== undefined) { + yield* Queue.offer(captures.rlmQuiescenceObserved, token); + } + if (captures.rlmQuiescenceRelease !== undefined) { + yield* Deferred.await(captures.rlmQuiescenceRelease); + } + if (captures.rlmQuiescenceFailure) { + return yield* new PrimeAgentDaemonSessionRuntimeError({ + operation: "rlm-quiescence", + reason: "request-failed", + detail: "quiescence failed", + }); + } + yield* Queue.offer(queue, { + _tag: "RlmQuiesced", + token, + connectionGeneration: captures.rlmConnectionGeneration, + ...(captures.rlmQuiescenceUsage === undefined + ? {} + : { usage: captures.rlmQuiescenceUsage }), + }); + }), + isRlmQuiescenceGenerationCurrent: (generation) => + generation === captures.rlmConnectionGeneration, prompt: (prompt) => - Effect.sync(() => { + Effect.gen(function* () { captures.order.push("prompt"); captures.prompts.push({ text: prompt.text, images: prompt.images ?? [], signal: prompt.signal, + rlmQuiescenceToken: prompt.rlmQuiescenceToken, }); - }).pipe(Effect.andThen(Queue.offer(promptObserved, undefined)), Effect.asVoid), + yield* Queue.offer(promptObserved, undefined); + }), steer: (steer) => Effect.sync(() => { captures.order.push("steer"); @@ -4616,6 +4661,421 @@ describe("PrimeAgentDaemonAdapter", () => { ).pipe(Effect.provide(testLayer)), ); + it.effect( + "keeps asynchronous child continuations attached through authoritative quiescence", + () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.rlmQuiescenceAvailable = true; + captures.rlmQuiescenceRelease = yield* Deferred.make(); + captures.rlmQuiescenceUsage = { + inputTokens: 101, + outputTokens: 37, + cachedInputTokens: 503, + cacheWriteTokens: 11, + totalTokens: 652, + totalCostUsd: 0.321, + }; + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + const running = yield* adapter + .sendTurn({ threadId, input: "wait for asynchronous children" }) + .pipe(Effect.forkChild); + const started = yield* awaitObservedType(subscription.observed, "turn.started"); + yield* Queue.take(captures.promptObserved!); + + // The child roster can arrive just after the root model boundary. The + // native barrier, rather than local timing, keeps the Pylon turn open. + yield* offer(captures, { _tag: "RunCompleted", messages: [] }); + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-first", label: "first", status: "running" }, + }); + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-second", label: "second", status: "running" }, + }); + yield* offer(captures, { + _tag: "SessionInfoChanged", + name: "children admitted barrier", + }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + expect(running.pollUnsafe()).toBeUndefined(); + expect(subscription.events.some((event) => event.type === "turn.completed")).toBe(false); + + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-second", label: "second", status: "done" }, + }); + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-first", label: "first", status: "done" }, + }); + yield* offer(captures, { _tag: "RunStarted" }); + const finalMessage = assistantMessage("the asynchronous parent final response"); + yield* offer(captures, { _tag: "MessageStarted", message: finalMessage }); + yield* offer(captures, { _tag: "MessageCompleted", message: finalMessage }); + yield* offer(captures, { _tag: "RunCompleted", messages: [finalMessage] }); + yield* offer(captures, { + _tag: "SessionInfoChanged", + name: "parent continuation barrier", + }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + expect(running.pollUnsafe()).toBeUndefined(); + expect(subscription.events.some((event) => event.type === "turn.completed")).toBe(false); + + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + const result = yield* Fiber.join(running); + + expect(result.turnId).toBe(started.turnId); + const turnEvents = subscription.events.filter((event) => event.turnId === result.turnId); + expect(turnEvents.filter((event) => event.type === "turn.completed")).toHaveLength(1); + expect(turnEvents.find((event) => event.type === "turn.completed")).toMatchObject({ + payload: { + usage: { + inputTokens: 101, + outputTokens: 37, + cachedInputTokens: 503, + cacheWriteTokens: 11, + totalTokens: 652, + }, + totalCostUsd: 0.321, + }, + }); + expect( + turnEvents.some( + (event) => + event.type === "content.delta" && + event.payload.streamKind === "assistant_text" && + event.payload.delta === finalMessage.text, + ), + ).toBe(true); + expect( + turnEvents + .filter((event) => event.type === "task.completed") + .map((event) => event.payload), + ).toEqual( + expect.arrayContaining([ + expect.objectContaining({ taskId: "child-first", status: "completed" }), + expect.objectContaining({ taskId: "child-second", status: "completed" }), + ]), + ); + expect( + turnEvents.some( + (event) => + (event.type === "runtime.warning" || event.type === "runtime.error") && + typeof event.payload.detail === "object" && + event.payload.detail !== null && + "kind" in event.payload.detail && + event.payload.detail.kind === "missing-final-response", + ), + ).toBe(false); + yield* Fiber.interrupt(subscription.fiber); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("settles once when asynchronous children quiesce without a parent reply", () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.rlmQuiescenceAvailable = true; + captures.rlmQuiescenceRelease = yield* Deferred.make(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + const running = yield* adapter + .sendTurn({ threadId, input: "handle a cancelled child" }) + .pipe(Effect.forkChild); + const started = yield* awaitObservedType(subscription.observed, "turn.started"); + yield* Queue.take(captures.promptObserved!); + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-cancelled", label: "cancelled", status: "running" }, + }); + yield* offer(captures, { _tag: "RunCompleted", messages: [] }); + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-cancelled", label: "cancelled", status: "cancelled" }, + }); + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + + const result = yield* Fiber.join(running); + expect(result.turnId).toBe(started.turnId); + const turnEvents = subscription.events.filter((event) => event.turnId === result.turnId); + expect(turnEvents.filter((event) => event.type === "turn.completed")).toHaveLength(1); + expect(turnEvents.find((event) => event.type === "runtime.warning")).toMatchObject({ + payload: { + detail: { kind: "missing-final-response", outcome: "completed" }, + }, + }); + expect(turnEvents.find((event) => event.type === "task.completed")).toMatchObject({ + payload: { taskId: "child-cancelled", status: "stopped" }, + }); + yield* Fiber.interrupt(subscription.fiber); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("ignores an old quiescence marker after a steer rearms the active turn", () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.rlmQuiescenceAvailable = true; + captures.rlmQuiescenceRelease = yield* Deferred.make(); + captures.rlmQuiescenceObserved = yield* Queue.unbounded(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + const running = yield* adapter + .sendTurn({ threadId, input: "start child work" }) + .pipe(Effect.forkChild); + const started = yield* awaitObservedType(subscription.observed, "turn.started"); + const initialToken = yield* Queue.take(captures.rlmQuiescenceObserved); + yield* offer(captures, { + _tag: "RunCompleted", + messages: [assistantMessage("initial response", "toolUse")], + }); + yield* offer(captures, { _tag: "SessionInfoChanged", name: "initial boundary" }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + + const steered = yield* adapter.sendTurn({ threadId, input: "include this follow-up" }); + expect(steered.turnId).toBe(started.turnId); + const currentToken = yield* Queue.take(captures.rlmQuiescenceObserved); + expect(currentToken).not.toBe(initialToken); + + yield* offer(captures, { + _tag: "RlmQuiesced", + token: initialToken, + connectionGeneration: 0, + }); + yield* offer(captures, { _tag: "SessionInfoChanged", name: "stale marker drained" }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + expect(subscription.events.some((event) => event.type === "turn.completed")).toBe(false); + + yield* offer(captures, { _tag: "RunStarted" }); + yield* offer(captures, { + _tag: "QueueChanged", + queuedCount: 0, + steeringCount: 0, + followUpCount: 0, + }); + const finalMessage = assistantMessage("final response after steering"); + yield* offer(captures, { _tag: "MessageCompleted", message: finalMessage }); + yield* offer(captures, { _tag: "RunCompleted", messages: [finalMessage] }); + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + const result = yield* Fiber.join(running); + + expect(result.turnId).toBe(started.turnId); + const turnEvents = subscription.events.filter((event) => event.turnId === result.turnId); + expect(turnEvents.filter((event) => event.type === "turn.completed")).toHaveLength(1); + expect( + turnEvents.some( + (event) => + event.type === "content.delta" && + event.payload.streamKind === "assistant_text" && + event.payload.delta === finalMessage.text, + ), + ).toBe(true); + yield* Fiber.interrupt(subscription.fiber); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("keeps queue-clear settlement behind its marker and ignores it in the next turn", () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.rlmQuiescenceAvailable = true; + captures.rlmQuiescenceRelease = yield* Deferred.make(); + captures.rlmQuiescenceObserved = yield* Queue.unbounded(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + const first = yield* adapter + .sendTurn({ threadId, input: "queue work then clear it" }) + .pipe(Effect.forkChild); + yield* awaitObservedType(subscription.observed, "turn.started"); + yield* Queue.take(captures.rlmQuiescenceObserved); + yield* offer(captures, { + _tag: "RunCompleted", + messages: [assistantMessage("waiting for queued input", "toolUse")], + }); + yield* offer(captures, { _tag: "SessionInfoChanged", name: "queue clear boundary" }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + const steered = yield* adapter.sendTurn({ threadId, input: "remove this input" }); + const staleToken = yield* Queue.take(captures.rlmQuiescenceObserved); + yield* adapter.clearSessionInputQueue!(threadId); + expect(subscription.events.some((event) => event.type === "turn.completed")).toBe(false); + + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + const firstResult = yield* Fiber.join(first); + expect(firstResult.turnId).toBe(steered.turnId); + expect( + subscription.events.filter( + (event) => event.turnId === firstResult.turnId && event.type === "turn.completed", + ), + ).toHaveLength(1); + + captures.rlmQuiescenceRelease = yield* Deferred.make(); + const second = yield* adapter + .sendTurn({ threadId, input: "a new canonical turn" }) + .pipe(Effect.forkChild); + const secondStarted = yield* awaitObservedType(subscription.observed, "turn.started"); + const secondToken = yield* Queue.take(captures.rlmQuiescenceObserved); + expect(secondToken).not.toBe(staleToken); + const secondMessage = assistantMessage("second turn final"); + yield* offer(captures, { _tag: "RunCompleted", messages: [secondMessage] }); + yield* offer(captures, { + _tag: "RlmQuiesced", + token: staleToken, + connectionGeneration: 0, + }); + yield* offer(captures, { _tag: "SessionInfoChanged", name: "old turn marker drained" }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + expect( + subscription.events.filter( + (event) => event.turnId === secondStarted.turnId && event.type === "turn.completed", + ), + ).toHaveLength(0); + + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + const secondResult = yield* Fiber.join(second); + expect(secondResult.turnId).toBe(secondStarted.turnId); + expect( + subscription.events.filter( + (event) => event.turnId === secondResult.turnId && event.type === "turn.completed", + ), + ).toHaveLength(1); + yield* Fiber.interrupt(subscription.fiber); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("fails and disposes a session whose current quiescence marker crossed reconnect", () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.rlmQuiescenceAvailable = true; + captures.rlmQuiescenceRelease = yield* Deferred.make(); + captures.rlmQuiescenceObserved = yield* Queue.unbounded(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + const running = yield* adapter + .sendTurn({ threadId, input: "reconnect during child work" }) + .pipe(Effect.forkChild); + const started = yield* awaitObservedType(subscription.observed, "turn.started"); + const token = yield* Queue.take(captures.rlmQuiescenceObserved); + yield* offer(captures, { _tag: "RunCompleted", messages: [] }); + yield* offer(captures, { _tag: "SessionInfoChanged", name: "pending reconnect" }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + + captures.rlmConnectionGeneration = 1; + yield* offer(captures, { + _tag: "RlmQuiesced", + token, + connectionGeneration: 0, + }); + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + yield* Fiber.join(running); + yield* awaitObservedType(subscription.observed, "session.exited"); + + const turnEvents = subscription.events.filter((event) => event.turnId === started.turnId); + expect(turnEvents.filter((event) => event.type === "turn.completed")).toHaveLength(1); + expect(turnEvents.find((event) => event.type === "turn.completed")).toMatchObject({ + payload: { state: "failed" }, + }); + expect(captures.disposeCount).toBe(1); + const nextError = yield* adapter + .sendTurn({ threadId, input: "must not reuse uncertain native state" }) + .pipe(Effect.flip); + expect(nextError).toMatchObject({ _tag: "ProviderAdapterSessionNotFoundError" }); + yield* Fiber.interrupt(subscription.fiber); + }), + ).pipe(Effect.provide(testLayer)), + ); + + it.effect("fails and disposes the session when its authoritative barrier rejects", () => + Effect.scoped( + Effect.gen(function* () { + const captures = makeCaptures(); + captures.rlmQuiescenceAvailable = true; + captures.rlmQuiescenceRelease = yield* Deferred.make(); + captures.rlmQuiescenceObserved = yield* Queue.unbounded(); + const adapter = yield* makePrimeAgentDaemonAdapter(decodeSettings({}), manager, { + instanceId, + runtimeFactory: fakeRuntimeFactory(captures), + }); + const subscription = yield* subscribe(adapter); + yield* adapter.startSession({ threadId, cwd: process.cwd(), runtimeMode: "full-access" }); + yield* awaitObservedType(subscription.observed, "thread.started"); + + const running = yield* adapter + .sendTurn({ threadId, input: "barrier failure with a running child" }) + .pipe(Effect.forkChild); + const started = yield* awaitObservedType(subscription.observed, "turn.started"); + yield* Queue.take(captures.rlmQuiescenceObserved); + yield* offer(captures, { + _tag: "ChildUpdated", + child: { id: "child-running", label: "running", status: "running" }, + }); + yield* offer(captures, { _tag: "RunCompleted", messages: [] }); + yield* offer(captures, { _tag: "SessionInfoChanged", name: "barrier failure pending" }); + yield* awaitObservedType(subscription.observed, "thread.metadata.updated"); + + captures.rlmQuiescenceFailure = true; + yield* Deferred.succeed(captures.rlmQuiescenceRelease, undefined); + yield* Fiber.join(running); + yield* awaitObservedType(subscription.observed, "session.exited"); + + const turnEvents = subscription.events.filter((event) => event.turnId === started.turnId); + expect(turnEvents.filter((event) => event.type === "turn.completed")).toHaveLength(1); + expect(turnEvents.find((event) => event.type === "turn.completed")).toMatchObject({ + payload: { state: "failed" }, + }); + expect(captures.disposeCount).toBe(1); + expect( + subscription.events.filter( + (event) => event.turnId === started.turnId && event.type === "content.delta", + ), + ).toHaveLength(0); + const nextError = yield* adapter + .sendTurn({ threadId, input: "must start a replacement session first" }) + .pipe(Effect.flip); + expect(nextError).toMatchObject({ _tag: "ProviderAdapterSessionNotFoundError" }); + yield* Fiber.interrupt(subscription.fiber); + }), + ).pipe(Effect.provide(testLayer)), + ); + it.effect("keeps an automatic reconnect continuation attached to the original turn", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts index 1329894d4..4fab25b72 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonAdapter.ts @@ -159,6 +159,8 @@ interface PrimeAgentDaemonActiveTurn { activeAssistantItemId: RuntimeItemId | undefined; lastAssistantHadRenderableText: boolean; runCompletionHandoffSequence: number; + terminalQuiescenceGeneration: number; + terminalQuiescenceToken: string | undefined; pendingRunCompletionHandoff: | { readonly sequence: number; @@ -454,6 +456,10 @@ function runtimeOperationError( }); } +function rlmQuiescenceToken(turnId: TurnId, generation: number): string { + return `${turnId}:${generation}`; +} + function runtimeStartError( threadId: ThreadId, error: PrimeAgentDaemonSessionRuntimeError, @@ -1224,6 +1230,10 @@ export function makePrimeAgentDaemonAdapter( Effect.gen(function* () { const pending = turn.pendingRunCompletionHandoff; if (pending === undefined || pending.sequence !== sequence) return; + // Prime 0.8's public RLM barrier is the authoritative boundary for + // descendant-triggered parent continuations. Its FIFO marker settles + // the turn; a heuristic timer must never overtake it. + if (turn.terminalQuiescenceToken !== undefined) return; // Prime emits agent_end before it checks for automatic compaction. A // long compaction must not exhaust the short reconnect handoff grace; // its terminal event or snapshot starts fresh continuation grace. @@ -1439,6 +1449,9 @@ export function makePrimeAgentDaemonAdapter( } // An idle snapshot can be the gap before RunStarted. Keep // the bounded handoff alive until the event or timeout. + } else if (turn.terminalQuiescenceToken !== undefined && authoritativeIdle) { + // The public RLM barrier, not an idle reconnect snapshot, owns + // settlement while descendant continuations may still arrive. } else if (turn.awaitingQueuedRun && authoritativeIdle) { const explicitClear = context.inputQueueClearPending; context.inputQueueClearPending = false; @@ -1781,6 +1794,64 @@ export function makePrimeAgentDaemonAdapter( return; } + if (event._tag === "RlmQuiesced") { + const outcome = yield* withThreadLock( + context.threadId, + Effect.gen(function* () { + const turn = context.activeTurn; + if (turn === undefined || turn.terminalQuiescenceToken !== event.token) { + return { settled: false, stop: false }; + } + turn.terminalQuiescenceToken = undefined; + if (!context.runtime.isRlmQuiescenceGenerationCurrent(event.connectionGeneration)) { + const settled = yield* settleActiveTurnLocked(context, turn, { + state: "failed", + errorMessage: + "Prime Agent reconnected before descendant quiescence could be confirmed.", + }); + context.stopRequested = true; + return { settled, stop: true }; + } + const pending = turn.pendingRunCompletionHandoff; + if (pending === undefined && !turn.awaitingQueuedRun) { + return { settled: false, stop: false }; + } + const completionEvent: Extract = + { + ...(pending ?? { event: { _tag: "RunCompleted", messages: [] } }).event, + messages: + pending === undefined + ? turn.completedRunMessages + : [...turn.completedRunMessages, ...pending.event.messages], + ...(event.usage === undefined ? {} : { usageOverride: event.usage }), + }; + const settled = yield* settleActiveTurnLocked(context, turn, { + state: "completed", + event: completionEvent, + }); + return { settled, stop: false }; + }), + ); + if (outcome.settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + if (outcome.stop) { + yield* Effect.forkDetach( + Effect.yieldNow.pipe( + Effect.andThen( + withThreadMutationLock( + context.threadId, + stopSessionInternal( + context, + "Prime Agent session closed after descendant quiescence became uncertain.", + ), + ), + ), + Effect.ignore, + ), + ); + } + return; + } + if (event._tag === "RunCompleted") { const settled = yield* withThreadLock( context.threadId, @@ -1794,6 +1865,18 @@ export function makePrimeAgentDaemonAdapter( turn.queuedActionObserved = false; return false; } + if (turn.terminalQuiescenceToken !== undefined) { + const previous = turn.pendingRunCompletionHandoff; + if (previous !== undefined) { + turn.completedRunMessages.push(...previous.event.messages); + } + turn.runCompletionHandoffSequence += 1; + turn.pendingRunCompletionHandoff = { + sequence: turn.runCompletionHandoffSequence, + event, + }; + return false; + } if (primeAgentRunCompletedNeedsHandoff(event)) { turn.lastAssistantHadRenderableText = false; // A daemon/kernel reconnect can emit a non-final agent_end @@ -2030,11 +2113,13 @@ export function makePrimeAgentDaemonAdapter( event.queuedCount === 0 ) { context.inputQueueClearPending = false; - const settled = yield* settleActiveTurnLocked(context, turn, { - state: "completed", - event: { _tag: "RunCompleted", messages: turn.completedRunMessages }, - }); - if (settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + if (turn.terminalQuiescenceToken === undefined) { + const settled = yield* settleActiveTurnLocked(context, turn, { + state: "completed", + event: { _tag: "RunCompleted", messages: turn.completedRunMessages }, + }); + if (settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + } } else if (turn.awaitingQueuedRun && turn.queuedActionObserved) { const clearExit = yield* context.runtime.abortAndClearQueue.pipe(Effect.exit); yield* settleActiveTurnLocked(context, turn, { @@ -2181,6 +2266,60 @@ export function makePrimeAgentDaemonAdapter( } }); + const awaitRlmQuiescence = (context: PrimeAgentDaemonSessionContext, token: string) => + context.runtime + .waitForRlmQuiescence(token) + .pipe( + Effect.mapError((error) => + runtimeOperationError(context.threadId, "session/rlm-quiescence", error), + ), + ); + + const stopAfterRlmQuiescenceFailure = ( + context: PrimeAgentDaemonSessionContext, + turn: PrimeAgentDaemonActiveTurn, + token: string, + ) => + withThreadMutationLock( + context.threadId, + Effect.gen(function* () { + if ( + sessions.get(context.threadId) !== context || + context.stopped || + context.activeTurn !== turn || + turn.cancellationRequested || + turn.terminalQuiescenceToken !== token + ) { + return; + } + yield* settleActiveTurnLocked(context, turn, { + state: "failed", + errorMessage: "Prime Agent could not confirm descendant quiescence.", + }); + yield* stopSessionInternal( + context, + "Prime Agent session closed after descendant quiescence could not be confirmed.", + ).pipe(Effect.ignore); + }), + ); + + /** Must be called with the thread lock held. */ + const rearmRlmQuiescenceLocked = ( + context: PrimeAgentDaemonSessionContext, + turn: PrimeAgentDaemonActiveTurn, + ) => + Effect.gen(function* () { + if (!context.runtime.rlmQuiescenceAvailable) return; + turn.terminalQuiescenceGeneration += 1; + const token = rlmQuiescenceToken(turn.id, turn.terminalQuiescenceGeneration); + turn.terminalQuiescenceToken = token; + yield* Effect.forkDetach( + awaitRlmQuiescence(context, token).pipe( + Effect.catch(() => stopAfterRlmQuiescenceFailure(context, turn, token)), + ), + ); + }); + const startSession: PrimeAgentAdapterShape["startSession"] = (input) => withThreadMutationLock( input.threadId, @@ -2819,6 +2958,7 @@ export function makePrimeAgentDaemonAdapter( ); activeTurn.queuedInputCount += 1; promotePendingRunCompletionToQueuedRun(activeTurn); + yield* rearmRlmQuiescenceLocked(context, activeTurn); return { _tag: "Steered" as const, result: { @@ -2873,6 +3013,10 @@ export function makePrimeAgentDaemonAdapter( activeAssistantItemId: undefined, lastAssistantHadRenderableText: false, runCompletionHandoffSequence: 0, + terminalQuiescenceGeneration: context.runtime.rlmQuiescenceAvailable ? 1 : 0, + terminalQuiescenceToken: context.runtime.rlmQuiescenceAvailable + ? rlmQuiescenceToken(turnId, 1) + : undefined, pendingRunCompletionHandoff: undefined, queuedInputCount: 0, awaitingQueuedRun: false, @@ -2907,6 +3051,7 @@ export function makePrimeAgentDaemonAdapter( resumeCursor: context.session.resumeCursor, }; + const initialRlmQuiescenceToken = turn.terminalQuiescenceToken; const runPrompt = Effect.gen(function* () { const turnModel = requestedModel || context.session.model || "default"; yield* offerRuntimeEvent({ @@ -2922,6 +3067,9 @@ export function makePrimeAgentDaemonAdapter( .prompt({ text, ...(images.length === 0 ? {} : { images }), + ...(initialRlmQuiescenceToken === undefined + ? {} + : { rlmQuiescenceToken: initialRlmQuiescenceToken }), signal: turn.controller.signal, }) .pipe( @@ -2929,6 +3077,35 @@ export function makePrimeAgentDaemonAdapter( runtimeOperationError(input.threadId, "session/prompt", error), ), ); + if (initialRlmQuiescenceToken !== undefined) { + yield* awaitRlmQuiescence(context, initialRlmQuiescenceToken).pipe( + Effect.catch((error) => + withThreadMutationLock( + context.threadId, + Effect.gen(function* () { + if ( + context.activeTurn !== turn || + turn.terminalQuiescenceToken !== initialRlmQuiescenceToken + ) { + return; + } + if (turn.cancellationRequested || turn.controller.signal.aborted) { + return yield* error; + } + yield* settleActiveTurnLocked(context, turn, { + state: "failed", + errorMessage: "Prime Agent could not confirm descendant quiescence.", + }); + yield* stopSessionInternal( + context, + "Prime Agent session closed after descendant quiescence could not be confirmed.", + ).pipe(Effect.ignore); + return yield* error; + }), + ), + ), + ); + } yield* Deferred.await(turn.completed); return result; }); @@ -3898,6 +4075,7 @@ export function makePrimeAgentDaemonAdapter( yield* updateInputQueueProjection(context, next); turn.queuedInputCount = Math.max(1, next.steeringCount + next.followUpCount); promotePendingRunCompletionToQueuedRun(turn); + yield* rearmRlmQuiescenceLocked(context, turn); return context.inputQueue; } if (Exit.isSuccess(reconciled)) { @@ -3943,11 +4121,13 @@ export function makePrimeAgentDaemonAdapter( !status.isStreaming ) { context.inputQueueClearPending = false; - const settled = yield* settleActiveTurnLocked(context, turn, { - state: "completed", - event: { _tag: "RunCompleted", messages: turn.completedRunMessages }, - }); - if (settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + if (turn.terminalQuiescenceToken === undefined) { + const settled = yield* settleActiveTurnLocked(context, turn, { + state: "completed", + event: { _tag: "RunCompleted", messages: turn.completedRunMessages }, + }); + if (settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + } } else if (status.activeAction || status.isStreaming) { context.inputQueueClearPending = false; } @@ -4050,11 +4230,13 @@ export function makePrimeAgentDaemonAdapter( !status.isStreaming ) { context.inputQueueClearPending = false; - const settled = yield* settleActiveTurnLocked(context, turn, { - state: "completed", - event: { _tag: "RunCompleted", messages: turn.completedRunMessages }, - }); - if (settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + if (turn.terminalQuiescenceToken === undefined) { + const settled = yield* settleActiveTurnLocked(context, turn, { + state: "completed", + event: { _tag: "RunCompleted", messages: turn.completedRunMessages }, + }); + if (settled) yield* refreshContextUsage(context).pipe(Effect.forkDetach); + } } } context.inputQueueClearPending = false; diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts b/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts index 6b934f01f..577c7a002 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonBridge.ts @@ -123,6 +123,9 @@ export interface PrimeAgentDaemonAgentConnection { message: string, options?: PrimeAgentDaemonPromptOptions, ) => Promise; + readonly waitForHeadlessCompletion?: (options?: { + readonly waitForRlmQuiescence?: boolean; + }) => Promise; readonly steer?: ( message: string, images?: ReadonlyArray, diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts b/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts index 074763baf..4b3a3bcb4 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts @@ -676,7 +676,19 @@ export interface PrimeDaemonSessionState { export type PrimeDaemonEvent = | { readonly _tag: "RunStarted" } - | { readonly _tag: "RunCompleted"; readonly messages: ReadonlyArray } + | { + readonly _tag: "RunCompleted"; + readonly messages: ReadonlyArray; + /** Authoritative cumulative-session delta captured at descendant quiescence. */ + readonly usageOverride?: PrimeDaemonUsage | undefined; + } + /** Server-owned FIFO marker; untrusted daemon notifications can never decode to this tag. */ + | { + readonly _tag: "RlmQuiesced"; + readonly token: string; + readonly connectionGeneration: number; + readonly usage?: PrimeDaemonUsage | undefined; + } | { readonly _tag: "TurnStarted" } | { readonly _tag: "TurnCompleted"; diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonRuntimeEvents.ts b/apps/server/src/provider/prime/PrimeAgentDaemonRuntimeEvents.ts index 5914e1929..b21980b83 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonRuntimeEvents.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonRuntimeEvents.ts @@ -379,7 +379,7 @@ export function mapPrimeAgentDaemonRuntimeEventDrafts(input: { message?.stopReason !== "toolUse" && message?.toolCalls.length === 0; const missingFinalResponse = state !== "cancelled" && !hasFinalResponse; - const usage = aggregateAssistantUsage(runMessages); + const usage = event.usageOverride ?? aggregateAssistantUsage(runMessages); const completed: PrimeAgentRuntimeEventDraft = { ...base, type: "turn.completed", @@ -673,6 +673,7 @@ export function mapPrimeAgentDaemonRuntimeEventDrafts(input: { }, ]; } + case "RlmQuiesced": case "AgentMessageSent": case "QueueChanged": case "ThinkingLevelChanged": diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts index 6d45e1968..c580eaa3e 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.test.ts @@ -147,6 +147,7 @@ function fixture(options?: { readonly watchSessionMalformed?: boolean; readonly getWatchMessages?: () => ReadonlyArray | Promise>; readonly sessionStats?: unknown; + readonly getSessionStatsImpl?: () => Promise; readonly getQueueImpl?: () => Promise; readonly clearQueueImpl?: () => Promise; readonly mutateQueuedMessageImpl?: ( @@ -166,6 +167,10 @@ function fixture(options?: { readonly abortCompactionImpl?: () => Promise; readonly setAutoCompactionImpl?: (enabled: boolean) => Promise; readonly setModelImpl?: (provider: string, modelId: string) => Promise; + readonly waitForHeadlessCompletionImpl?: (options: { + readonly waitForRlmQuiescence?: boolean; + }) => Promise; + readonly omitRlmQuiescence?: boolean; }) { const captures: Captures = { order: [], @@ -285,6 +290,9 @@ function fixture(options?: { releaseAcpMcpServers: { value: undefined }, }); } + if (options?.omitRlmQuiescence === true) { + Object.defineProperty(this, "waitForHeadlessCompletion", { value: undefined }); + } if (options?.authoritativeRlmChildren === undefined) { Object.defineProperty(this, "getRlmChildSnapshots", { value: undefined }); } @@ -338,6 +346,18 @@ function fixture(options?: { captures.connectionCalls.push({ method: "prompt", args: [message, promptOptions] }); return Promise.resolve(undefined); } + waitForHeadlessCompletion( + waitOptions: { readonly waitForRlmQuiescence?: boolean } = {}, + ): Promise { + captures.connectionCalls.push({ + method: "waitForHeadlessCompletion", + args: [waitOptions], + }); + return ( + options?.waitForHeadlessCompletionImpl?.(waitOptions) ?? + Promise.resolve({ privateAutonomousStatus: "discarded" }) + ); + } steer(message: string, images?: ReadonlyArray): Promise { captures.connectionCalls.push({ method: "steer", args: [message, images] }); return Promise.resolve(undefined); @@ -513,6 +533,7 @@ function fixture(options?: { } getSessionStats(): Promise { captures.connectionCalls.push({ method: "getSessionStats", args: [] }); + if (options?.getSessionStatsImpl !== undefined) return options.getSessionStatsImpl(); return Promise.resolve( options?.sessionStats ?? { sessionFile: "/daemon/private/session.jsonl", @@ -1948,6 +1969,257 @@ describe("PrimeAgentDaemonSessionRuntime", () => { }), ); + it.effect("orders the authoritative RLM quiescence marker after native run events", () => + Effect.scoped( + Effect.gen(function* () { + let emitNative: ((event: unknown) => Promise) | undefined; + const test = fixture({ + waitForHeadlessCompletionImpl: async (waitOptions) => { + expect(waitOptions).toEqual({ waitForRlmQuiescence: true }); + await emitNative?.({ + type: "session_event", + event: { type: "agent_end", messages: [] }, + }); + return { privateAutonomousStatus: "discarded" }; + }, + }); + emitNative = test.emit; + const runtime = yield* test.make(); + const collecting = yield* collectEvents(runtime, 3).pipe( + Effect.forkChild({ startImmediately: true }), + ); + + const token = "turn-1:1"; + yield* runtime.prompt({ text: "wait for descendants", rlmQuiescenceToken: token }); + yield* runtime.waitForRlmQuiescence(token); + const events = yield* Fiber.join(collecting); + + expect(runtime.rlmQuiescenceAvailable).toBe(true); + expect(events.map((event) => event._tag)).toEqual([ + "SessionResynced", + "RunCompleted", + "RlmQuiesced", + ]); + expect(events.every((event) => !("privateAutonomousStatus" in event))).toBe(true); + expect(events.at(-1)).toMatchObject({ + _tag: "RlmQuiesced", + token, + connectionGeneration: 0, + usage: { + inputTokens: 0, + outputTokens: 0, + cachedInputTokens: 0, + cacheWriteTokens: 0, + totalTokens: 0, + totalCostUsd: 0, + }, + }); + }), + ), + ); + + it.effect("serializes rearmed quiescence barriers", () => + Effect.scoped( + Effect.gen(function* () { + let activeBarriers = 0; + let maxActiveBarriers = 0; + let barrierStarts = 0; + const releases: Array<() => void> = []; + let resolveFirstStart: (() => void) | undefined; + let resolveSecondStart: (() => void) | undefined; + const firstStarted = new Promise((resolve) => { + resolveFirstStart = resolve; + }); + const secondStarted = new Promise((resolve) => { + resolveSecondStart = resolve; + }); + const test = fixture({ + waitForHeadlessCompletionImpl: () => + new Promise((resolve) => { + activeBarriers += 1; + maxActiveBarriers = Math.max(maxActiveBarriers, activeBarriers); + barrierStarts += 1; + (barrierStarts === 1 ? resolveFirstStart : resolveSecondStart)?.(); + releases.push(() => { + activeBarriers -= 1; + resolve({ result: "completed" }); + }); + }), + }); + const runtime = yield* test.make(); + yield* runtime.prompt({ text: "start descendants", rlmQuiescenceToken: "turn-1:1" }); + const first = yield* runtime + .waitForRlmQuiescence("turn-1:1") + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => firstStarted); + const second = yield* runtime + .waitForRlmQuiescence("turn-1:2") + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.yieldNow; + + expect(barrierStarts).toBe(1); + expect(maxActiveBarriers).toBe(1); + releases.shift()?.(); + yield* Fiber.join(first); + yield* Effect.promise(() => secondStarted); + expect(barrierStarts).toBe(2); + expect(maxActiveBarriers).toBe(1); + releases.shift()?.(); + yield* Fiber.join(second); + }), + ), + ); + + it.effect("rejects a quiescence barrier that overlaps daemon reconnect recovery", () => + Effect.scoped( + Effect.gen(function* () { + let releaseBarrier: (() => void) | undefined; + let reportBarrierStarted: (() => void) | undefined; + const barrierStarted = new Promise((resolve) => { + reportBarrierStarted = resolve; + }); + const barrierRelease = new Promise((resolve) => { + releaseBarrier = resolve; + }); + let releaseMcpRecovery: (() => void) | undefined; + let reportMcpRecoveryStarted: (() => void) | undefined; + const mcpRecoveryStarted = new Promise((resolve) => { + reportMcpRecoveryStarted = resolve; + }); + const mcpRecoveryRelease = new Promise((resolve) => { + releaseMcpRecovery = resolve; + }); + let replaceMcpCalls = 0; + const test = fixture({ + waitForHeadlessCompletionImpl: () => { + reportBarrierStarted?.(); + return barrierRelease; + }, + replaceMcpImpl: () => { + replaceMcpCalls += 1; + if (replaceMcpCalls === 1) return Promise.resolve(undefined); + reportMcpRecoveryStarted?.(); + return mcpRecoveryRelease; + }, + }); + const runtime = yield* test.make(undefined, undefined, undefined, undefined, { + ownerId: "pylon:provider-session-reconnect", + server: { + name: "t3-code", + type: "http", + url: "http://127.0.0.1:4321/mcp/provider-session-reconnect", + headers: { Authorization: "Bearer scoped-secret" }, + }, + }); + const token = "turn-reconnect:1"; + yield* runtime.prompt({ text: "wait through reconnect", rlmQuiescenceToken: token }); + const waiting = yield* runtime + .waitForRlmQuiescence(token) + .pipe(Effect.flip, Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => barrierStarted); + + yield* Effect.promise(() => + test.emit({ type: "connection_status", status: "reconnecting" }), + ); + const resyncDelivery = test.emit({ + type: "session_resynced", + snapshot: snapshot(99), + }); + yield* Effect.promise(() => mcpRecoveryStarted); + expect(runtime.isRlmQuiescenceGenerationCurrent(0)).toBe(false); + releaseBarrier?.(); + + const error = yield* Fiber.join(waiting); + expect(error).toMatchObject({ + operation: "rlm-quiescence", + reason: "request-failed", + detail: "Prime Agent reconnected before descendant quiescence could be confirmed.", + }); + releaseMcpRecovery?.(); + yield* Effect.promise(() => resyncDelivery); + }), + ), + ); + + it.effect("reports the full cumulative usage delta at authoritative quiescence", () => + Effect.scoped( + Effect.gen(function* () { + const stats = [ + { + sessionId: "session-1", + tokens: { input: 100, output: 20, cacheRead: 300, cacheWrite: 5, total: 425 }, + cost: 0.25, + }, + { + sessionId: "session-1", + tokens: { input: 160, output: 35, cacheRead: 500, cacheWrite: 8, total: 703 }, + cost: 0.5, + }, + ]; + const test = fixture({ getSessionStatsImpl: () => Promise.resolve(stats.shift()) }); + const runtime = yield* test.make(); + const collecting = yield* collectEvents(runtime, 2).pipe( + Effect.forkChild({ startImmediately: true }), + ); + const token = "turn-usage:1"; + + yield* runtime.prompt({ text: "include child usage", rlmQuiescenceToken: token }); + yield* runtime.waitForRlmQuiescence(token); + + expect((yield* Fiber.join(collecting)).at(-1)).toMatchObject({ + _tag: "RlmQuiesced", + token, + usage: { + inputTokens: 60, + outputTokens: 15, + cachedInputTokens: 200, + cacheWriteTokens: 3, + totalTokens: 278, + totalCostUsd: 0.25, + }, + }); + }), + ), + ); + + it.effect("fails a prompt safely when the advertised RLM barrier cannot complete", () => + Effect.scoped( + Effect.gen(function* () { + const { make } = fixture({ + waitForHeadlessCompletionImpl: () => + Promise.reject(new Error("private daemon failure at /secret/path")), + }); + const runtime = yield* make(); + + yield* runtime.prompt({ text: "wait safely", rlmQuiescenceToken: "turn-1:1" }); + const error = yield* runtime.waitForRlmQuiescence("turn-1:1").pipe(Effect.flip); + + expect(error).toMatchObject({ + operation: "rlm-quiescence", + reason: "request-failed", + detail: "Prime Agent could not confirm descendant quiescence.", + }); + expect(error.detail).not.toContain("/secret/path"); + }), + ), + ); + + it.effect("retains prompt settlement for older connections without an RLM barrier", () => + Effect.scoped( + Effect.gen(function* () { + const { captures, make } = fixture({ omitRlmQuiescence: true }); + const runtime = yield* make(); + + yield* runtime.prompt({ text: "legacy prompt" }); + + expect(runtime.rlmQuiescenceAvailable).toBe(false); + expect( + captures.connectionCalls.some((call) => call.method === "waitForHeadlessCompletion"), + ).toBe(false); + }), + ), + ); + it.effect("exposes typed operations and strips native model payloads", () => Effect.scoped( Effect.gen(function* () { @@ -1955,7 +2227,13 @@ describe("PrimeAgentDaemonSessionRuntime", () => { const runtime = yield* make(); const images = [{ type: "image", data: "aGVsbG8=", mimeType: "image/png" }] as const; const signal = new AbortController().signal; - yield* runtime.prompt({ text: "prompt", images, signal }); + yield* runtime.prompt({ + text: "prompt", + images, + signal, + rlmQuiescenceToken: "turn-typed:1", + }); + yield* runtime.waitForRlmQuiescence("turn-typed:1"); yield* runtime.steer({ text: "steer", images }); yield* runtime.followUp({ text: "follow", images }); yield* runtime.abort; @@ -1974,6 +2252,14 @@ describe("PrimeAgentDaemonSessionRuntime", () => { expect(selected).not.toHaveProperty("baseUrl"); expect(selected).not.toHaveProperty("headers"); expect(stats).toEqual({ + usage: { + inputTokens: 120, + outputTokens: 30, + cachedInputTokens: 850, + cacheWriteTokens: 10, + totalTokens: 1_010, + totalCostUsd: 0.42, + }, contextUsage: { usedTokens: 320, maxTokens: 200_000 }, }); expect(stats).not.toHaveProperty("sessionFile"); @@ -1990,7 +2276,10 @@ describe("PrimeAgentDaemonSessionRuntime", () => { ["getResourceSnapshot", []], ["getCommands", []], ["getRlmMaxDepthStatus", []], + ["getSessionStats", []], ["prompt", ["prompt", { queueIfBusy: false, images, signal }]], + ["waitForHeadlessCompletion", [{ waitForRlmQuiescence: true }]], + ["getSessionStats", []], ["steer", ["steer", images]], ["followUp", ["follow", images]], ["abort", []], diff --git a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts index b8f089acc..f76cdd49d 100644 --- a/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts +++ b/apps/server/src/provider/prime/PrimeAgentDaemonSessionRuntime.ts @@ -31,6 +31,7 @@ import * as Predicate from "effect/Predicate"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { @@ -48,6 +49,7 @@ import { decodePrimeAgentDaemonEvent, decodePrimeAgentDaemonSessionState, type PrimeDaemonEvent, + type PrimeDaemonUsage, } from "./PrimeAgentDaemonEvents.ts"; import type { PrimeAgentDaemonManager } from "./PrimeAgentDaemonManager.ts"; import { PRIME_AGENT_EVENT_BUFFER_CAPACITY } from "./PrimeAgentEventBuffer.ts"; @@ -61,6 +63,7 @@ import { export { PRIME_AGENT_DAEMON_RESUME_CURSOR } from "./PrimeAgentResumeCursor.ts"; const COMMAND_TIMEOUT_MS = 30_000; +const RLM_QUIESCENCE_STATS_TIMEOUT_MS = 2_000; const SIDE_QUESTION_TERMINAL_MAX_BYTES = 8_192; const SIDE_QUESTION_TERMINAL_MAX_CODEPOINTS = 8_192; const SIDE_QUESTION_MAX_UPDATES = 512; @@ -465,11 +468,23 @@ const commandsSchema = Schema.Array( }), }), ); +const nonNegativeInt = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)); +const nonNegativeFinite = Schema.Finite.check(Schema.isGreaterThanOrEqualTo(0)); const sessionStatsSchema = Schema.Struct({ sessionId: Schema.NonEmptyString, + tokens: Schema.optional( + Schema.Struct({ + input: nonNegativeInt, + output: nonNegativeInt, + cacheRead: nonNegativeInt, + cacheWrite: nonNegativeInt, + total: nonNegativeInt, + }), + ), + cost: Schema.optional(nonNegativeFinite), contextUsage: Schema.optional( Schema.Struct({ - tokens: Schema.NullOr(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), + tokens: Schema.NullOr(nonNegativeInt), contextWindow: Schema.Int.check(Schema.isGreaterThan(0)), }), ), @@ -502,6 +517,24 @@ const decodeRlmMaxDepthStatus = Schema.decodeUnknownOption(rlmMaxDepthStatusSche const decodeAgentMessageReceipt = Schema.decodeUnknownOption(agentMessageReceiptSchema); const decodeRefinementResult = Schema.decodeUnknownOption(refinementResultSchema); +function subtractCumulativeUsage( + current: PrimeDaemonUsage | undefined, + baseline: PrimeDaemonUsage | undefined, +): PrimeDaemonUsage | undefined { + if (current === undefined || baseline === undefined) return undefined; + const usage = { + inputTokens: current.inputTokens - baseline.inputTokens, + outputTokens: current.outputTokens - baseline.outputTokens, + cachedInputTokens: current.cachedInputTokens - baseline.cachedInputTokens, + cacheWriteTokens: current.cacheWriteTokens - baseline.cacheWriteTokens, + totalTokens: current.totalTokens - baseline.totalTokens, + totalCostUsd: current.totalCostUsd - baseline.totalCostUsd, + }; + return Object.values(usage).every((value) => Number.isFinite(value) && value >= 0) + ? usage + : undefined; +} + function providerAgentDepthSource( source: (typeof rlmMaxDepthStatusSchema.Type)["source"], ): Exclude { @@ -643,6 +676,7 @@ const runtimeErrorOperation = Schema.Literals([ "message-agent", "watch-agent-activity", "prompt", + "rlm-quiescence", "steer", "follow-up", "get-input-queue", @@ -719,6 +753,8 @@ export interface PrimeAgentDaemonSessionRuntimeInput { export interface PrimeAgentDaemonPromptInput { readonly text: string; readonly images?: ReadonlyArray; + /** Correlates the initial descendant barrier with one Pylon turn/input generation. */ + readonly rlmQuiescenceToken?: string; /** Cancels prompt admission before the daemon accepts ownership of the turn. */ readonly signal?: AbortSignal; } @@ -779,6 +815,7 @@ export interface PrimeAgentDaemonReloadResourcesResult { /** Provider-neutral session usage fields projected from Prime's private daemon response. */ export interface PrimeAgentDaemonSessionStats { + readonly usage?: PrimeDaemonUsage | undefined; readonly contextUsage?: | { readonly usedTokens: number | null; @@ -860,6 +897,12 @@ export interface PrimeAgentDaemonSessionRuntime { PrimeAgentDaemonSessionRuntimeError >; readonly events: Stream.Stream; + /** True only when Prime exposes its authoritative descendant-quiescence barrier. */ + readonly rlmQuiescenceAvailable: boolean; + readonly waitForRlmQuiescence: ( + token: string, + ) => Effect.Effect; + readonly isRlmQuiescenceGenerationCurrent: (generation: number) => boolean; readonly prompt: ( input: PrimeAgentDaemonPromptInput, ) => Effect.Effect; @@ -1134,6 +1177,10 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo let unsubscribe: (() => void) | undefined; let disposed = false; let disposeStarted = false; + let connectionGeneration = 0; + let rlmEventContinuityValid = true; + let rlmTurnUsageBaseline: PrimeDaemonUsage | undefined; + const rlmQuiescenceSemaphore = yield* Semaphore.make(1); const closeClient = Effect.sync(() => { client.close(); @@ -1564,6 +1611,16 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo typeof raw === "object" && raw !== null && "type" in raw && typeof raw.type === "string" ? raw.type : undefined; + const connectionStatus = + rawType === "connection_status" && + "status" in (raw as object) && + typeof (raw as { readonly status?: unknown }).status === "string" + ? (raw as { readonly status: string }).status + : undefined; + if (connectionStatus === "reconnecting") { + connectionGeneration += 1; + rlmEventContinuityValid = false; + } if ( input.mcpServer === undefined || (rawType !== "session_resynced" && rawType !== "connection_status" && rawType !== "closed") @@ -1574,11 +1631,7 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo runPromise( Effect.gen(function* () { if (rawType === "connection_status") { - const status = - "status" in (raw as object) && - typeof (raw as { readonly status?: unknown }).status === "string" - ? (raw as { readonly status: string }).status - : undefined; + const status = connectionStatus; if (status === "reconnecting") { mcpAttached = false; mcpRecoveryPending = true; @@ -1932,6 +1985,7 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo const agentMessageAvailable = input.requiredExtension === undefined && Predicate.isFunction(connection!.sendAgentMessage); + const rlmQuiescenceAvailable = Predicate.isFunction(connection!.waitForHeadlessCompletion); const compactionAvailable = input.requiredExtension === undefined && Predicate.isFunction(connection!.getState) && @@ -2745,6 +2799,73 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo ).pipe(Effect.catch((error) => Queue.fail(queue, error))), ); + const readRlmUsage = Effect.fn("PrimeAgentDaemonSessionRuntime.readRlmUsage")(function* () { + const statsOption = yield* getSessionStats.pipe( + Effect.timeoutOption(RLM_QUIESCENCE_STATS_TIMEOUT_MS), + Effect.orElseSucceed(() => Option.none()), + ); + return Option.isSome(statsOption) ? statsOption.value.usage : undefined; + }); + + const waitForRlmQuiescence = Effect.fn("PrimeAgentDaemonSessionRuntime.waitForRlmQuiescence")( + function* (token: string) { + yield* rlmQuiescenceSemaphore.withPermit( + Effect.gen(function* () { + yield* ensureOpen("rlm-quiescence"); + if (!rlmQuiescenceAvailable) return; + if (!rlmEventContinuityValid) { + return yield* runtimeError( + "rlm-quiescence", + "request-failed", + "Prime Agent reconnected before descendant quiescence could be confirmed.", + ); + } + const waitForHeadlessCompletion = yield* requireMethod( + "rlm-quiescence", + connection!.waitForHeadlessCompletion, + ); + const startedConnectionGeneration = connectionGeneration; + // Prime's result is autonomous-provider state. Pylon needs only the + // authoritative ordering boundary and never projects the native payload. + yield* Effect.tryPromise({ + try: () => + waitForHeadlessCompletion.call(connection, { + waitForRlmQuiescence: true, + }), + catch: () => + runtimeError( + "rlm-quiescence", + "request-failed", + "Prime Agent could not confirm descendant quiescence.", + ), + }); + if (!rlmEventContinuityValid || connectionGeneration !== startedConnectionGeneration) { + return yield* runtimeError( + "rlm-quiescence", + "request-failed", + "Prime Agent reconnected before descendant quiescence could be confirmed.", + ); + } + const currentUsage = yield* readRlmUsage(); + if (!rlmEventContinuityValid || connectionGeneration !== startedConnectionGeneration) { + return yield* runtimeError( + "rlm-quiescence", + "request-failed", + "Prime Agent reconnected before descendant quiescence could be confirmed.", + ); + } + const usage = subtractCumulativeUsage(currentUsage, rlmTurnUsageBaseline); + yield* Queue.offer(eventQueue, { + _tag: "RlmQuiesced", + token, + connectionGeneration: startedConnectionGeneration, + ...(usage === undefined ? {} : { usage }), + }); + }), + ); + }, + ); + const prompt = Effect.fn("PrimeAgentDaemonSessionRuntime.prompt")(function* ( promptInput: PrimeAgentDaemonPromptInput, ) { @@ -2752,6 +2873,12 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo yield* awaitMcpRecovery; const images = yield* validateImages("prompt", promptInput.images); yield* validatePromptContent("prompt", promptInput.text, images); + if (rlmQuiescenceAvailable && promptInput.rlmQuiescenceToken !== undefined) { + rlmEventContinuityValid = true; + rlmTurnUsageBaseline = yield* readRlmUsage(); + } else { + rlmTurnUsageBaseline = undefined; + } yield* callVoid("prompt", () => connection!.promptAndWait(promptInput.text, { queueIfBusy: false, @@ -3388,14 +3515,28 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo "The daemon returned invalid session usage.", ); } - return decoded.value.contextUsage === undefined - ? {} - : ({ - contextUsage: { - usedTokens: decoded.value.contextUsage.tokens, - maxTokens: decoded.value.contextUsage.contextWindow, - }, - } satisfies PrimeAgentDaemonSessionStats); + const usage = + decoded.value.tokens === undefined || decoded.value.cost === undefined + ? undefined + : { + inputTokens: decoded.value.tokens.input, + outputTokens: decoded.value.tokens.output, + cachedInputTokens: decoded.value.tokens.cacheRead, + cacheWriteTokens: decoded.value.tokens.cacheWrite, + totalTokens: decoded.value.tokens.total, + totalCostUsd: decoded.value.cost, + }; + return { + ...(usage === undefined ? {} : { usage }), + ...(decoded.value.contextUsage === undefined + ? {} + : { + contextUsage: { + usedTokens: decoded.value.contextUsage.tokens, + maxTokens: decoded.value.contextUsage.contextWindow, + }, + }), + } satisfies PrimeAgentDaemonSessionStats; }); // Keep initialization admission active through every control-plane read. @@ -3500,6 +3641,10 @@ export const makePrimeAgentDaemonSessionRuntime = Effect.fn("makePrimeAgentDaemo watchAgentActivityAvailable, watchAgentActivity, events: Stream.fromQueue(eventQueue), + rlmQuiescenceAvailable, + waitForRlmQuiescence, + isRlmQuiescenceGenerationCurrent: (generation) => + rlmEventContinuityValid && generation === connectionGeneration, prompt, steer, followUp, diff --git a/docs/internals/prime-agent-daemon-parity.md b/docs/internals/prime-agent-daemon-parity.md index e66a77646..f2b9931ec 100644 --- a/docs/internals/prime-agent-daemon-parity.md +++ b/docs/internals/prime-agent-daemon-parity.md @@ -30,50 +30,59 @@ Prime event queues are bounded to 256 entries and preserve FIFO delivery with ba Daemon assistant messages receive opaque subscriber-local segment identifiers so separate model/tool cycles within one Pylon turn retain their chronological positions without persisting Prime's native -identifiers. An error- or tool-terminated native run completion without a final public response is held for a bounded three-second handoff because Prime -may reconnect and automatically continue that same turn; a following native run remains attached to -the original Pylon turn, while an exhausted handoff settles as a real failure. Daemon and ACP modes +identifiers. After an ordinary daemon prompt reaches its first response boundary, Pylon invokes Prime +Agent 0.8.0's public RLM-quiescence barrier. Its server-owned FIFO marker carries the Pylon turn and input +generation, so a late marker cannot settle later steering, follow-up, or a different turn. Every admitted +input rearms the barrier. Native barrier calls are serialized so re-arming cannot duplicate autonomous +continuation checks. A daemon connection-generation change invalidates quiescence for the rest of the +canonical turn, including a queued re-arm; Pylon fails the turn and disposes the uncertain native session +rather than reconstructing missed messages from a private transcript. When both bounded stats reads +succeed, the usage delta includes child billing that Prime attributes after the original message event. The native autonomous-status result is +discarded at the provider boundary. Older connections without the barrier retain response-boundary +behavior. An error- or tool-terminated native run completion without a +final public response is also held for a bounded three-second reconnect handoff; a following native run +remains attached to the original Pylon turn, while an exhausted handoff settles as a real failure. Daemon and ACP modes both emit a fixed provider-neutral status before an authoritative terminal event when no public final assistant text follows the latest tool boundary. Reasoning, tool data, native errors, and identifiers are never used to synthesize assistant prose. -| Public API outcome | Pylon status | Decision | -| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `attach`, root `subscribe`, `getState`, `getInitialSnapshot`, `dispose` | Integrated internally | Own one private daemon session per active Pylon thread, resume exact verified identity, reconnect through public snapshots/events, and close with the thread scope. On 0.8.0 recovery Pylon resupplies the owner `cwd`, session/agent directories, execution policy, and current Pylon-selected model and thinking level; successful model and thinking mutations update that transient recovery context, which Prime never persists. The root connection's `getMessages` transcript API is never called; Pylon's event-sourced transcript remains authoritative. These are lifecycle primitives, not client RPCs. | -| `promptAndWait`, `steer`, `followUp`, `abort` | Integrated | Typed turn admission, active steering, explicit follow-up, queue modes, and interruption remain under Pylon turn ownership. Admission is never retried after ambiguous ownership. | -| `prompt`, `waitForIdle` | Intentionally redundant | Pylon uses the cancellable `promptAndWait` admission path and daemon events for exact turn settlement. A second fire-and-observe prompt path or an unscoped idle waiter would weaken turn and checkpoint ownership. | -| `getQueue`, `clearQueue`, `setSteeringMode`, `setFollowUpMode` | Integrated | Expose only counts and delivery modes. Queued text stays private. | -| `mutateQueuedMessage` | Partially integrated | Pylon exposes sole-lane deletion only. Preview text stays server-private. A serialized, non-recovering compare-delete is followed by reconciliation; ambiguous mutations are never retried. Multi-item delete, move, and replace remain unavailable without opaque IDs or revisions. | -| `abortAndClearQueue` | Intentionally folded into Stop | The public operation combines interruption and queue deletion. Pylon keeps non-interrupting Clear and authoritative Stop separate so the reverse state is unambiguous. | -| `setModel`, `setThinkingLevel`, `setServiceTier` | Integrated | Exact selection is owned by the durable thread model projection and reconciled against sanitized session state. | -| `getAvailableModels`, `getModelCatalog` | Integrated provider-catalog and auth-readiness enrichment | Attached sessions use `getModelCatalog`, with `getAvailableModels` as fallback. Strict decoding maps configured models into the existing provider snapshot and reports authenticated readiness only from a healthy current catalog containing a configured native provider. Empty catalogs and non-ready probes leave authentication unknown. This does not verify live network access or expose credential data. Failed or late reads keep the last good list without letting it override probe health; discovery never creates a session. | -| `cycleModel`, `cycleThinkingLevel`, `setScopedModels` | Intentionally redundant | Pylon already has an exact multi-client model picker. Prime's scoped list is memory-only and native cycling cannot atomically update Pylon's durable selection, so exposing both would create split-brain state. | -| `setTransport` | Intentionally excluded | Transport is environment/provider plumbing owned by Pylon, not a per-thread user setting. Prime does not expose authoritative current transport in session state. | -| `getSessionStats` | Integrated | Only bounded context usage and finite reported turn-cost outcomes cross the boundary. | -| `compact`, `abortCompaction`, `setAutoCompactionEnabled` | Integrated | Manual compaction is argument-free; summaries, instructions, paths, and native results are discarded. Automatic state is reconciled before publication. | -| `refine({ global: false })` | Integrated | Explicit local-only refinement accepts no instructions or rollback identity. RPC success contains aggregate edit counts only. Timeout/rejection is outcome-unknown and is never retried. | -| `abortBranchSummary` | Intentionally folded into Stop | Pylon does not start a standalone native branch-summary operation; stopping the owning turn remains authoritative. | -| `setAutoRetryEnabled`, `abortRetry` | Deferred | Retry lifecycle is observed safely, but enabled state has no authoritative readback and the setter writes shared provider settings. Stop already cancels the owning turn. A distinct retry control needs truthful session state and receipts. | -| `reload`, `getCommands` | Integrated | Full-access sessions can reload while idle and show bounded safe command metadata. Supervised sessions fail closed. | -| `acquireSessionInputPause` | Intentionally redundant | Every Pylon prompt and resource reload is serialized by the per-thread adapter lock, and reload is admitted only while the owned session is idle. Pylon's scoped MCP server is attached before the first snapshot and released only after turn ownership ends, so it is never replaced during live input. Holding a native lease would add reconnect failure modes without fencing additional work. Revisit if Pylon supports live MCP configuration changes. | -| `supportsAcpMcpServers`, `replaceAcpMcpServers`, `releaseAcpMcpServers` | Integrated with scoped Pylon ownership | `McpProviderSession` is the single per-thread source of truth. Before the first daemon snapshot, Pylon replaces one `t3-code` HTTP server under the stable owner `pylon:` and fails closed if Prime cannot own it. After a daemon reconnect, Pylon reclaims that ownership before publishing the resynced session; if it cannot, the session closes instead of continuing without browser tools. Session teardown releases only that owner and server name before disposing the connection. ACP fallback sends the same scoped server in `session/new`. Browser-disabled sessions send nothing, and Prime-owned MCP settings and catalogs remain private. | -| `getResourceSnapshot` | Partially integrated by safe outcome | Commands and safe skill/prompt metadata are decoded internally. Native paths, diagnostics, extensions, themes, packages, and MCP configuration are not sent to clients. | -| `respondToExtensionUiRequest` | Partially integrated by safe outcome | Select, confirm, and input dialogs plus bounded notifications, status, and widgets are correlated without exposing native request envelopes. Submitted free-form input uses a transient provider RPC and is redacted from durable activities. Editor replacement is cancelled because its prefill may contain sensitive model or tool material that cannot safely enter Pylon's synchronized event stream. | -| `getRlmChildSnapshots`, `getRlmMaxDepthStatus`, `setRlmMaxDepth`, `cancelRlmChild`, `sendAgentMessage` | Integrated | On 0.8.0 the authoritative roster atomically replaces Pylon's private cache after strict bounded decoding; older versions retain the event-derived roster. Canonical Pylon task IDs resolve through that private live roster. Messaging is ephemeral. | -| `watchSession`; watcher `subscribe`, `getMessages`, `close` | Integrated for bounded child live activity | A short-lived watcher attaches only to an explicitly selected active descendant. Its `getMessages` supplies one bounded, sanitized committed-message snapshot with assistant text and a coarse safe-label tool skeleton, then watcher message and tool lifecycle events maintain active-only live activity until the watcher closes. Native tool IDs are immediately reduced to attachment-salted in-memory correlation digests; arguments, results, reasoning, paths, timestamps, metadata, and error text never cross the boundary. The public watcher does not atomically expose attach-time streaming activity, so Pylon does not attempt a private-API workaround for a partial already in flight. This is distinct from, and does not enable, root transcript reads. | -| `getAgentMessageStatus`, `pauseAgentMessages`, `resumeAgentMessages`, `clearAgentMessages` | Intentionally excluded | These controls are daemon-global and can change or clear traffic belonging to unrelated sessions. | -| `startSideQuestion`, `abortSideQuestion` | Integrated as constrained transient quick questions | Supervised, fresh sessions may run one bounded tool-free question through a requester-owned unary RPC. Pylon uses separate public/native IDs, returns only one temporary answer, requests one bounded abort on cancellation, timeout, or disconnect, and never retries or persists the prompt, answer, native errors, or lifecycle. Full-access extension hooks, restored sessions, ACP, follow-up transcripts, and reconnect recovery fail closed. | -| `getHeartbeat`, `setHeartbeat`, `updateHeartbeat`, `listHeartbeats`, `listCronJobs`, `addCronJob`, `cancelCronJob`, `manageHeartbeat` | Deferred on lifecycle ownership | Scheduling promotes work to resident ownership, but public APIs do not provide Pylon an authoritative autonomous-turn/checkpoint identity, demotion, reattachment, or fail-safe delete flow. Shipping now could leave invisible mutations or orphaned work. | -| `getContextTree` | Blocked by upstream execution safety | Prime 0.8.0 synchronously follows and recursively scans unbounded `sub-*` directories before Pylon can decode or time out the result. Until Prime adds intrinsic symlink, cycle, depth, node, and byte bounds, Pylon uses bounded session stats and its own observed agent usage instead; native labels, IDs, model metadata, costs, and history remain private. | -| `getSessionContext`, `getSessionTree`, `getUserMessagesForForking`, `getLastAssistantText` | Intentionally redundant/sensitive | Pylon's event-sourced transcript and checkpoints are authoritative; mirroring Prime's private transcript/tree would create a second history source and expose hidden context. | -| `getSystemPrompt`, `getToolDefinition` | Intentionally excluded | These return hidden instructions, extension schemas, and prompt internals. | -| `listSavedSessions`, `newSession`, `switchSession`, `fork`, `navigateTree`, `importFromJsonl`, `exportToHtml`, `exportToJsonl`, `renameSavedSession`, `deleteSavedSession` | Deferred on history coordination | Public DTOs are filesystem/path-shaped and can enumerate unrelated Prime history. Native history mutation must first coordinate atomically with Pylon threads, worktrees, and checkpoints. | -| `setSessionName`, `setSessionEntryLabel` | Intentionally redundant | Pylon thread titles and durable activities are the user-visible source of truth. | -| `executeBash`, `executeBashAndWait`, `abortBash` | Intentionally redundant | Pylon's terminal and an agent turn have separate ownership and audit semantics; a raw session bash tunnel would bypass both. | -| `waitForHeadlessCompletion({ waitForRlmQuiescence })` | Deferred for daemon autonomous ownership; ACP boundary integrated | Pylon does not start native headless/resident daemon turns, so consuming autonomous status remains meaningless until those turns have checkpoint identity, reattachment, stop, and deletion semantics. ACP compatibility mode instead consumes 0.8.0's correlated terminal-quiescence metadata, which is the transport-owned completion boundary for an ordinary ACP prompt. | -| `getSessionHeader` | Intentionally redundant/sensitive | The header can repeat native saved-session identity and metadata. Pylon uses its private verified resume sidecar plus the durable thread projection instead of exposing or persisting a second header source. | -| `onBeforeSessionInvalidate` | Unavailable in daemon mode 0.8.0 | The public daemon implementation is a no-op returning only an unsubscribe function, so there is no lifecycle outcome to integrate. Pylon tears down its owned connection scope explicitly. | -| `promoteToResident` | Intentionally excluded until automation ownership exists | Client-owned workers must remain stoppable and reapable by their Pylon thread. | +| Public API outcome | Pylon status | Decision | +| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `attach`, root `subscribe`, `getState`, `getInitialSnapshot`, `dispose` | Integrated internally | Own one private daemon session per active Pylon thread, resume exact verified identity, reconnect through public snapshots/events, and close with the thread scope. On 0.8.0 recovery Pylon resupplies the owner `cwd`, session/agent directories, execution policy, and current Pylon-selected model and thinking level; successful model and thinking mutations update that transient recovery context, which Prime never persists. The root connection's `getMessages` transcript API is never called; Pylon's event-sourced transcript remains authoritative. These are lifecycle primitives, not client RPCs. | +| `promptAndWait`, `steer`, `followUp`, `abort` | Integrated | Typed turn admission, active steering, explicit follow-up, queue modes, and interruption remain under Pylon turn ownership. On 0.8.0 the prompt boundary is followed by the public RLM-quiescence barrier so descendant-triggered parent continuations remain in the same canonical turn. Every admitted steer or follow-up advances a correlation generation and rearms that boundary. Barrier or reconnect ambiguity fails the turn and disposes the native session; admission is never retried. | +| `prompt`, `waitForIdle` | Intentionally redundant | Pylon uses the cancellable `promptAndWait` admission path and daemon events for exact turn settlement. A second fire-and-observe prompt path or an unscoped idle waiter would weaken turn and checkpoint ownership. | +| `getQueue`, `clearQueue`, `setSteeringMode`, `setFollowUpMode` | Integrated | Expose only counts and delivery modes. Queued text stays private. | +| `mutateQueuedMessage` | Partially integrated | Pylon exposes sole-lane deletion only. Preview text stays server-private. A serialized, non-recovering compare-delete is followed by reconciliation; ambiguous mutations are never retried. Multi-item delete, move, and replace remain unavailable without opaque IDs or revisions. | +| `abortAndClearQueue` | Intentionally folded into Stop | The public operation combines interruption and queue deletion. Pylon keeps non-interrupting Clear and authoritative Stop separate so the reverse state is unambiguous. | +| `setModel`, `setThinkingLevel`, `setServiceTier` | Integrated | Exact selection is owned by the durable thread model projection and reconciled against sanitized session state. | +| `getAvailableModels`, `getModelCatalog` | Integrated provider-catalog and auth-readiness enrichment | Attached sessions use `getModelCatalog`, with `getAvailableModels` as fallback. Strict decoding maps configured models into the existing provider snapshot and reports authenticated readiness only from a healthy current catalog containing a configured native provider. Empty catalogs and non-ready probes leave authentication unknown. This does not verify live network access or expose credential data. Failed or late reads keep the last good list without letting it override probe health; discovery never creates a session. | +| `cycleModel`, `cycleThinkingLevel`, `setScopedModels` | Intentionally redundant | Pylon already has an exact multi-client model picker. Prime's scoped list is memory-only and native cycling cannot atomically update Pylon's durable selection, so exposing both would create split-brain state. | +| `setTransport` | Intentionally excluded | Transport is environment/provider plumbing owned by Pylon, not a per-thread user setting. Prime does not expose authoritative current transport in session state. | +| `getSessionStats` | Integrated | Only bounded context usage and finite reported turn-cost outcomes cross the boundary. | +| `compact`, `abortCompaction`, `setAutoCompactionEnabled` | Integrated | Manual compaction is argument-free; summaries, instructions, paths, and native results are discarded. Automatic state is reconciled before publication. | +| `refine({ global: false })` | Integrated | Explicit local-only refinement accepts no instructions or rollback identity. RPC success contains aggregate edit counts only. Timeout/rejection is outcome-unknown and is never retried. | +| `abortBranchSummary` | Intentionally folded into Stop | Pylon does not start a standalone native branch-summary operation; stopping the owning turn remains authoritative. | +| `setAutoRetryEnabled`, `abortRetry` | Deferred | Retry lifecycle is observed safely, but enabled state has no authoritative readback and the setter writes shared provider settings. Stop already cancels the owning turn. A distinct retry control needs truthful session state and receipts. | +| `reload`, `getCommands` | Integrated | Full-access sessions can reload while idle and show bounded safe command metadata. Supervised sessions fail closed. | +| `acquireSessionInputPause` | Intentionally redundant | Every Pylon prompt and resource reload is serialized by the per-thread adapter lock, and reload is admitted only while the owned session is idle. Pylon's scoped MCP server is attached before the first snapshot and released only after turn ownership ends, so it is never replaced during live input. Holding a native lease would add reconnect failure modes without fencing additional work. Revisit if Pylon supports live MCP configuration changes. | +| `supportsAcpMcpServers`, `replaceAcpMcpServers`, `releaseAcpMcpServers` | Integrated with scoped Pylon ownership | `McpProviderSession` is the single per-thread source of truth. Before the first daemon snapshot, Pylon replaces one `t3-code` HTTP server under the stable owner `pylon:` and fails closed if Prime cannot own it. After a daemon reconnect, Pylon reclaims that ownership before publishing the resynced session; if it cannot, the session closes instead of continuing without browser tools. Session teardown releases only that owner and server name before disposing the connection. ACP fallback sends the same scoped server in `session/new`. Browser-disabled sessions send nothing, and Prime-owned MCP settings and catalogs remain private. | +| `getResourceSnapshot` | Partially integrated by safe outcome | Commands and safe skill/prompt metadata are decoded internally. Native paths, diagnostics, extensions, themes, packages, and MCP configuration are not sent to clients. | +| `respondToExtensionUiRequest` | Partially integrated by safe outcome | Select, confirm, and input dialogs plus bounded notifications, status, and widgets are correlated without exposing native request envelopes. Submitted free-form input uses a transient provider RPC and is redacted from durable activities. Editor replacement is cancelled because its prefill may contain sensitive model or tool material that cannot safely enter Pylon's synchronized event stream. | +| `getRlmChildSnapshots`, `getRlmMaxDepthStatus`, `setRlmMaxDepth`, `cancelRlmChild`, `sendAgentMessage` | Integrated | On 0.8.0 the authoritative roster atomically replaces Pylon's private cache after strict bounded decoding; older versions retain the event-derived roster. Canonical Pylon task IDs resolve through that private live roster. Messaging is ephemeral. | +| `watchSession`; watcher `subscribe`, `getMessages`, `close` | Integrated for bounded child live activity | A short-lived watcher attaches only to an explicitly selected active descendant. Its `getMessages` supplies one bounded, sanitized committed-message snapshot with assistant text and a coarse safe-label tool skeleton, then watcher message and tool lifecycle events maintain active-only live activity until the watcher closes. Native tool IDs are immediately reduced to attachment-salted in-memory correlation digests; arguments, results, reasoning, paths, timestamps, metadata, and error text never cross the boundary. The public watcher does not atomically expose attach-time streaming activity, so Pylon does not attempt a private-API workaround for a partial already in flight. This is distinct from, and does not enable, root transcript reads. | +| `getAgentMessageStatus`, `pauseAgentMessages`, `resumeAgentMessages`, `clearAgentMessages` | Intentionally excluded | These controls are daemon-global and can change or clear traffic belonging to unrelated sessions. | +| `startSideQuestion`, `abortSideQuestion` | Integrated as constrained transient quick questions | Supervised, fresh sessions may run one bounded tool-free question through a requester-owned unary RPC. Pylon uses separate public/native IDs, returns only one temporary answer, requests one bounded abort on cancellation, timeout, or disconnect, and never retries or persists the prompt, answer, native errors, or lifecycle. Full-access extension hooks, restored sessions, ACP, follow-up transcripts, and reconnect recovery fail closed. | +| `getHeartbeat`, `setHeartbeat`, `updateHeartbeat`, `listHeartbeats`, `listCronJobs`, `addCronJob`, `cancelCronJob`, `manageHeartbeat` | Deferred on lifecycle ownership | Scheduling promotes work to resident ownership, but public APIs do not provide Pylon an authoritative autonomous-turn/checkpoint identity, demotion, reattachment, or fail-safe delete flow. Shipping now could leave invisible mutations or orphaned work. | +| `getContextTree` | Blocked by upstream execution safety | Prime 0.8.0 synchronously follows and recursively scans unbounded `sub-*` directories before Pylon can decode or time out the result. Until Prime adds intrinsic symlink, cycle, depth, node, and byte bounds, Pylon uses bounded session stats and its own observed agent usage instead; native labels, IDs, model metadata, costs, and history remain private. | +| `getSessionContext`, `getSessionTree`, `getUserMessagesForForking`, `getLastAssistantText` | Intentionally redundant/sensitive | Pylon's event-sourced transcript and checkpoints are authoritative; mirroring Prime's private transcript/tree would create a second history source and expose hidden context. | +| `getSystemPrompt`, `getToolDefinition` | Intentionally excluded | These return hidden instructions, extension schemas, and prompt internals. | +| `listSavedSessions`, `newSession`, `switchSession`, `fork`, `navigateTree`, `importFromJsonl`, `exportToHtml`, `exportToJsonl`, `renameSavedSession`, `deleteSavedSession` | Deferred on history coordination | Public DTOs are filesystem/path-shaped and can enumerate unrelated Prime history. Native history mutation must first coordinate atomically with Pylon threads, worktrees, and checkpoints. | +| `setSessionName`, `setSessionEntryLabel` | Intentionally redundant | Pylon thread titles and durable activities are the user-visible source of truth. | +| `executeBash`, `executeBashAndWait`, `abortBash` | Intentionally redundant | Pylon's terminal and an agent turn have separate ownership and audit semantics; a raw session bash tunnel would bypass both. | +| `waitForHeadlessCompletion({ waitForRlmQuiescence })` | Integrated as an ordinary-turn barrier; autonomous ownership deferred | Daemon mode uses only the method's authoritative RLM ordering boundary after a Pylon-owned prompt and discards the autonomous-status result. This keeps descendants and their parent continuations inside the canonical Pylon turn without claiming native headless/resident turns. Those autonomous turns remain deferred until they have checkpoint identity, reattachment, stop, and deletion semantics. The valid boundary also supplies the cumulative usage delta used for terminal root, child, and continued-parent billing. ACP compatibility mode consumes 0.8.0's correlated terminal-quiescence metadata instead. | +| `getSessionHeader` | Intentionally redundant/sensitive | The header can repeat native saved-session identity and metadata. Pylon uses its private verified resume sidecar plus the durable thread projection instead of exposing or persisting a second header source. | +| `onBeforeSessionInvalidate` | Unavailable in daemon mode 0.8.0 | The public daemon implementation is a no-op returning only an unsubscribe function, so there is no lifecycle outcome to integrate. Pylon tears down its owned connection scope explicitly. | +| `promoteToResident` | Intentionally excluded until automation ownership exists | Client-owned workers must remain stoppable and reapable by their Pylon thread. | Prime Agent 0.8.0's goal-continuation and durable refinement-message changes remain native session behavior. Pylon continues to project only bounded goal state and aggregate refinement lifecycle; it does diff --git a/docs/user/providers-prime-agent.md b/docs/user/providers-prime-agent.md index 69ee75bda..9d4adbb69 100644 --- a/docs/user/providers-prime-agent.md +++ b/docs/user/providers-prime-agent.md @@ -122,6 +122,12 @@ native session rather than risking a partially reloaded runtime; it never retrie send resource paths, diagnostics, or extension source details to clients. Supervised sessions keep discovered commands disabled. Observed Prime subagents appear in Pylon's Agents hierarchy. In Full access, an active agent can be stopped from its Agents row on web or desktop, or from the **Agents** control on mobile. Pylon waits for Prime's native cancelled status instead of marking the agent stopped optimistically; completed output and activity remain in the thread. A cancellation racing natural completion is treated as already settled, and Pylon never retries an uncertain cancellation automatically. Supervised sessions do not offer this control because child-agent spawning is disabled. In Full access, a live agent with a native message endpoint can also receive a direct message from its Agents row. Pylon reports only whether Prime delivered the message immediately or queued it behind current work; that receipt does not mean the agent read, answered, or completed it. Pylon does not copy the message or Prime's receipt identifiers into its event store, activity history, diagnostics, or other clients. Prime necessarily adds the text to the selected child agent's private native transcript and context so the agent can act on it. Sending is never retried automatically; if delivery becomes uncertain, sending again may duplicate the message. Supervised and ACP sessions do not offer native agent messaging. +When a daemon-backed parent waits for asynchronous children, the Pylon turn stays **Working** until +Prime reports descendant quiescence and finishes any parent continuation triggered by their replies. +The continued parent answer appears in the same turn rather than as hidden background work. If the +daemon reconnects during this boundary or cannot confirm it, Pylon fails the turn and closes that native +session instead of reusing work whose ownership is uncertain. + In the main thread, each Prime tool call uses one activity row as it starts, updates, and completes. Pylon shows only a fixed friendly label such as **Code**, **Shell**, **Edit**, **Read**, **Search**, **Web search**, **Image**, or **Tool**, plus its coarse lifecycle state. Commands, code, paths, tool input, progress output, results, native titles and identifiers, and error text are not copied into thread activity. For an active agent, **Live activity** opens an on-demand view on web, desktop, or mobile. It is a bounded replacement snapshot from Prime's public live-session watcher, not a durable transcript: Pylon does not persist it in the thread, share it with clients that did not open the view, or keep it after the panel closes. The panel shows assistant text plus a coarse tool timeline containing only a friendly label and **Started**, **Completed**, or **Failed**; IPython appears as **Code**. It also repeats the safe aggregate status already shown in the agent roster, such as token or tool counts. Child prompts, tool arguments, partial and final results, thinking, paths, timestamps, native identifiers and metadata, error text, attachments, and usage details are excluded. **No activity yet** means the agent may still be thinking; it does not mean the agent is inactive. The subscription closes when the view closes, the agent exits, the thread or provider changes, or the client disconnects. Pylon can build a bounded coarse skeleton from committed messages returned by the public watcher, but Prime Agent 0.8.0 cannot reopen an exited child, provide lossless historical child activity, or atomically expose activity that was already streaming when the view opened, so Pylon labels the view **Live only** rather than implying complete history.