diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 18ce3954a6..dd74c5c00d 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -23,6 +23,7 @@ - Defense-in-depth: when an Anthropic-origin assistant transcript message carrying directly adjacent `thinking`/`redacted_thinking` blocks is persisted, a single bounded warn is emitted per session manager instance — but only in development/test builds, never in production. The diagnostic names only the envelope shape (block count, adjacency presence, provider), never raw thinking text, signatures, redacted payloads, or transcript-path metadata. Storage is never mutated — the send-boundary collapse remains the wire source of truth; this is a read-only observation that helps surface upstream producers of the rejected shape (#4443). ### Fixed - Cursor split responses now preserve canonical provider order across pre-admission artifact spilling, so a server-side tool result cannot be persisted after the assistant continuation that follows it. The canonical admission reservation is owned by each emission's own handler, so a host bridge or replay emitting the same `message_end` event object twice can no longer overwrite the reservation and deadlock every later canonical admission (#4536). +- Uncontended canonical `message_end` admissions no longer cost a microtask: the admission lane keeps a synchronous fast path when its predecessor slot is already released, restoring synchronous visibility of persisted appends for external emitters (successor-finalization and deep-interview continuation flows) while FIFO admission still holds under a genuinely in-flight predecessor (#4536). - Default-model selection now reserves a causal fence before credential probing, so an already accepted prompt preflight cannot be overtaken and a selection accepted first blocks later prompt preflight through durable publication. The fence does not hold session admission across `waitForIdle`, allowing inherited auto-compaction continuations to obtain prompt admission; same-session reentrancy still fails fast, successor sessions remain protected, and disposal deterministically drains accepted selections while rejecting queued prompts without unhandled rejections (#4519). - Ordinary sessions no longer import or execute Claude Code and Codex directory hooks as competing runtime authorities. Runtime hook discovery is fail-closed to canonical native `.gjc/hooks/` providers while explicit configured paths, constrained plugin hooks, and foreign-provider import/diagnostic discovery remain available (#4516). - Telegram/Slack/Discord outbound publications no longer freeze after a session-host rehost. A rehosted fleet re-attaches every session in one reconcile pass, and each attachment's initial `event_replay` was awaited inside the serialized `#reconcileTail`, so one slow replay (up to its full retry budget) wedged all later reconciles and the sends funneling through them; leases and inbound polling stayed green while delivery died until daemon restart. Reconcile-driven attachments now publish immediately and run initial replay on the attachment's ready tail (matching the reconnect path); `start()` still drains those tails so bootstrap callers observe replay completion. Replay ordering, generation fences, cross-session isolation, and provider hooks are unchanged (#4527). diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 047716b27a..aadb5147fb 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1929,6 +1929,22 @@ function deobfuscateSessionContext(context: SessionContext, obfuscator: SecretOb return { ...context, messages }; } +/** + * Canonical message_end admission slot. A reservation whose slot is already + * `released` needs no await at the canonical append site, so an uncontended + * admission never costs a microtask and external emitters keep synchronous + * visibility of the persisted append. + */ +interface CanonicalMessageAdmissionSlot { + promise: Promise; + released: boolean; +} + +interface CanonicalMessageAdmission { + predecessor: CanonicalMessageAdmissionSlot; + release: () => void; +} + export class AgentSession { #provisionalStreamingToolCallIds = new Set(); readonly agent: Agent; @@ -4312,25 +4328,25 @@ export class AgentSession { this.#coordinatorToolObservations.set(event, Object.freeze({ label, observedAt: new Date().toISOString() })); } - #canonicalMessageAdmissionTail: Promise = Promise.resolve(); + #canonicalMessageAdmissionTail: CanonicalMessageAdmissionSlot = { promise: Promise.resolve(), released: true }; - #reserveCanonicalMessageAdmission( - event: AgentEvent, - ): { predecessor: Promise; release: () => void } | undefined { + #reserveCanonicalMessageAdmission(event: AgentEvent): CanonicalMessageAdmission | undefined { if (event.type !== "message_end") return undefined; const predecessor = this.#canonicalMessageAdmissionTail; const settled = Promise.withResolvers(); + const slot: CanonicalMessageAdmissionSlot = { promise: settled.promise, released: false }; let released = false; const release = () => { if (released) return; released = true; + slot.released = true; settled.resolve(); }; // The reservation is owned by this emission's handler: keying it by the // event object would let a replayed/bridged duplicate emission overwrite // it and leave the first handler awaiting a promise only its own handler // will ever release. - this.#canonicalMessageAdmissionTail = settled.promise; + this.#canonicalMessageAdmissionTail = slot; return { predecessor, release }; } @@ -4544,6 +4560,10 @@ export class AgentSession { // Track last assistant message for auto-compaction check #lastAssistantMessage: AssistantMessage | undefined = undefined; + // Admission slot of the last message_end per assistant message, so the + // agent_end handler can join the terminal's canonical admission before any + // post-turn write reaches the branch. + #lastAssistantAdmissionByMessage = new WeakMap(); // Provider context construction must wait for this chain. Agent event listeners // are synchronous dispatch only; their async work cannot otherwise gate the // next tool-result provider request. @@ -4614,7 +4634,7 @@ export class AgentSession { #handleAgentEvent = async ( event: AgentEvent, activePromptHandle?: string, - canonicalAdmission?: { predecessor: Promise; release: () => void }, + canonicalAdmission?: CanonicalMessageAdmission, ): Promise => { const attemptScope = (event as AgentEvent & { scope?: AttemptScope }).scope; @@ -4765,8 +4785,28 @@ export class AgentSession { // Canonical persistence follows synchronous message_end reservation order. // Only the admission predecessor and this event's own pre-admission work are // inside the lane; release before extension delivery and unrelated post-work. + // An already-released predecessor must not cost a microtask: external emitters + // and tests rely on canonical append being visible synchronously after + // emitExternalEvent returns whenever no admission is actually contended. + // Track the terminal assistant synchronously, before any admission wait: + // externally emitted terminals (host bridges, replays, tests) dispatch + // agent_end immediately after message_end, and the agent_end handler's + // post-turn read must see THIS stop even when this admission is still + // parked behind a contended predecessor — otherwise post-turn logic + // (deep-interview continuation, compaction, retry classification) runs + // against the previous turn's assistant. The per-message admission slot + // also lets agent_end processing wait for this admission to finish, so + // post-turn writes (continuation reminders, compaction rewrites) never + // reorder ahead of the branch entries they respond to. + if (event.type === "message_end" && event.message.role === "assistant") { + this.#lastAssistantMessage = event.message; + this.#lastAssistantAdmissionByMessage.set(event.message, canonicalAdmission); + } + if (event.type === "message_end") { - await canonicalAdmission?.predecessor; + if (canonicalAdmission && !canonicalAdmission.predecessor.released) { + await canonicalAdmission.predecessor.promise; + } if ( (event.message.role === "hookMessage" || event.message.role === "custom") && !(event.message.role === "custom" && event.message.customType === "hindsight-recall") @@ -5082,9 +5122,9 @@ export class AgentSession { this.#markTtsrInjected(this.#extractTtsrRuleNames(event.message.details)); } - // Track assistant message for auto-compaction (checked on agent_end) + // (#lastAssistantMessage is captured synchronously before the admission + // wait above; the block below handles assistant side effects only.) if (event.message.role === "assistant") { - this.#lastAssistantMessage = event.message; const assistantMsg = event.message as AssistantMessage; const currentGrantsAnthropicPriority = this.serviceTier === "priority" || this.serviceTier === "claude-only"; @@ -5268,6 +5308,16 @@ export class AgentSession { .find((message): message is AssistantMessage => message.role === "assistant"); const msg = this.#lastAssistantMessage ?? fallbackAssistant; this.#lastAssistantMessage = undefined; + // Join the terminal's canonical admission before any post-turn write: + // an externally emitted terminal dispatches agent_end while its own + // admission may still be parked behind a contended predecessor, and a + // continuation reminder or compaction rewrite that runs first would + // persist ahead of the branch entries it responds to. + const terminalAdmission = msg ? this.#lastAssistantAdmissionByMessage.get(msg) : undefined; + if (msg) this.#lastAssistantAdmissionByMessage.delete(msg); + if (terminalAdmission && !terminalAdmission.predecessor.released) { + await terminalAdmission.predecessor.promise; + } if (!msg) { this.#lastSuccessfulYieldToolCallId = undefined; this.#resolveRetry(); diff --git a/packages/coding-agent/test/agent-session-contended-terminal-capture.test.ts b/packages/coding-agent/test/agent-session-contended-terminal-capture.test.ts new file mode 100644 index 0000000000..30e7a9c94e --- /dev/null +++ b/packages/coding-agent/test/agent-session-contended-terminal-capture.test.ts @@ -0,0 +1,148 @@ +import { afterEach, describe, expect, it, vi } from "bun:test"; +import * as path from "node:path"; +import { Agent } from "@gajae-code/agent-core"; +import type { AssistantMessage, ToolResultMessage } from "@gajae-code/ai"; +import { getBundledModel } from "@gajae-code/ai/models"; +import { ModelRegistry } from "@gajae-code/coding-agent/config/model-registry"; +import { Settings } from "@gajae-code/coding-agent/config/settings"; +import { ensureWorkflowSkillActivationState } from "@gajae-code/coding-agent/hooks/skill-state"; +import { AgentSession } from "@gajae-code/coding-agent/session/agent-session"; +import { AuthStorage } from "@gajae-code/coding-agent/session/auth-storage"; +import { SessionManager } from "@gajae-code/coding-agent/session/session-manager"; +import { TempDir, withTimeout } from "@gajae-code/utils"; + +describe("AgentSession contended terminal assistant capture (#4565 finding)", () => { + let tempDir: TempDir | undefined; + let session: AgentSession | undefined; + let authStorage: AuthStorage | undefined; + let sessionManager: SessionManager | undefined; + + afterEach(async () => { + vi.restoreAllMocks(); + await session?.dispose(); + authStorage?.close(); + tempDir?.removeSync(); + }); + + it("still schedules the deep-interview continuation when the terminal's admission is contended behind a gated spill", async () => { + tempDir = TempDir.createSync("@gjc-contended-di-capture-"); + authStorage = await AuthStorage.create(path.join(tempDir.path(), "auth.db")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + const model = getBundledModel("anthropic", "claude-sonnet-4-5"); + if (!model) throw new Error("Expected bundled Anthropic model"); + + // Gate the oversized tool result's spill so the terminal assistant's + // admission predecessor stays in flight: the contended case the + // synchronous fast path cannot shortcut. + const spillGate = Promise.withResolvers(); + const agent = new Agent({ + initialState: { + model: { ...model, contextWindow: 200_000, maxTokens: 128_000 }, + systemPrompt: ["Test"], + tools: [], + messages: [], + }, + }); + sessionManager = SessionManager.inMemory(tempDir.path()); + (sessionManager as unknown as { saveArtifact: typeof sessionManager.saveArtifact }).saveArtifact = async () => { + await spillGate.promise; + return { uri: "artifact://1", bytes: 4, sha256: "0".repeat(64) } as never; + }; + session = new AgentSession({ + agent, + sessionManager, + settings: Settings.isolated({ "tools.preAdmissionArtifactSpill": true }), + modelRegistry: new ModelRegistry(authStorage), + }); + await ensureWorkflowSkillActivationState({ + cwd: tempDir.path(), + skill: "deep-interview", + sessionId: sessionManager.getSessionId(), + }); + + const continueSpy = vi.spyOn(agent, "continue").mockImplementation(async () => Promise.resolve()) as never; + + const mkAssistant = ( + text: string, + timestamp: number, + stopReason: AssistantMessage["stopReason"], + withToolCall = false, + ): AssistantMessage => ({ + role: "assistant", + content: withToolCall + ? [ + { type: "text", text }, + { type: "toolCall", id: `call-${timestamp}`, name: "read", arguments: { path: "/tmp/x" } }, + ] + : [{ type: "text", text }], + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason, + timestamp, + }); + + // Previous turn: a mid-loop toolUse assistant. If the agent_end read + // resolves to THIS stale capture, hasToolCalls short-circuits the stop + // handling and the terminal's deep-interview continuation is silently + // skipped — exactly the #4565 double-process/skip failure mode. + const midLoop = mkAssistant("mid-loop turn", 100, "toolUse", true); + const gatedToolResult: ToolResultMessage = { + role: "toolResult", + toolCallId: "call-100", + toolName: "read", + content: [{ type: "text", text: "z".repeat(40_000) }], + isError: false, + timestamp: Date.now(), + }; + const terminal = mkAssistant("interview round complete", 200, "stop"); + + agent.emitExternalEvent({ type: "turn_start" }); + agent.emitExternalEvent({ type: "message_end", message: midLoop }); + await Bun.sleep(10); + agent.emitExternalEvent({ type: "message_end", message: gatedToolResult }); + await Bun.sleep(10); + agent.emitExternalEvent({ type: "message_end", message: terminal }); + await Bun.sleep(25); + + // Terminal agent_end arrives while the terminal admission is parked. + agent.emitExternalEvent({ type: "agent_end", messages: [terminal] }); + + // The deep-interview stop check runs async durable state reads; allow + // them, then release the gate and settle. + await Bun.sleep(50); + spillGate.resolve(); + await withTimeout(session.awaitSessionSettlement(), 5_000, "gated admission deadlocked"); + await session.waitForIdle(); + + // The terminal stop must have reached the deep-interview stop gate and + // scheduled its continuation. Stale capture -> hasToolCalls -> no call. + expect(continueSpy).toHaveBeenCalled(); + const entries = sessionManager.getBranch(); + const reminderIndex = entries.findIndex( + entry => + entry.type === "message" && + entry.message.role === "developer" && + JSON.stringify(entry.message.content).includes("stop gate: gjc_skill_deep_interview_"), + ); + expect(reminderIndex).toBeGreaterThanOrEqual(0); + // FIFO on reload: the continuation reminder must persist AFTER the + // terminal assistant (and the gated tool result) it responds to. + const terminalIndex = entries.findIndex( + entry => + entry.type === "message" && + entry.message.role === "assistant" && + JSON.stringify((entry.message as { content?: unknown }).content).includes("interview round complete"), + ); + expect(terminalIndex).toBeGreaterThanOrEqual(0); + expect(reminderIndex).toBeGreaterThan(terminalIndex); + }); +}); diff --git a/packages/coding-agent/test/agent-session-midrun-maintenance.test.ts b/packages/coding-agent/test/agent-session-midrun-maintenance.test.ts index b8be4ac520..924afb83b1 100644 --- a/packages/coding-agent/test/agent-session-midrun-maintenance.test.ts +++ b/packages/coding-agent/test/agent-session-midrun-maintenance.test.ts @@ -521,13 +521,15 @@ describe("AgentSession mid-run maintenance outcomes", () => { { role: "user", content: "second distinct steering", timestamp: Date.now() }, ]); - // With protectRecentTurns: 2 (default), the session manager's canonical - // entry ordering places the large orphan tool results inside the fence - // window, so they are not eligible for pruning — maintenance falls - // through to compaction. The short-circuit extension produces a - // compaction entry that preserves the recent paired result and steering. + // Canonical entry order equals emission order, so the three large orphan + // tool results land before the recent user turns and sit outside the + // protectRecentTurns: 2 fence. They are therefore prunable, and prune — + // the cheaper preferred rewrite — legitimately wins over compaction. + // The assertions below prove the fence classification is what changed: + // the protected paired result and both steering messages survive the + // rewrite, and the codex provider epoch is still reset. const outcome = await session.runMidRunMaintenanceForTests(contextOf(session)); - expect(outcome).toBe("compacted"); + expect(outcome).toBe("pruned"); const persisted = session.sessionManager .getBranch() .flatMap(entry => diff --git a/packages/coding-agent/test/agent-session-pre-admission-artifact-spill.test.ts b/packages/coding-agent/test/agent-session-pre-admission-artifact-spill.test.ts index 1a3a421486..b7bd980bac 100644 --- a/packages/coding-agent/test/agent-session-pre-admission-artifact-spill.test.ts +++ b/packages/coding-agent/test/agent-session-pre-admission-artifact-spill.test.ts @@ -281,4 +281,114 @@ describe("AgentSession pre-admission artifact spill", () => { .filter(entry => entry.type === "message" && entry.message.role === "toolResult"); expect(persistedToolResults).toHaveLength(2); }); + it("appends canonical messages synchronously when no admission is contended", async () => { + tempDir = TempDir.createSync("@gjc-admission-sync-visibility-"); + const model = getBundledModel("anthropic", "claude-sonnet-4-5"); + if (!model) throw new Error("Expected bundled Anthropic model"); + const agent = new Agent({ + initialState: { + model: { ...model, contextWindow: 200_000, maxTokens: 128_000 }, + systemPrompt: ["Test"], + tools: [], + messages: [], + }, + }); + const sessionManager = SessionManager.inMemory(tempDir.path()); + session = new AgentSession({ + agent, + sessionManager, + settings: Settings.isolated({ "tools.preAdmissionArtifactSpill": true }), + modelRegistry: {} as never, + }); + + // An uncontended message_end (predecessor already released) must not cost + // a microtask: hosts emit external messages and read the persisted branch + // immediately (issue #2261 flow). A regression to an unconditional await + // breaks this contract even though the promise is already settled. + agent.emitExternalEvent({ + type: "message_end", + message: { role: "user", content: "sync visible", timestamp: Date.now() }, + }); + const lastEntry = sessionManager.getBranch().at(-1); + expect(lastEntry?.type).toBe("message"); + expect(lastEntry?.type === "message" && (lastEntry.message as { content?: unknown }).content).toBe( + "sync visible", + ); + }); + + it("preserves FIFO admission when a contended predecessor is still in flight", async () => { + tempDir = TempDir.createSync("@gjc-admission-fifo-"); + authStorage = await AuthStorage.create(path.join(tempDir.path(), "auth.db")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + const model = getBundledModel("anthropic", "claude-sonnet-4-5"); + if (!model) throw new Error("Expected bundled Anthropic model"); + + // Gate the spill so the first tool result's admission is still pending when + // the assistant continuation is emitted; the continuation must wait for it. + const spillGate = Promise.withResolvers(); + const agent = new Agent({ + initialState: { + model: { ...model, contextWindow: 200_000, maxTokens: 128_000 }, + systemPrompt: ["Test"], + tools: [], + messages: [], + }, + }); + const sessionManager = SessionManager.inMemory(tempDir.path()); + (sessionManager as unknown as { saveArtifact: typeof sessionManager.saveArtifact }).saveArtifact = async () => { + await spillGate.promise; + return { uri: "artifact://1", bytes: 4, sha256: "0".repeat(64) } as never; + }; + session = new AgentSession({ + agent, + sessionManager, + settings: Settings.isolated({ "tools.preAdmissionArtifactSpill": true }), + modelRegistry: new ModelRegistry(authStorage), + }); + + const toolResult: ToolResultMessage = { + role: "toolResult", + toolCallId: "fifo-gated", + toolName: "read", + content: [{ type: "text", text: "y".repeat(40_000) }], + isError: false, + timestamp: Date.now(), + }; + agent.emitExternalEvent({ type: "message_end", message: toolResult }); + agent.emitExternalEvent({ + type: "message_end", + message: { + role: "assistant", + content: [{ type: "text", text: "continuation after gated spill" }], + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + } as AssistantMessage, + }); + + await Bun.sleep(25); + // While the spill is gated, neither message may have appended out of order; + // the continuation is blocked behind the tool result's pending admission. + const rolesSoFar = sessionManager + .getBranch() + .flatMap(entry => (entry.type === "message" ? [entry.message.role] : [])); + expect(rolesSoFar).toEqual([]); + + spillGate.resolve(); + await withTimeout(session.awaitSessionSettlement(), 5_000, "Gated spill admission deadlocked"); + const rolesAfter = sessionManager + .getBranch() + .flatMap(entry => (entry.type === "message" ? [entry.message.role] : [])); + expect(rolesAfter).toEqual(["toolResult", "assistant"]); + }); });