diff --git a/sdk/ARCHITECTURE.md b/sdk/ARCHITECTURE.md index 8bc95ed30b..06b3c87799 100644 --- a/sdk/ARCHITECTURE.md +++ b/sdk/ARCHITECTURE.md @@ -130,12 +130,16 @@ Completion telemetry is anchored to the assistant's explicit completion declaration, not session shutdown. After each agent turn, the local runtime inspects `AgentResult.toolCalls` and emits `task.completed` the moment a successful `submit_and_exit` (the SDK analog of original -Cline's `attempt_completion`) is observed. `shutdownSession(...)` -retains a fallback emission for completed sessions that finished -without an explicit completion-tool observation, so non-interactive -runs not using the yolo preset still produce a `task.completed` signal. -Each session emits at most one `task.completed`. See `DOC.md` for the -event payload and `source` field. +Cline's `attempt_completion`) is observed. A single teardown choke +point (`emitTaskCompletedOnTeardown(...)`) retains a fallback emission +for sessions whose final turn finished cleanly without an explicit +completion-tool observation (non-interactive runs not using the yolo +preset, or hosts that disable `submit_and_exit`). It is invoked from +every session exit path — both `shutdownSession(...)` and +`releaseSessionRuntime(...)` — so the emission never depends on which +teardown branch a stop routes through. Each session emits at most one +`task.completed`. See `DOC.md` for the event payload and `source` +field. ### Hub-Backed Runtime diff --git a/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts b/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts index 18fdca7625..3e766aa3b2 100644 --- a/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts +++ b/sdk/packages/core/src/runtime/host/local-runtime-host.test.ts @@ -6845,6 +6845,18 @@ describe("LocalRuntimeHost", () => { .map(([, payload]) => payload as Record); } + function getActiveSession( + manager: RuntimeHostUnderTest, + sessionId: string, + ): { status: string } { + const sessions = ( + manager as unknown as { sessions: Map } + ).sessions; + const session = sessions.get(sessionId); + if (!session) throw new Error("session was not registered"); + return session; + } + it("emits task.completed once with source=submit_and_exit when the assistant calls the completion tool in a non-interactive run", async () => { const sessionId = "sess-task-completed-submit-non-interactive"; const manifest = createManifest(sessionId); @@ -7117,6 +7129,287 @@ describe("LocalRuntimeHost", () => { expect(emissions).toHaveLength(1); expect(emissions[0]).toMatchObject({ source: "shutdown" }); }); + + it("emits task.completed exactly once when a stopped interactive session takes the release path after a clean final turn", async () => { + const sessionId = "sess-task-completed-release-path"; + const manifest = createManifest(sessionId); + const adapter = createTaskCompletedAdapter(); + const telemetry = new TelemetryService({ + adapters: [adapter], + distinctId, + }); + const sessionService = createMockSessionService(manifest); + const runtimeBuilder = { + build: vi.fn().mockReturnValue({ + tools: [], + shutdown: vi.fn(), + }), + }; + const agent = { + // Clean final turn without submit_and_exit — records + // lastInteractiveTurnFinishReason === "completed". + run: vi.fn().mockResolvedValue(createResult({ toolCalls: [] })), + continue: vi.fn(), + getMessages: vi.fn().mockReturnValue([]), + getAgentId: vi.fn().mockReturnValue("agent-root-1"), + getConversationId: vi.fn().mockReturnValue("conv-root-1"), + abort: vi.fn(), + subscribeEvents: vi.fn().mockReturnValue(() => {}), + canStartRun: vi.fn().mockReturnValue(true), + shutdown: vi.fn().mockResolvedValue(undefined), + }; + const manager = new RuntimeHostUnderTest({ + distinctId, + sessionService: sessionService as never, + runtimeBuilder: runtimeBuilder as never, + createAgent: () => agent as never, + telemetry, + }); + + await manager.startSession( + normalizeStartInput({ + config: createConfig({ telemetry, sessionId }), + prompt: "finish the task cleanly", + interactive: true, + }), + ); + + // Truthful status reporting can leave a resident interactive session + // with a terminal reported status (e.g. adopted from a resumed + // manifest), which routes stopSession through releaseSessionRuntime + // instead of shutdownSession. + getActiveSession(manager, sessionId).status = "completed"; + const statusWritesBeforeStop = + sessionService.updateSessionStatus.mock.calls.length; + await manager.stopSession(sessionId); + // The release branch never writes a session status — this proves the + // stop really took the path that used to drop the emission. + expect(sessionService.updateSessionStatus.mock.calls.length).toBe( + statusWritesBeforeStop, + ); + + const emissions = countTaskCompletedEmissions(adapter); + expect(emissions).toHaveLength(1); + expect(emissions[0]).toMatchObject({ + ulid: sessionId, + source: "shutdown", + provider: "mock-provider", + modelId: "mock-model", + }); + }); + + it("emits nothing when a released interactive session's final turn aborted", async () => { + const sessionId = "sess-task-completed-release-aborted"; + const manifest = createManifest(sessionId); + const adapter = createTaskCompletedAdapter(); + const telemetry = new TelemetryService({ + adapters: [adapter], + distinctId, + }); + const sessionService = createMockSessionService(manifest); + const runtimeBuilder = { + build: vi.fn().mockReturnValue({ + tools: [], + shutdown: vi.fn(), + }), + }; + const agent = { + run: vi + .fn() + .mockResolvedValue( + createResult({ finishReason: "aborted", toolCalls: [] }), + ), + continue: vi.fn(), + getMessages: vi.fn().mockReturnValue([]), + getAgentId: vi.fn().mockReturnValue("agent-root-1"), + getConversationId: vi.fn().mockReturnValue("conv-root-1"), + abort: vi.fn(), + subscribeEvents: vi.fn().mockReturnValue(() => {}), + canStartRun: vi.fn().mockReturnValue(true), + shutdown: vi.fn().mockResolvedValue(undefined), + }; + const manager = new RuntimeHostUnderTest({ + distinctId, + sessionService: sessionService as never, + runtimeBuilder: runtimeBuilder as never, + createAgent: () => agent as never, + telemetry, + }); + + await manager.startSession( + normalizeStartInput({ + config: createConfig({ telemetry, sessionId }), + prompt: "turn that gets aborted", + interactive: true, + }), + ); + + getActiveSession(manager, sessionId).status = "cancelled"; + await manager.stopSession(sessionId); + + expect(countTaskCompletedEmissions(adapter)).toHaveLength(0); + }); + + it("emits nothing when a session's final turn errors after an earlier clean turn", async () => { + const sessionId = "sess-task-completed-late-error"; + const manifest = createManifest(sessionId); + const adapter = createTaskCompletedAdapter(); + const telemetry = new TelemetryService({ + adapters: [adapter], + distinctId, + }); + const sessionService = createMockSessionService(manifest); + const runtimeBuilder = { + build: vi.fn().mockReturnValue({ + tools: [], + shutdown: vi.fn(), + }), + }; + const agent = { + // First turn completes cleanly, second turn throws — the errored + // turn is the session's final turn, so no task.completed may be + // emitted from the stale "completed" of the first turn. + run: vi.fn().mockResolvedValue(createResult({ toolCalls: [] })), + continue: vi.fn().mockRejectedValue(new Error("provider exploded")), + getMessages: vi.fn().mockReturnValue([]), + getAgentId: vi.fn().mockReturnValue("agent-root-1"), + getConversationId: vi.fn().mockReturnValue("conv-root-1"), + abort: vi.fn(), + subscribeEvents: vi.fn().mockReturnValue(() => {}), + canStartRun: vi.fn().mockReturnValue(true), + shutdown: vi.fn().mockResolvedValue(undefined), + }; + const manager = new RuntimeHostUnderTest({ + distinctId, + sessionService: sessionService as never, + runtimeBuilder: runtimeBuilder as never, + createAgent: () => agent as never, + telemetry, + }); + + await manager.startSession( + normalizeStartInput({ + config: createConfig({ telemetry, sessionId }), + prompt: "first turn finishes cleanly", + interactive: true, + }), + ); + await expect( + manager.runTurn({ sessionId, prompt: "second turn blows up" }), + ).rejects.toThrow("provider exploded"); + + expect(countTaskCompletedEmissions(adapter)).toHaveLength(0); + }); + + it("does not double-fire from the release path after submit_and_exit already emitted", async () => { + const sessionId = "sess-task-completed-release-no-double-fire"; + const manifest = createManifest(sessionId); + const adapter = createTaskCompletedAdapter(); + const telemetry = new TelemetryService({ + adapters: [adapter], + distinctId, + }); + const sessionService = createMockSessionService(manifest); + const runtimeBuilder = { + build: vi.fn().mockReturnValue({ + tools: [], + shutdown: vi.fn(), + }), + }; + const agent = { + run: vi.fn().mockResolvedValue( + createResult({ + toolCalls: [createSubmitAndExitToolCall()], + }), + ), + continue: vi.fn(), + getMessages: vi.fn().mockReturnValue([]), + getAgentId: vi.fn().mockReturnValue("agent-root-1"), + getConversationId: vi.fn().mockReturnValue("conv-root-1"), + abort: vi.fn(), + subscribeEvents: vi.fn().mockReturnValue(() => {}), + canStartRun: vi.fn().mockReturnValue(true), + shutdown: vi.fn().mockResolvedValue(undefined), + }; + const manager = new RuntimeHostUnderTest({ + distinctId, + sessionService: sessionService as never, + runtimeBuilder: runtimeBuilder as never, + createAgent: () => agent as never, + telemetry, + }); + + await manager.startSession( + normalizeStartInput({ + config: createConfig({ telemetry, sessionId }), + prompt: "complete via submit_and_exit", + interactive: true, + }), + ); + + getActiveSession(manager, sessionId).status = "completed"; + await manager.stopSession(sessionId); + + const emissions = countTaskCompletedEmissions(adapter); + expect(emissions).toHaveLength(1); + expect(emissions[0]).toMatchObject({ + ulid: sessionId, + source: "submit_and_exit", + }); + }); + + it("emits task.completed exactly once when dispose() releases a cleanly finished interactive session", async () => { + const sessionId = "sess-task-completed-dispose-release"; + const manifest = createManifest(sessionId); + const adapter = createTaskCompletedAdapter(); + const telemetry = new TelemetryService({ + adapters: [adapter], + distinctId, + }); + const sessionService = createMockSessionService(manifest); + const runtimeBuilder = { + build: vi.fn().mockReturnValue({ + tools: [], + shutdown: vi.fn(), + }), + }; + const agent = { + run: vi.fn().mockResolvedValue(createResult({ toolCalls: [] })), + continue: vi.fn(), + getMessages: vi.fn().mockReturnValue([]), + getAgentId: vi.fn().mockReturnValue("agent-root-1"), + getConversationId: vi.fn().mockReturnValue("conv-root-1"), + abort: vi.fn(), + subscribeEvents: vi.fn().mockReturnValue(() => {}), + canStartRun: vi.fn().mockReturnValue(true), + shutdown: vi.fn().mockResolvedValue(undefined), + }; + const manager = new RuntimeHostUnderTest({ + distinctId, + sessionService: sessionService as never, + runtimeBuilder: runtimeBuilder as never, + createAgent: () => agent as never, + telemetry, + }); + + await manager.startSession( + normalizeStartInput({ + config: createConfig({ telemetry, sessionId }), + prompt: "finish then get disposed", + interactive: true, + }), + ); + + getActiveSession(manager, sessionId).status = "completed"; + await manager.dispose("hub_restart"); + + const emissions = countTaskCompletedEmissions(adapter); + expect(emissions).toHaveLength(1); + expect(emissions[0]).toMatchObject({ + ulid: sessionId, + source: "shutdown", + }); + }); }); describe("LocalRuntimeHost releasing a session mid-run", () => { diff --git a/sdk/packages/core/src/runtime/host/local-runtime-host.ts b/sdk/packages/core/src/runtime/host/local-runtime-host.ts index 6db5d2dda8..9fc01f784c 100644 --- a/sdk/packages/core/src/runtime/host/local-runtime-host.ts +++ b/sdk/packages/core/src/runtime/host/local-runtime-host.ts @@ -889,6 +889,7 @@ export class LocalRuntimeHost implements RuntimeHost { drainingPendingPrompts: false, pluginSandboxShutdown: bootstrap.pluginSandboxShutdown, submitAndExitObserved: false, + taskCompletedEmitted: false, lastInteractiveTurnFinishReason: undefined, }; activeSessionRef = active; @@ -1966,10 +1967,11 @@ export class LocalRuntimeHost implements RuntimeHost { * `attempt_completion`-driven emission and works for both interactive * and non-interactive sessions. * - * `shutdownSession(...)` retains a fallback emission for completed - * sessions that finish without an explicit completion-tool observation - * (e.g., non-interactive runs not using the yolo preset). This helper - * sets `submitAndExitObserved` so the shutdown fallback can suppress a + * `emitTaskCompletedOnTeardown(...)` retains a fallback emission for + * completed sessions that finish without an explicit completion-tool + * observation (e.g., non-interactive runs not using the yolo preset, + * or hosts that disable `submit_and_exit` entirely). This helper sets + * `taskCompletedEmitted` so the teardown fallback can suppress a * duplicate emission for the same logical completion. */ private observeTaskCompletionTool( @@ -1984,6 +1986,7 @@ export class LocalRuntimeHost implements RuntimeHost { ); if (!completedWithSubmitAndExit) return; session.submitAndExitObserved = true; + session.taskCompletedEmitted = true; captureTaskCompleted(session.config.telemetry, { ulid: session.sessionId, provider: session.config.providerId, @@ -1995,6 +1998,54 @@ export class LocalRuntimeHost implements RuntimeHost { }); } + /** + * Single choke point for the fallback `task.completed` emission on + * session teardown. Every path a session can end through funnels into + * `shutdownSession(...)` or `releaseSessionRuntime(...)`, and BOTH must + * call this helper — the emission must never depend on which teardown + * branch a stop happens to route through. (The 4.1.11 regression: + * truthful session-status reporting re-routed many interactive stops + * onto the release branch, and the fallback that lived only inside + * `shutdownSession` silently stopped firing for them.) + * + * Emits at most once per session (`taskCompletedEmitted`), and never + * after the `submit_and_exit` observer already reported the completion. + * The completion criterion deliberately does not read `session.status` + * (whose lifecycle is what changed in 4.1.11): + * + * - Interactive sessions use the recorded final-turn outcome, + * `lastInteractiveTurnFinishReason === "completed"`. The extra guards + * suppress emission when teardown arrives mid-run (the in-flight turn + * being aborted is the real final turn, and it did not complete). + * - Non-interactive sessions use the terminal status their run result + * resolved to (`finalStatus`, from `finalizeSingleRun`), preserving + * the pre-existing `input.status === "completed"` semantics. + * + * Sessions whose final turn errored or aborted emit nothing. + */ + private emitTaskCompletedOnTeardown( + session: ActiveSession, + finalStatus?: SessionStatus, + ): void { + if (session.taskCompletedEmitted || session.submitAndExitObserved) return; + const completedCleanly = session.interactive + ? session.lastInteractiveTurnFinishReason === "completed" && + !session.aborting && + session.agent.canStartRun() + : finalStatus === "completed"; + if (!completedCleanly) return; + session.taskCompletedEmitted = true; + captureTaskCompleted(session.config.telemetry, { + ulid: session.sessionId, + provider: session.config.providerId, + modelId: session.config.modelId, + mode: session.config.mode, + durationMs: Date.now() - Date.parse(session.startedAt), + source: "shutdown", + ...this.getSessionAgentTelemetryIdentity(session), + }); + } + private async prepareTurnInput( session: ActiveSession, input: { @@ -2173,6 +2224,11 @@ export class LocalRuntimeHost implements RuntimeHost { } private async failSession(session: ActiveSession): Promise { + // The failing turn is this session's final turn. Record it so the + // teardown completion criterion (`lastInteractiveTurnFinishReason`) + // cannot read a stale "completed" left over from an earlier + // successful turn and emit `task.completed` for an errored session. + session.lastInteractiveTurnFinishReason = "error"; await this.shutdownSession(session, { status: "failed", exitCode: 1, @@ -2190,21 +2246,11 @@ export class LocalRuntimeHost implements RuntimeHost { endReason: string; }, ): Promise { - // Fallback `task.completed` emission for completed sessions that - // did not observe an explicit `submit_and_exit` tool call. The - // observer in `executeAgentTurn(...)` already emitted the event in - // that case, so we suppress here to avoid double-counting. - if (input.status === "completed" && !session.submitAndExitObserved) { - captureTaskCompleted(session.config.telemetry, { - ulid: session.sessionId, - provider: session.config.providerId, - modelId: session.config.modelId, - mode: session.config.mode, - durationMs: Date.now() - Date.parse(session.startedAt), - source: "shutdown", - ...this.getSessionAgentTelemetryIdentity(session), - }); - } + // Fallback `task.completed` emission for completed sessions that did + // not observe an explicit `submit_and_exit` tool call, routed through + // the shared teardown choke point so it can neither double-fire nor + // be skipped by teardown routing. + this.emitTaskCompletedOnTeardown(session, input.status); notifyTeamRunWaiters(session); // Drain an in-flight run before tearing anything down. `stopSession` aborts @@ -2286,6 +2332,12 @@ export class LocalRuntimeHost implements RuntimeHost { session: ActiveSession, reason: string, ): Promise { + // Releasing is a full session exit too: interactive sessions whose + // reported status is already terminal are stopped/disposed through + // this branch. The completion emission must happen here as well — + // this is the branch that silently dropped `task.completed` when + // truthful status reporting re-routed interactive stops onto it. + this.emitTaskCompletedOnTeardown(session); const cleanupErrors: unknown[] = []; const recordCleanupError = (stage: string, error: unknown) => { cleanupErrors.push(error); diff --git a/sdk/packages/core/src/types/session.ts b/sdk/packages/core/src/types/session.ts index 972f8f13f7..b929a94a1f 100644 --- a/sdk/packages/core/src/types/session.ts +++ b/sdk/packages/core/src/types/session.ts @@ -48,14 +48,21 @@ export type ActiveSession = { * declares completion (parity with original Cline's * `attempt_completion`). * 2. Suppress the fallback `task.completed` emission from - * `shutdownSession(...)` so the same logical completion is not - * reported twice. + * `emitTaskCompletedOnTeardown(...)` so the same logical completion + * is not reported twice. * * Non-interactive sessions that finish without ever calling the - * completion tool still receive a `task.completed` from the shutdown + * completion tool still receive a `task.completed` from the teardown * fallback. */ submitAndExitObserved: boolean; + /** + * Set to `true` the moment `task.completed` is emitted for this session, + * whether by the `submit_and_exit` observer or by the teardown fallback + * (`emitTaskCompletedOnTeardown`). Enforces the invariant of exactly one + * `task.completed` per session regardless of which teardown path runs. + */ + taskCompletedEmitted: boolean; }; export type PendingPrompt = {