Fix bottom anchoring and Claude wake routing

This commit is contained in:
Mohamed Boudra
2026-03-08 12:54:08 +07:00
parent 0c44e7db80
commit b339c5e61d
27 changed files with 3801 additions and 326 deletions
@@ -2,7 +2,10 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js";
import type {
AgentPersistenceHandle,
AgentStreamEvent,
} from "../agent-sdk-types.js";
type Deferred<T> = {
promise: Promise<T>;
@@ -297,7 +300,7 @@ describe("ClaudeAgentSession interrupt restart regression", () => {
expect(sdkMocks.firstQuery).toBeTruthy();
expect(sdkMocks.secondQuery).toBeTruthy();
expect(sdkMocks.firstQuery).not.toBe(sdkMocks.secondQuery);
expect(sdkMocks.firstQuery?.interrupt).toHaveBeenCalledTimes(2);
expect(sdkMocks.firstQuery?.interrupt).toHaveBeenCalledTimes(1);
expect(sdkMocks.secondQuery?.next).toHaveBeenCalled();
expect(secondAssistantText).toContain("NEW_TURN_RESPONSE");
expect(secondAssistantText).not.toContain("OLD_TURN_RESPONSE");
@@ -979,6 +982,146 @@ describe("ClaudeAgentSession interrupt restart regression", () => {
await session.close();
});
test("ignores unmatched resumed-session errors without starting an autonomous run", async () => {
const logger = createTestLogger();
const keepQueryAlive = deferred<void>();
sdkMocks.query.mockImplementation(() => {
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "error_during_execution",
session_id: "new-session-after-missing-conversation",
errors: [
"No conversation found with session ID: persisted-stale-session",
],
num_turns: 0,
duration_ms: 0,
duration_api_ms: 0,
is_error: true,
stop_reason: null,
total_cost_usd: 0,
usage: buildUsage(),
},
};
}
await keepQueryAlive.promise;
return { done: true, value: undefined };
}),
interrupt: vi.fn(async () => undefined),
return: vi.fn(async () => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
} satisfies QueryMock;
});
const client = new ClaudeAgentClient({ logger });
const handle: AgentPersistenceHandle = {
provider: "claude",
sessionId: "persisted-stale-session",
nativeHandle: "persisted-stale-session",
metadata: {
provider: "claude",
cwd: process.cwd(),
},
};
const session = await client.resumeSession(handle, { cwd: process.cwd() });
await new Promise((resolve) => setTimeout(resolve, 25));
const activeRuns = (
session as unknown as {
runTracker: { listActiveRuns: (owner?: "foreground" | "autonomous") => unknown[] };
}
).runTracker.listActiveRuns();
expect(activeRuns).toHaveLength(0);
keepQueryAlive.resolve(undefined);
await session.close();
});
test("stops retrying live query pump when resumed Claude session no longer exists", async () => {
const logger = createTestLogger();
let queryCreateCount = 0;
sdkMocks.query.mockImplementation(() => {
queryCreateCount += 1;
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "error_during_execution",
session_id: "new-session-after-missing-conversation",
errors: [
"No conversation found with session ID: persisted-stale-session",
],
num_turns: 0,
duration_ms: 0,
duration_api_ms: 0,
is_error: true,
stop_reason: null,
total_cost_usd: 0,
usage: buildUsage(),
},
};
}
return { done: true, value: undefined };
}),
interrupt: vi.fn(async () => undefined),
return: vi.fn(async () => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
} satisfies QueryMock;
});
const client = new ClaudeAgentClient({ logger });
const handle: AgentPersistenceHandle = {
provider: "claude",
sessionId: "persisted-stale-session",
nativeHandle: "persisted-stale-session",
metadata: {
provider: "claude",
cwd: process.cwd(),
},
};
const session = await client.resumeSession(handle, { cwd: process.cwd() });
const liveIterator = (
session as unknown as {
streamLiveEvents: () => AsyncGenerator<AgentStreamEvent>;
}
).streamLiveEvents();
const liveNext = liveIterator.next();
await new Promise((resolve) => setTimeout(resolve, 25));
await new Promise((resolve) => setTimeout(resolve, 650));
expect(queryCreateCount).toBe(1);
expect(session.describePersistence()).toBeNull();
await session.close();
await liveNext;
});
test("does not emit live autonomous turn events for local_agent task_started during a foreground run", async () => {
const logger = createTestLogger();
const keepQueryAlive = deferred<void>();
@@ -1150,6 +1293,216 @@ describe("ClaudeAgentSession interrupt restart regression", () => {
await session.close();
});
test("does not let task_notification reservations steal a foreground terminal result", async () => {
const logger = createTestLogger();
const keepQueryAlive = deferred<void>();
sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
const readPromptUuid = createPromptUuidReader(prompt);
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "task-notification-foreground-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "task_notification",
task_id: "task-foreground-1",
tool_use_id: "toolu_foreground_1",
status: "completed",
output_file: "/tmp/task-foreground-1.txt",
summary: "Check Phase 1",
session_id: "task-notification-foreground-session",
uuid: "task-note-foreground-1",
},
};
}
if (step === 2) {
step += 1;
const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid";
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "verify prompt" },
parent_tool_use_id: null,
uuid: promptUuid,
session_id: "task-notification-foreground-session",
isReplay: true,
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "message_start",
message: {
id: "foreground-tool-msg",
role: "assistant",
model: "opus",
usage: { input_tokens: 1, output_tokens: 0 },
},
},
session_id: "task-notification-foreground-session",
parent_tool_use_id: null,
uuid: "foreground-message-start",
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "content_block_start",
index: 0,
content_block: {
type: "tool_use",
id: "toolu_foreground_2",
name: "Agent",
input: {},
caller: { type: "direct" },
},
},
session_id: "task-notification-foreground-session",
parent_tool_use_id: null,
uuid: "foreground-tool-use-start",
},
};
}
if (step === 5) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
id: "foreground-final-msg",
content: "FOREGROUND_RESULT_STAYS_ATTACHED",
},
session_id: "task-notification-foreground-session",
},
};
}
if (step === 6) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "message_delta",
delta: { stop_reason: "end_turn", stop_sequence: null },
usage: { input_tokens: 1, output_tokens: 1 },
},
session_id: "task-notification-foreground-session",
parent_tool_use_id: null,
uuid: "foreground-message-delta",
},
};
}
if (step === 7) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: { type: "message_stop" },
session_id: "task-notification-foreground-session",
parent_tool_use_id: null,
uuid: "foreground-message-stop",
},
};
}
if (step === 8) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
stop_reason: "end_turn",
session_id: "task-notification-foreground-session",
},
};
}
if (step === 9) {
await keepQueryAlive.promise;
return { done: true, value: undefined };
}
return { done: true, value: undefined };
}),
interrupt: vi.fn(async () => undefined),
return: vi.fn(async () => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
} satisfies QueryMock;
});
const client = new ClaudeAgentClient({ logger });
const session = await client.createSession({
provider: "claude",
cwd: process.cwd(),
});
const foregroundEvents = await collectUntilTerminal(session.stream("verify prompt"));
const liveIterator = (
session as unknown as {
streamLiveEvents: () => AsyncGenerator<AgentStreamEvent>;
}
).streamLiveEvents();
const timedReader = createTimedIteratorReader({ iterator: liveIterator });
const liveEvents: AgentStreamEvent[] = [];
for (let attempt = 0; attempt < 3; attempt += 1) {
try {
const next = await timedReader.nextWithTimeout(25);
if (next.done) {
break;
}
liveEvents.push(next.value);
} catch {
break;
}
}
expect(collectAssistantText(foregroundEvents)).toContain(
"FOREGROUND_RESULT_STAYS_ATTACHED"
);
expect(foregroundEvents.some((event) => event.type === "turn_completed")).toBe(true);
expect(liveEvents.some((event) => event.type === "turn_started")).toBe(false);
expect(liveEvents.some((event) => event.type === "turn_completed")).toBe(false);
keepQueryAlive.resolve(undefined);
await session.close();
});
test("emits autonomous live events from SDK stream when Claude wakes itself", async () => {
const logger = createTestLogger();
let queryCreateCount = 0;
@@ -74,6 +74,24 @@ import {
} from "../provider-launch-config.js";
import { getOrchestratorModeInstructions } from "../orchestrator-instructions.js";
/*
* Routing invariant:
* While a foreground Claude turn is active, identifier-less assistant/stream/result
* events must stay attached to that foreground run unless we have explicit evidence
* that a distinct autonomous run has started.
*
* We previously allowed task_notification metadata to reserve autonomous ownership,
* then consumed that reservation on the next unbound chunk. In practice Claude often
* emits task_notification records and foreground tool-use stream chunks interleaved
* within the same turn. That let a foreground turn's terminal result get misrouted
* into an autonomous side run, which stranded the agent in "running" because the
* foreground stream never received turn_completed.
*
* The rule below is intentionally conservative: foreground turns get first claim on
* same-turn traffic, and autonomous wake reservations are only consumed once no
* foreground turn is active. This keeps task_notification advisory instead of letting
* it steal ownership from the active user turn.
*/
const fsPromises = promises;
const CLAUDE_SETTING_SOURCES: NonNullable<Options["settingSources"]> = [
"user",
@@ -102,8 +120,9 @@ type RoutingReason =
| "parent_message_id"
| "message_id"
| "unbound_autonomous"
| "fallback"
| "metadata";
| "reserved_autonomous"
| "metadata"
| "ignored_unmatched";
type EventIdentifiers = {
taskId: string | null;
@@ -114,6 +133,7 @@ type EventIdentifiers = {
type RunRoute = {
run: RunRecord | null;
reason: RoutingReason;
dispatchWithoutRun?: boolean;
};
type RunRecord = {
@@ -2583,8 +2603,10 @@ class ClaudeAgentSession implements AgentSession {
): RunRoute {
if (normalized.metadataOnly) {
if (
normalized.message.type === "user" &&
isTaskNotificationUserContent(normalized.message.message?.content)
(normalized.message.type === "user" &&
isTaskNotificationUserContent(normalized.message.message?.content)) ||
(normalized.message.type === "system" &&
normalized.message.subtype === "task_notification")
) {
this.reserveAutonomousWake("task_notification");
}
@@ -2603,13 +2625,6 @@ class ClaudeAgentSession implements AgentSession {
return byIdentifiers;
}
if (this.turnState === "autonomous") {
const activeAutonomousRun = this.runTracker.getLatestActiveRun("autonomous");
if (activeAutonomousRun) {
return { run: activeAutonomousRun, reason: "unbound_autonomous" };
}
}
const foregroundRun = this.activeForegroundTurn
? this.runTracker.getRun(this.activeForegroundTurn.runId)
: null;
@@ -2648,32 +2663,62 @@ class ClaudeAgentSession implements AgentSession {
return { run: foregroundRun, reason: "foreground" };
}
if (
this.pendingAutonomousWakeReservations > 0 &&
!normalized.identifiers.taskId &&
!normalized.identifiers.parentMessageId &&
!normalized.identifiers.messageId
) {
const reservedAutonomousRun = this.claimOrCreateAutonomousRun(
"reservation_unbound"
);
return {
run: reservedAutonomousRun,
reason: "reserved_autonomous",
};
}
if (!hasIdentifiers) {
if (this.pendingAutonomousWakeReservations > 0) {
const reservedAutonomousRun = this.claimOrCreateAutonomousRun(
"reservation_unbound"
);
return {
run: reservedAutonomousRun,
reason: "unbound_autonomous",
};
}
const activeAutonomousRun = this.runTracker.getLatestActiveRun("autonomous");
if (activeAutonomousRun) {
return { run: activeAutonomousRun, reason: "unbound_autonomous" };
}
if (
!foregroundRun &&
(normalized.message.type === "assistant" ||
normalized.message.type === "stream_event" ||
normalized.message.type === "result" ||
normalized.message.type === "tool_progress")
) {
const autonomousRun = this.claimOrCreateAutonomousRun("unbound_implicit");
return { run: autonomousRun, reason: "unbound_autonomous" };
}
}
if (this.pendingAutonomousWakeReservations > 0) {
const reservedAutonomousRun = this.claimOrCreateAutonomousRun(
"reservation_fallback"
);
return { run: reservedAutonomousRun, reason: "fallback" };
return { run: reservedAutonomousRun, reason: "reserved_autonomous" };
}
const autonomousRun = this.createRun("autonomous", null);
this.emitRunEvent(autonomousRun, { type: "turn_started", provider: "claude" });
return { run: autonomousRun, reason: "fallback" };
this.logger.debug(
{
messageType: normalized.message.type,
hasIdentifiers,
taskId: normalized.identifiers.taskId,
parentMessageId: normalized.identifiers.parentMessageId,
messageId: normalized.identifiers.messageId,
turnState: this.turnState,
pendingAutonomousWakeReservations: this.pendingAutonomousWakeReservations,
},
"Ignoring unmatched Claude SDK message without explicit run start signal"
);
return {
run: null,
reason: "ignored_unmatched",
dispatchWithoutRun: false,
};
}
private shouldPreferForegroundRun(input: {
@@ -2746,10 +2791,14 @@ class ClaudeAgentSession implements AgentSession {
if (!foregroundRun || foregroundRun.promptReplaySeen) {
return;
}
// System metadata (init/hook callbacks/etc.) can precede the first prompt
// replay for a legitimate foreground run. Treating that as churn strands
// one-shot helper runs in autonomous fallback.
if (message.type === "system") {
// Most system metadata (init/hook callbacks/etc.) can precede the first prompt
// replay for a legitimate foreground run. Treating all of it as churn strands
// one-shot helper runs. task_notification is the exception: it represents
// background agent activity and should suppress pre-replay foreground routing.
if (
message.type === "system" &&
message.subtype !== "task_notification"
) {
return;
}
this.preReplayMetadataSeen = true;
@@ -2886,6 +2935,10 @@ class ClaudeAgentSession implements AgentSession {
continue;
}
if (await this.handleMissingResumedConversation(sdkMessage, q)) {
continue;
}
try {
this.routeSdkMessageFromPump(sdkMessage);
} catch (error) {
@@ -2954,6 +3007,9 @@ class ClaudeAgentSession implements AgentSession {
}
if (!route.run) {
if (route.dispatchWithoutRun === false) {
return;
}
this.dispatchMetadataEvents(events);
return;
}
@@ -2963,6 +3019,45 @@ class ClaudeAgentSession implements AgentSession {
}
}
private async handleMissingResumedConversation(
message: SDKMessage,
query: Query
): Promise<boolean> {
const staleResumeError = this.readMissingResumedConversationError(message);
if (!staleResumeError) {
return false;
}
this.logger.warn(
{
claudeSessionId: this.claudeSessionId,
error: staleResumeError,
},
"Claude resumed session no longer exists; invalidating persisted session"
);
for (const run of this.runTracker.listActiveRuns()) {
this.failRun(run, staleResumeError);
}
this.transitionTurnStateFromActiveRuns("missing resumed conversation");
this.input?.end();
await this.awaitWithTimeout(
query.return?.(),
"query pump return on missing resumed conversation"
);
if (this.query === query) {
this.query = null;
this.input = null;
}
this.claudeSessionId = null;
this.persistence = null;
this.persistedHistory = [];
this.historyPending = false;
this.cachedRuntimeInfo = null;
this.queryRestartNeeded = false;
return true;
}
private shouldSuppressReplayResultTerminal(input: {
run: RunRecord | null;
message: SDKMessage;
@@ -3691,6 +3786,30 @@ class ClaudeAgentSession implements AgentSession {
return threadStartedSessionId;
}
private readMissingResumedConversationError(message: SDKMessage): string | null {
if (message.type !== "result" || message.subtype !== "error_during_execution") {
return null;
}
if (!this.claudeSessionId) {
return null;
}
const errors =
"errors" in message && Array.isArray(message.errors) ? message.errors : [];
for (const entry of errors) {
if (typeof entry !== "string") {
continue;
}
const match = entry.match(/^No conversation found with session ID:\s*(.+)$/);
if (!match) {
continue;
}
if (match[1]?.trim() === this.claudeSessionId) {
return entry.trim();
}
}
return null;
}
private convertUsage(message: SDKResultMessage): AgentUsage | undefined {
if (!message.usage) {
return undefined;
@@ -0,0 +1,116 @@
import { describe, expect, test } from "vitest";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import pino from "pino";
import { ClaudeAgentClient } from "../agent/providers/claude-agent.js";
import { isCommandAvailable } from "../agent/provider-launch-config.js";
import { DaemonClient } from "../test-utils/daemon-client.js";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { getFullAccessConfig } from "./agent-configs.js";
function tmpCwd(): string {
return mkdtempSync(path.join(tmpdir(), "daemon-real-claude-autonomous-simple-"));
}
function compactText(value: string): string {
return value.replace(/\s+/g, "").toLowerCase();
}
describe("daemon E2E (real claude) - autonomous wake simple", () => {
test.runIf(isCommandAvailable("claude"))(
"hello + background sleep returns idle, then wakes once on completion",
async () => {
const logger = pino({ level: "silent" });
const cwd = tmpCwd();
const daemon = await createTestPaseoDaemon({
agentClients: { claude: new ClaudeAgentClient({ logger }) },
logger,
});
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
try {
await client.connect();
await client.fetchAgents({
subscribe: { subscriptionId: "claude-autonomous-simple-real" },
});
const agent = await client.createAgent({
cwd,
title: "claude-autonomous-simple-real",
...getFullAccessConfig("claude"),
});
const autonomousWakeToken = `AUTONOMOUS_SIMPLE_${Date.now().toString(36)}`;
await client.sendMessage(
agent.id,
[
"Hello.",
"Use the Bash tool with run_in_background.",
"Run exactly: sleep 5",
"Do not wait for the task result.",
"Reply immediately with exactly: SPAWNED",
`When the background task completes later, reply with exactly: ${autonomousWakeToken}`,
].join(" ")
);
const firstFinish = await client.waitForFinish(agent.id, 240_000);
expect(firstFinish.status).toBe("idle");
const timelineAtIdle = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const idleAssistantText = timelineAtIdle.entries
.filter(
(entry): entry is typeof entry & {
item: { type: "assistant_message"; text: string };
} => entry.item.type === "assistant_message"
)
.map((entry) => entry.item.text)
.join("\n");
expect(compactText(idleAssistantText)).toContain("spawned");
expect(
timelineAtIdle.entries.some(
(entry) =>
entry.item.type === "tool_call" &&
entry.item.name === "Bash"
)
).toBe(true);
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
const autonomousFinish = await client.waitForFinish(agent.id, 120_000);
expect(autonomousFinish.status).toBe("idle");
const finalTimeline = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const finalAssistantText = finalTimeline.entries
.filter(
(entry): entry is typeof entry & {
item: { type: "assistant_message"; text: string };
} => entry.item.type === "assistant_message"
)
.map((entry) => entry.item.text)
.join("\n");
expect(compactText(finalAssistantText)).toContain(
autonomousWakeToken.toLowerCase()
);
} finally {
await client.close().catch(() => undefined);
await daemon.close().catch(() => undefined);
rmSync(cwd, { recursive: true, force: true });
}
},
420_000
);
});