feat: add thinking option support and autonomous run handling

This commit is contained in:
Mohamed Boudra
2026-02-21 22:09:48 +07:00
parent c3c8103563
commit d7e2ab4437
35 changed files with 4701 additions and 593 deletions
@@ -150,11 +150,9 @@ function SessionProviderWithClient({
client,
}: SessionProviderClientProps) {
return (
<SessionProviderInternal
children={children}
serverId={serverId}
client={client}
/>
<SessionProviderInternal serverId={serverId} client={client}>
{children}
</SessionProviderInternal>
);
}
@@ -378,13 +378,7 @@ function DraftAgentScreenContent({
const sessionClient = runtimeClient
const trimmedWorkingDir = workingDir.trim()
const shouldInspectRepo = trimmedWorkingDir.length > 0
const daemonAvailabilityError =
!selectedServerId || !isHostOnline ? 'Host is offline' : null
const repoAvailabilityError =
shouldInspectRepo && !isHostOnline
? (daemonAvailabilityError ??
'Repository details will load automatically once the selected host is back online.')
: null
const canQuerySelectedHost = Boolean(selectedServerId) && Boolean(sessionClient) && isHostOnline
const checkoutStatusQuery = useQuery({
queryKey: checkoutStatusQueryKey(selectedServerId ?? '', trimmedWorkingDir),
@@ -395,12 +389,7 @@ function DraftAgentScreenContent({
}
return await client.getCheckoutStatus(trimmedWorkingDir)
},
enabled:
Boolean(selectedServerId) &&
Boolean(trimmedWorkingDir) &&
!repoAvailabilityError &&
Boolean(sessionClient) &&
isHostOnline,
enabled: Boolean(trimmedWorkingDir) && canQuerySelectedHost,
retry: false,
staleTime: CHECKOUT_STATUS_STALE_TIME,
refetchOnMount: 'always',
@@ -424,8 +413,8 @@ function DraftAgentScreenContent({
const repoInfoStatus: 'idle' | 'loading' | 'ready' | 'error' = !shouldInspectRepo
? 'idle'
: repoAvailabilityError
? 'error'
: !canQuerySelectedHost
? 'idle'
: checkoutStatusQuery.isPending || checkoutStatusQuery.isFetching
? 'loading'
: checkoutStatusQuery.isError || Boolean(checkoutPayloadError)
@@ -435,9 +424,7 @@ function DraftAgentScreenContent({
: 'idle'
const repoInfoError =
repoAvailabilityError ??
(checkoutStatusQuery.isError ? checkoutQueryError : null) ??
checkoutPayloadError
(checkoutStatusQuery.isError ? checkoutQueryError : null) ?? checkoutPayloadError
const isCreateWorktree = worktreeMode === 'create'
const isAttachWorktree = worktreeMode === 'attach'
@@ -461,9 +448,7 @@ function DraftAgentScreenContent({
enabled:
isGitDirectory &&
Boolean(worktreeListRoot) &&
!repoAvailabilityError &&
Boolean(sessionClient) &&
isHostOnline &&
canQuerySelectedHost &&
!isNonGitDirectory,
retry: false,
staleTime: 0,
@@ -526,9 +511,7 @@ function DraftAgentScreenContent({
isGitDirectory &&
!isNonGitDirectory &&
Boolean(trimmedWorkingDir) &&
!repoAvailabilityError &&
Boolean(sessionClient) &&
isHostOnline,
canQuerySelectedHost,
retry: false,
staleTime: 15_000,
})
@@ -558,10 +541,7 @@ function DraftAgentScreenContent({
},
enabled:
Boolean(debouncedWorkingDirSearchQuery) &&
Boolean(selectedServerId) &&
!daemonAvailabilityError &&
Boolean(sessionClient) &&
isHostOnline,
canQuerySelectedHost,
retry: false,
staleTime: 15_000,
})
+12 -16
View File
@@ -9,6 +9,7 @@ import { createWorktreeCommand } from './commands/worktree/index.js'
import { startCommand as daemonStartCommand } from './commands/daemon/start.js'
import { runStatusCommand as runDaemonStatusCommand } from './commands/daemon/status.js'
import { runRestartCommand as runDaemonRestartCommand } from './commands/daemon/restart.js'
import { runDaemonUpdateCommandOrExit } from './commands/daemon/update.js'
import { runLsCommand } from './commands/agent/ls.js'
import { runRunCommand } from './commands/agent/run.js'
import { runLogsCommand } from './commands/agent/logs.js'
@@ -17,7 +18,6 @@ import { runSendCommand } from './commands/agent/send.js'
import { runInspectCommand } from './commands/agent/inspect.js'
import { runWaitCommand } from './commands/agent/wait.js'
import { runAttachCommand } from './commands/agent/attach.js'
import { runUpdateCommand } from './commands/agent/update.js'
import { withOutput } from './output/index.js'
import { onboardCommand } from './commands/onboard.js'
@@ -63,6 +63,7 @@ export function createCli(): Command {
.option('-a, --all', 'Include archived agents')
.option('-g, --global', 'Legacy no-op (kept for compatibility)')
.option('--label <key=value>', 'Filter by label (can be used multiple times)', collectMultiple, [])
.option('--thinking <id>', 'Filter by thinking option ID')
.option('--json', 'Output in JSON format')
.option('--host <host>', 'Daemon host:port (default: localhost:6767)')
.action(withOutput(runLsCommand))
@@ -75,6 +76,7 @@ export function createCli(): Command {
.option('--name <name>', 'Assign a name/title to the agent')
.option('--provider <provider>', 'Agent provider: claude | codex | opencode', 'claude')
.option('--model <model>', 'Model to use (e.g., claude-sonnet-4-20250514, claude-3-5-haiku-20241022)')
.option('--thinking <id>', 'Thinking option ID to use for this run')
.option('--mode <mode>', 'Provider-specific mode (e.g., plan, default, bypass)')
.option('--worktree <name>', 'Create agent in a new git worktree')
.option('--base <branch>', 'Base branch for worktree (default: current branch)')
@@ -143,25 +145,19 @@ export function createCli(): Command {
.option('--host <host>', 'Daemon host:port (default: localhost:6767)')
.action(withOutput(runWaitCommand))
program
.command('update')
.description('Update an agent (alias for "paseo agent update")')
.argument('<id>', 'Agent ID (or prefix)')
.option('--name <name>', "Update the agent's display name")
.option(
'--label <label>',
'Add/set label(s) on the agent (can be used multiple times or comma-separated)',
collectMultiple,
[]
)
.option('--json', 'Output in JSON format')
.option('--host <host>', 'Daemon host:port (default: localhost:6767)')
.action(withOutput(runUpdateCommand))
// Top-level local daemon shortcuts
program.addCommand(onboardCommand())
program.addCommand(daemonStartCommand())
program
.command('update')
.description('Update local daemon package (alias for "paseo daemon update")')
.option('--home <path>', 'Paseo home directory (default: ~/.paseo)')
.option('-y, --yes', 'Restart automatically after update')
.action(async (options: { home?: string; yes?: boolean }) => {
await runDaemonUpdateCommandOrExit(options)
})
program
.command('status')
.description('Show local daemon status (alias for "paseo daemon status")')
+2
View File
@@ -27,6 +27,7 @@ export function createAgentCommand(): Command {
.option('-a, --all', 'Include archived agents')
.option('-g, --global', 'Legacy no-op (kept for compatibility)')
.option('--label <key=value>', 'Filter by label (can be used multiple times)', collectMultiple, [])
.option('--thinking <id>', 'Filter by thinking option ID')
.option('--json', 'Output in JSON format')
.option('--host <host>', 'Daemon host:port (default: localhost:6767)')
.action(withOutput(runLsCommand))
@@ -39,6 +40,7 @@ export function createAgentCommand(): Command {
.option('--name <name>', 'Assign a name/title to the agent')
.option('--provider <provider>', 'Agent provider: claude | codex | opencode', 'claude')
.option('--model <model>', 'Model to use (e.g., claude-sonnet-4-20250514, claude-3-5-haiku-20241022)')
.option('--thinking <id>', 'Thinking option ID to use for this run')
.option('--mode <mode>', 'Provider-specific mode (e.g., plan, default, bypass)')
.option('--cwd <path>', 'Working directory (default: current)')
.option('--label <key=value>', 'Add label(s) to the agent (can be used multiple times)', collectMultiple, [])
+15 -1
View File
@@ -9,6 +9,7 @@ interface AgentInspect {
Name: string
Provider: string
Model: string
Thinking: string
Status: string
Mode: string
Cwd: string
@@ -84,6 +85,17 @@ function formatCost(costUsd: number): string {
return `$${costUsd.toFixed(2)}`
}
function normalizeModelId(value: string | null | undefined): string | null {
if (typeof value !== 'string') return null
const normalized = value.trim()
if (!normalized || normalized.toLowerCase() === 'default') return null
return normalized
}
function resolveModel(snapshot: AgentSnapshotPayload): string | null {
return normalizeModelId(snapshot.runtimeInfo?.model) ?? normalizeModelId(snapshot.model)
}
/** Convert agent snapshot to inspection data */
function toInspectData(snapshot: AgentSnapshotPayload): AgentInspect {
const lastUsage = snapshot.lastUsage
@@ -112,7 +124,8 @@ function toInspectData(snapshot: AgentSnapshotPayload): AgentInspect {
Id: snapshot.id,
Name: snapshot.title ?? '-',
Provider: snapshot.provider,
Model: snapshot.model ?? '-',
Model: resolveModel(snapshot) ?? '-',
Thinking: snapshot.effectiveThinkingOptionId ?? 'auto',
Status: snapshot.status,
Mode: snapshot.currentModeId ?? 'default',
Cwd: snapshot.cwd,
@@ -139,6 +152,7 @@ function toInspectRows(agent: AgentInspect): InspectRow[] {
{ key: 'Name', value: agent.Name },
{ key: 'Provider', value: agent.Provider },
{ key: 'Model', value: agent.Model },
{ key: 'Thinking', value: agent.Thinking },
{ key: 'Status', value: agent.Status },
{ key: 'Mode', value: agent.Mode },
{ key: 'Cwd', value: shortenPath(agent.Cwd) },
+52 -15
View File
@@ -9,6 +9,7 @@ export interface AgentListItem {
shortId: string
name: string
provider: string
thinking: string
status: string
cwd: string
created: string
@@ -35,6 +36,13 @@ function shortenPath(path: string): string {
return path
}
function normalizeModelId(modelId: string | null | undefined): string | null {
if (typeof modelId !== 'string') return null
const normalized = modelId.trim()
if (!normalized || normalized.toLowerCase() === 'default') return null
return normalized
}
/** Schema for agent ls output */
export const agentLsSchema: OutputSchema<AgentListItem> = {
idField: 'shortId',
@@ -42,6 +50,7 @@ export const agentLsSchema: OutputSchema<AgentListItem> = {
{ header: 'AGENT ID', field: 'shortId', width: 12 },
{ header: 'NAME', field: 'name', width: 20 },
{ header: 'PROVIDER', field: 'provider', width: 15 },
{ header: 'THINKING', field: 'thinking', width: 12 },
{
header: 'STATUS',
field: 'status',
@@ -60,11 +69,13 @@ export const agentLsSchema: OutputSchema<AgentListItem> = {
/** Transform agent snapshot to AgentListItem */
function toListItem(agent: AgentSnapshotPayload): AgentListItem {
const model = normalizeModelId(agent.runtimeInfo?.model) ?? normalizeModelId(agent.model)
return {
id: agent.id,
shortId: agent.id.slice(0, 7),
name: agent.title ?? '-',
provider: agent.model ? `${agent.provider}/${agent.model}` : agent.provider,
provider: model ? `${agent.provider}/${model}` : agent.provider,
thinking: agent.effectiveThinkingOptionId ?? 'auto',
status: agent.status,
cwd: shortenPath(agent.cwd),
created: relativeTime(agent.createdAt),
@@ -84,6 +95,8 @@ export interface AgentLsOptions extends CommandOptions {
cwd?: string
/** Filter by labels (key=value format) */
label?: string[]
/** Filter by thinking option ID */
thinking?: string
}
/**
@@ -111,8 +124,45 @@ export async function runLsCommand(
}
try {
const normalizedThinkingOptionId = options.thinking?.trim()
if (options.thinking !== undefined && !normalizedThinkingOptionId) {
const error: CommandError = {
code: 'INVALID_THINKING_OPTION',
message: '--thinking cannot be empty',
}
throw error
}
// Parse --label filters (key=value).
const labelFilters: Record<string, string> = {}
if (options.label) {
for (const labelStr of options.label) {
const eqIndex = labelStr.indexOf('=')
if (eqIndex !== -1) {
const key = labelStr.slice(0, eqIndex)
const value = labelStr.slice(eqIndex + 1)
labelFilters[key] = value
}
}
}
const daemonFilter: {
includeArchived?: boolean
labels?: Record<string, string>
thinkingOptionId?: string
} = {}
if (options.all) {
daemonFilter.includeArchived = true
}
if (Object.keys(labelFilters).length > 0) {
daemonFilter.labels = labelFilters
}
if (normalizedThinkingOptionId) {
daemonFilter.thinkingOptionId = normalizedThinkingOptionId
}
const fetchPayload = await client.fetchAgents({
filter: options.all ? { includeArchived: true } : undefined,
filter: Object.keys(daemonFilter).length > 0 ? daemonFilter : undefined,
})
let agents = fetchPayload.entries.map((entry) => entry.agent)
@@ -135,19 +185,6 @@ export async function runLsCommand(
})
}
// Parse --label filters (key=value).
const labelFilters: Record<string, string> = {}
if (options.label) {
for (const labelStr of options.label) {
const eqIndex = labelStr.indexOf('=')
if (eqIndex !== -1) {
const key = labelStr.slice(0, eqIndex)
const value = labelStr.slice(eqIndex + 1)
labelFilters[key] = value
}
}
}
// Apply label filtering only when explicitly requested.
if (Object.keys(labelFilters).length > 0) {
agents = agents.filter((a) => {
+13
View File
@@ -36,6 +36,7 @@ export interface AgentRunOptions extends CommandOptions {
name?: string
provider?: string
model?: string
thinking?: string
mode?: string
worktree?: string
base?: string
@@ -219,6 +220,16 @@ export async function runRunCommand(
try {
// Resolve working directory
const cwd = options.cwd ?? process.cwd()
const thinkingOptionId = options.thinking?.trim()
if (options.thinking !== undefined && !thinkingOptionId) {
const error: CommandError = {
code: 'INVALID_THINKING_OPTION',
message: '--thinking cannot be empty',
details:
'Provide a thinking option ID. Use "paseo provider models <provider> --thinking" to list valid IDs.',
}
throw error
}
// Process images if provided
let images: Array<{ data: string; mimeType: string }> | undefined
@@ -290,6 +301,7 @@ export async function runRunCommand(
title: options.name,
modeId: options.mode,
model: options.model,
thinkingOptionId,
initialPrompt: structuredPrompt,
images,
git,
@@ -387,6 +399,7 @@ export async function runRunCommand(
title: options.name,
modeId: options.mode,
model: options.model,
thinkingOptionId,
initialPrompt: prompt,
images,
git,
@@ -4,6 +4,7 @@ import { runStatusCommand } from './status.js'
import { runStopCommand } from './stop.js'
import { runRestartCommand } from './restart.js'
import { pairCommand } from './pair.js'
import { updateCommand } from './update.js'
import { withOutput } from '../../output/index.js'
export function createDaemonCommand(): Command {
@@ -11,6 +12,7 @@ export function createDaemonCommand(): Command {
daemon.addCommand(startCommand())
daemon.addCommand(pairCommand())
daemon.addCommand(updateCommand())
daemon
.command('status')
@@ -0,0 +1,131 @@
import { spawnSync } from 'node:child_process'
import { existsSync } from 'node:fs'
import path from 'node:path'
export type NodePathSource = 'daemon_pid' | 'current_process'
export interface NodePathResolution {
nodePath: string
source: NodePathSource
note?: string
}
export interface NodePathFromPidResult {
nodePath: string | null
error?: string
}
export interface NpmInvocation {
nodePath: string
npmPath: string
command: string
argsPrefix: string[]
}
function normalizeError(error: unknown): string {
if (error instanceof Error) {
return error.message
}
return String(error)
}
export function resolveNodePathFromPid(pid: number): NodePathFromPidResult {
const result = spawnSync('ps', ['-o', 'comm=', '-p', String(pid)], {
encoding: 'utf8',
stdio: ['ignore', 'pipe', 'pipe'],
})
if (result.error) {
return {
nodePath: null,
error: `ps failed: ${normalizeError(result.error)}`,
}
}
if ((result.status ?? 1) !== 0) {
const details = result.stderr?.trim()
return {
nodePath: null,
error: details ? `ps failed: ${details}` : `ps exited with code ${result.status ?? 1}`,
}
}
const resolved = result.stdout.trim()
if (!resolved) {
return {
nodePath: null,
error: 'ps returned an empty command path',
}
}
return { nodePath: resolved }
}
export function resolvePreferredNodePath(args: {
daemonPid?: number | null
fallbackNodePath?: string
}): NodePathResolution {
const fallback = args.fallbackNodePath ?? process.execPath
const daemonPid = args.daemonPid
if (typeof daemonPid === 'number' && Number.isInteger(daemonPid) && daemonPid > 0) {
const fromPid = resolveNodePathFromPid(daemonPid)
if (fromPid.nodePath) {
return {
nodePath: fromPid.nodePath,
source: 'daemon_pid',
}
}
return {
nodePath: fallback,
source: 'current_process',
note: `Could not resolve node from daemon PID ${daemonPid}; using current process node (${fromPid.error ?? 'unknown error'})`,
}
}
return {
nodePath: fallback,
source: 'current_process',
}
}
export function resolveNpmInvocationFromNode(nodePath: string): NpmInvocation {
const binDir = path.dirname(nodePath)
const prefix = path.dirname(binDir)
const npmBinary = path.join(binDir, process.platform === 'win32' ? 'npm.cmd' : 'npm')
if (existsSync(npmBinary)) {
return {
nodePath,
npmPath: npmBinary,
command: npmBinary,
argsPrefix: [],
}
}
const npmCliCandidates = [
path.join(prefix, 'lib', 'node_modules', 'npm', 'bin', 'npm-cli.js'),
path.join(prefix, 'node_modules', 'npm', 'bin', 'npm-cli.js'),
]
for (const candidate of npmCliCandidates) {
if (existsSync(candidate)) {
return {
nodePath,
npmPath: candidate,
command: nodePath,
argsPrefix: [candidate],
}
}
}
throw new Error(`Unable to resolve npm for node executable: ${nodePath}`)
}
export function formatNpmInvocation(invocation: NpmInvocation): string {
if (invocation.argsPrefix.length === 0) {
return invocation.command
}
return `${invocation.command} ${invocation.argsPrefix.join(' ')}`
}
+201 -3
View File
@@ -1,7 +1,15 @@
import type { Command } from 'commander'
import { spawnSync } from 'node:child_process'
import { createRequire } from 'node:module'
import { tryConnectToDaemon } from '../../utils/client.js'
import type { CommandOptions, ListResult, OutputSchema } from '../../output/index.js'
import { resolveLocalDaemonState, resolveTcpHostFromListen } from './local-daemon.js'
import {
formatNpmInvocation,
resolveNpmInvocationFromNode,
resolvePreferredNodePath,
type NpmInvocation,
} from './runtime-toolchain.js'
interface DaemonStatus {
status: 'running' | 'stopped' | 'unresponsive'
@@ -13,6 +21,11 @@ interface DaemonStatus {
logPath: string
runningAgents: number | null
idleAgents: number | null
runtimeNode: string
runtimeNpm: string
cliVersion: string
latestCliVersion: string
updateStatus: string
note?: string
}
@@ -21,6 +34,140 @@ interface StatusRow {
value: string
}
type CliPackageJson = {
version?: unknown
}
type LatestCliVersionResult = {
version: string | null
note?: string
}
const require = createRequire(import.meta.url)
const CLI_UPDATE_CHECK_TIMEOUT_MS = 3000
function normalizeError(error: unknown): string {
if (error instanceof Error) {
return error.message
}
return String(error)
}
function shortenMessage(message: string, max = 120): string {
const normalized = message.replace(/\s+/g, ' ').trim()
if (normalized.length <= max) {
return normalized
}
return `${normalized.slice(0, max - 3)}...`
}
function appendNote(current: string | undefined, next: string | undefined): string | undefined {
if (!next) return current
if (!current) return next
return `${current}; ${next}`
}
function resolveCliVersion(): string {
try {
const packageJson = require('../../../package.json') as CliPackageJson
if (typeof packageJson.version === 'string' && packageJson.version.trim().length > 0) {
return packageJson.version.trim()
}
} catch {
// Fall through.
}
return 'unknown'
}
function parseVersionFromNpmOutput(raw: string): string | null {
const trimmed = raw.trim()
if (!trimmed) {
return null
}
try {
const parsed = JSON.parse(trimmed) as unknown
if (typeof parsed === 'string' && parsed.trim().length > 0) {
return parsed.trim()
}
} catch {
// Fall back to plain output.
}
return trimmed.replace(/^"+|"+$/g, '').trim() || null
}
function parseSemver(version: string): [number, number, number] | null {
const match = version.trim().match(/^v?(\d+)\.(\d+)\.(\d+)(?:[-+].*)?$/)
if (!match) {
return null
}
const major = Number(match[1])
const minor = Number(match[2])
const patch = Number(match[3])
if (![major, minor, patch].every((part) => Number.isInteger(part) && part >= 0)) {
return null
}
return [major, minor, patch]
}
function compareSemver(left: string, right: string): number | null {
const leftParts = parseSemver(left)
const rightParts = parseSemver(right)
if (!leftParts || !rightParts) {
return null
}
for (let index = 0; index < leftParts.length; index += 1) {
if (leftParts[index]! < rightParts[index]!) return -1
if (leftParts[index]! > rightParts[index]!) return 1
}
return 0
}
function fetchLatestCliVersion(npm: NpmInvocation): LatestCliVersionResult {
const result = spawnSync(
npm.command,
[...npm.argsPrefix, 'view', '@getpaseo/cli', 'version', '--json'],
{
encoding: 'utf8',
stdio: ['ignore', 'pipe', 'pipe'],
env: process.env,
timeout: CLI_UPDATE_CHECK_TIMEOUT_MS,
}
)
if (result.error) {
return {
version: null,
note: `update check failed: ${shortenMessage(normalizeError(result.error))}`,
}
}
if ((result.status ?? 1) !== 0) {
const stderr = result.stderr?.trim()
return {
version: null,
note: stderr
? `update check failed: ${shortenMessage(stderr)}`
: `update check failed: npm exited with code ${result.status ?? 1}`,
}
}
const version = parseVersionFromNpmOutput(result.stdout)
if (!version) {
return {
version: null,
note: 'update check failed: empty npm response',
}
}
return { version }
}
function createStatusSchema(status: DaemonStatus): OutputSchema<StatusRow> {
return {
idField: 'key',
@@ -56,6 +203,11 @@ function toStatusRows(status: DaemonStatus): StatusRow[] {
{ key: 'Started', value: status.startedAt ?? '-' },
{ key: 'Owner', value: status.owner ?? '-' },
{ key: 'Logs', value: status.logPath },
{ key: 'Node', value: status.runtimeNode },
{ key: 'npm', value: status.runtimeNpm },
{ key: 'CLI', value: status.cliVersion },
{ key: 'Latest CLI', value: status.latestCliVersion },
{ key: 'Update', value: status.updateStatus },
]
if (status.runningAgents !== null && status.idleAgents !== null) {
@@ -96,14 +248,22 @@ export async function runStatusCommand(
const state = resolveLocalDaemonState({ home })
const owner = resolveOwnerLabel(state.pidInfo?.uid, state.pidInfo?.hostname)
const resolvedNode = resolvePreferredNodePath({
daemonPid: state.running ? state.pidInfo?.pid : null,
fallbackNodePath: process.execPath,
})
let status: DaemonStatus['status'] = state.running ? 'running' : 'stopped'
let runningAgents: number | null = null
let idleAgents: number | null = null
let note: string | undefined
let runtimeNpm = '-'
let latestCliVersion = 'unknown'
let updateStatus = 'unknown'
if (!state.running && state.stalePidFile && state.pidInfo) {
note = `Stale PID file found for PID ${state.pidInfo.pid}`
}
note = appendNote(note, resolvedNode.note)
if (state.running) {
const host = resolveTcpHostFromListen(state.listen)
@@ -117,19 +277,52 @@ export async function runStatusCommand(
idleAgents = agents.filter(a => a.status === 'idle').length
} catch {
status = 'unresponsive'
note = `Daemon PID is running but API requests to ${host} failed`
note = appendNote(note, `Daemon PID is running but API requests to ${host} failed`)
} finally {
await client.close().catch(() => {})
}
} else {
status = 'unresponsive'
note = `Daemon PID is running but websocket at ${host} is not reachable`
note = appendNote(note, `Daemon PID is running but websocket at ${host} is not reachable`)
}
} else {
note = 'Daemon is configured for unix socket listen; API probe skipped'
note = appendNote(note, 'Daemon is configured for unix socket listen; API probe skipped')
}
}
let npmInvocation: NpmInvocation | null = null
try {
npmInvocation = resolveNpmInvocationFromNode(resolvedNode.nodePath)
runtimeNpm = formatNpmInvocation(npmInvocation)
} catch (err) {
runtimeNpm = `unresolved (${shortenMessage(normalizeError(err))})`
}
const cliVersion = resolveCliVersion()
if (npmInvocation) {
const latest = fetchLatestCliVersion(npmInvocation)
if (latest.version) {
latestCliVersion = latest.version
if (cliVersion === 'unknown') {
updateStatus = 'unknown (local CLI version unavailable)'
} else {
const comparison = compareSemver(cliVersion, latest.version)
if (comparison === null) {
updateStatus = 'unknown (version format not comparable)'
} else if (comparison < 0) {
updateStatus = `update available (${cliVersion} -> ${latest.version})`
} else {
updateStatus = 'up to date'
}
}
} else {
latestCliVersion = 'unknown'
updateStatus = latest.note ?? 'unknown'
}
} else {
updateStatus = 'unknown (npm unresolved)'
}
const daemonStatus: DaemonStatus = {
status,
home: state.home,
@@ -140,6 +333,11 @@ export async function runStatusCommand(
logPath: state.logPath,
runningAgents,
idleAgents,
runtimeNode: `${resolvedNode.nodePath} (${resolvedNode.source})`,
runtimeNpm,
cliVersion,
latestCliVersion,
updateStatus,
note,
}
+105
View File
@@ -0,0 +1,105 @@
import { confirm, isCancel } from '@clack/prompts'
import { spawn } from 'node:child_process'
import { Command } from 'commander'
import chalk from 'chalk'
import { runRestartCommand } from './restart.js'
import { resolveLocalDaemonState } from './local-daemon.js'
import {
resolveNpmInvocationFromNode,
resolvePreferredNodePath,
type NpmInvocation,
} from './runtime-toolchain.js'
import { getErrorMessage } from '../../utils/errors.js'
export interface DaemonUpdateOptions {
home?: string
yes?: boolean
}
function runCommand(command: string, args: string[]): Promise<number> {
return new Promise((resolve, reject) => {
const child = spawn(command, args, {
stdio: 'inherit',
env: process.env,
})
child.once('error', reject)
child.once('exit', (code, signal) => {
if (signal) {
reject(new Error(`Command exited via signal ${signal}`))
return
}
resolve(code ?? 1)
})
})
}
export async function runDaemonUpdateCommand(options: DaemonUpdateOptions): Promise<void> {
const daemonState = resolveLocalDaemonState({ home: options.home })
const resolvedNode = resolvePreferredNodePath({
daemonPid: daemonState.running ? daemonState.pidInfo?.pid : null,
fallbackNodePath: process.execPath,
})
const npm: NpmInvocation = resolveNpmInvocationFromNode(resolvedNode.nodePath)
const args = [...npm.argsPrefix, 'install', '-g', '@getpaseo/cli@latest']
const exitCode = await runCommand(npm.command, args)
if (exitCode !== 0) {
throw new Error(`Update command failed with exit code ${exitCode}`)
}
let shouldRestart = options.yes === true
if (!shouldRestart) {
if (!process.stdin.isTTY || !process.stdout.isTTY) {
console.log(chalk.yellow('Update complete. Restart skipped (non-interactive terminal).'))
console.log(chalk.dim('Run `paseo daemon restart` when ready.'))
return
}
console.log(
chalk.yellow(
'Restart will stop running agents, but they can continue from persisted state after restart.'
)
)
const answer = await confirm({
message: 'Restart daemon now?',
active: 'Restart now',
inactive: 'Later',
initialValue: true,
})
if (isCancel(answer)) {
console.log(chalk.yellow('Update complete. Restart skipped.'))
return
}
shouldRestart = answer
}
if (!shouldRestart) {
console.log(chalk.yellow('Update complete. Restart skipped.'))
return
}
const restartResult = await runRestartCommand({ home: options.home }, new Command())
console.log(chalk.green(restartResult.data.message))
}
export async function runDaemonUpdateCommandOrExit(options: DaemonUpdateOptions): Promise<void> {
try {
await runDaemonUpdateCommand(options)
} catch (err) {
console.error(chalk.red(`Failed to update daemon: ${getErrorMessage(err)}`))
process.exit(1)
}
}
export function updateCommand(): Command {
return new Command('update')
.description('Update local daemon package')
.option('--home <path>', 'Paseo home directory (default: ~/.paseo)')
.option('-y, --yes', 'Restart automatically after update')
.action(async (options: DaemonUpdateOptions) => {
await runDaemonUpdateCommandOrExit(options)
})
}
@@ -17,6 +17,7 @@ export function createProviderCommand(): Command {
.command('models')
.description('List models for a provider')
.argument('<provider>', 'Provider name (claude, codex, opencode)')
.option('--thinking', 'Include thinking option IDs for each model')
.option('--json', 'Output in JSON format')
.option('--host <host>', 'Daemon host:port (default: localhost:6767)')
.action(withOutput(runModelsCommand))
+23 -11
View File
@@ -7,10 +7,11 @@ export interface ModelListItem {
model: string
id: string
description: string
thinkingOptionIds: string[]
defaultThinkingOptionId: string | null
thinkingOptions: string
}
const VALID_PROVIDERS = ['claude', 'codex', 'opencode']
/** Schema for provider models output */
export const providerModelsSchema: OutputSchema<ModelListItem> = {
idField: 'id',
@@ -21,10 +22,25 @@ export const providerModelsSchema: OutputSchema<ModelListItem> = {
],
}
const providerModelsWithThinkingSchema: OutputSchema<ModelListItem> = {
idField: 'id',
columns: [
{ header: 'ID', field: 'id', width: 30 },
{ header: 'MODEL', field: 'model', width: 30 },
{ header: 'THINKING IDS', field: 'thinkingOptions', width: 40 },
{
header: 'DEFAULT THINKING',
field: (item) => item.defaultThinkingOptionId ?? 'auto',
width: 18,
},
],
}
export type ProviderModelsResult = ListResult<ModelListItem>
export interface ProviderModelsOptions extends CommandOptions {
host?: string
thinking?: boolean
}
export async function runModelsCommand(
@@ -34,14 +50,6 @@ export async function runModelsCommand(
): Promise<ProviderModelsResult> {
const normalizedProvider = provider.toLowerCase()
if (!VALID_PROVIDERS.includes(normalizedProvider)) {
throw {
code: 'UNKNOWN_PROVIDER',
message: `Unknown provider: ${provider}`,
details: `Valid providers: ${VALID_PROVIDERS.join(', ')}`,
}
}
const client = await connectToDaemon({ host: options.host })
try {
const result = await client.listProviderModels(normalizedProvider)
@@ -57,12 +65,16 @@ export async function runModelsCommand(
model: m.label,
id: m.id,
description: m.description ?? '',
thinkingOptionIds: (m.thinkingOptions ?? []).map((option) => option.id),
defaultThinkingOptionId: m.defaultThinkingOptionId ?? null,
thinkingOptions:
(m.thinkingOptions ?? []).map((option) => option.id).join(', ') || 'none',
}))
return {
type: 'list',
data: models,
schema: providerModelsSchema,
schema: options.thinking ? providerModelsWithThinkingSchema : providerModelsSchema,
}
} finally {
await client.close()
@@ -0,0 +1,21 @@
import { readFileSync } from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
const THIS_DIR = path.dirname(fileURLToPath(import.meta.url));
const DIST_ENCRYPTED_CHANNEL_PATH = path.resolve(THIS_DIR, "../dist/encrypted-channel.js");
describe("relay dist handshake parity", () => {
it("keeps Node dist handshake message types in sync with src", () => {
const distCode = readFileSync(DIST_ENCRYPTED_CHANNEL_PATH, "utf8");
expect(distCode).toContain('type: "e2ee_hello"');
expect(distCode).toContain('type: "e2ee_ready"');
// Guard against accidentally shipping the legacy hello/ready protocol.
expect(distCode).not.toMatch(/\btype:\s*"hello"\b/);
expect(distCode).not.toMatch(/\btype:\s*"ready"\b/);
});
});
+27 -3
View File
@@ -56,6 +56,23 @@ interface E2EEReadyMessage {
type: "e2ee_ready";
}
function buildInvalidHelloError(rawText: string, parsed?: unknown): Error {
const parsedRecord =
parsed && typeof parsed === "object"
? (parsed as Record<string, unknown>)
: null;
const rawType = parsedRecord?.type;
const receivedType =
typeof rawType === "string" ? rawType : rawType === undefined ? "undefined" : typeof rawType;
const hasKey =
typeof parsedRecord?.key === "string" && parsedRecord.key.trim().length > 0;
const compact = rawText.replace(/\s+/g, " ").trim();
const preview = compact.length > 160 ? `${compact.slice(0, 157)}...` : compact;
return new Error(
`Invalid hello message (receivedType=${receivedType}, hasKey=${hasKey}, preview=${JSON.stringify(preview)})`
);
}
const HANDSHAKE_RETRY_MS = 1000;
const MAX_PENDING_SENDS = 200;
@@ -154,9 +171,16 @@ export async function createDaemonChannel(
const helloText =
typeof data === "string" ? data : new TextDecoder().decode(data);
const msg = JSON.parse(helloText) as E2EEHelloMessage;
if (msg.type !== "e2ee_hello" || !msg.key) {
throw new Error("Invalid hello message");
let parsed: unknown;
try {
parsed = JSON.parse(helloText);
} catch {
throw buildInvalidHelloError(helloText);
}
const msg = parsed as Partial<E2EEHelloMessage>;
if (msg.type !== "e2ee_hello" || typeof msg.key !== "string" || !msg.key.trim()) {
throw buildInvalidHelloError(helloText, parsed);
}
// Buffer any subsequent messages that arrive while we're doing async
@@ -1004,6 +1004,86 @@ describe("AgentManager", () => {
expect(lifecycleUpdates).toContain("idle");
});
test("cancelAgentRun can interrupt autonomous running state without a foreground pendingRun", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-cancel-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const liveEvents = new EventPushable<AgentStreamEvent>();
class LiveInterruptSession extends TestAgentSession {
public interruptCount = 0;
async *streamLiveEvents(): AsyncGenerator<AgentStreamEvent> {
for await (const event of liveEvents) {
yield event;
}
}
override async interrupt(): Promise<void> {
this.interruptCount += 1;
}
override async close(): Promise<void> {
liveEvents.end();
}
}
class LiveInterruptClient extends TestAgentClient {
lastSession: LiveInterruptSession | null = null;
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
const session = new LiveInterruptSession(config);
this.lastSession = session;
return session;
}
}
const client = new LiveInterruptClient();
const manager = new AgentManager({
clients: {
codex: client,
},
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000129",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
});
await new Promise<void>((resolve) => {
const unsubscribe = manager.subscribe(
(event) => {
if (event.type !== "agent_state") {
return;
}
if (event.agent.id !== snapshot.id) {
return;
}
if (event.agent.lifecycle !== "running") {
return;
}
unsubscribe();
resolve();
},
{ agentId: snapshot.id, replayState: false }
);
liveEvents.push({ type: "turn_started", provider: "codex" });
});
const beforeCancel = manager.getAgent(snapshot.id);
expect(beforeCancel?.lifecycle).toBe("running");
expect(Boolean(beforeCancel && "pendingRun" in beforeCancel && beforeCancel.pendingRun)).toBe(
false
);
const cancelled = await manager.cancelAgentRun(snapshot.id);
expect(cancelled).toBe(true);
expect(client.lastSession?.interruptCount).toBe(1);
});
test("waitForAgentEvent waitForActive resolves for autonomous live-event run", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-wait-"));
const storagePath = join(workdir, "agents");
@@ -1050,6 +1130,167 @@ describe("AgentManager", () => {
expect(result.status).toBe("idle");
});
test("buffers autonomous live events during foreground run and flushes after run settles", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-buffer-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const liveEvents = new EventPushable<AgentStreamEvent>();
const releaseForeground = deferred<void>();
class BufferedLiveSession extends TestAgentSession {
override async *stream(): AsyncGenerator<AgentStreamEvent> {
yield { type: "turn_started", provider: this.provider };
await releaseForeground.promise;
yield { type: "turn_completed", provider: this.provider };
}
async *streamLiveEvents(): AsyncGenerator<AgentStreamEvent> {
for await (const event of liveEvents) {
yield event;
}
}
override async close(): Promise<void> {
liveEvents.end();
}
}
class BufferedLiveClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new BufferedLiveSession(config);
}
}
const manager = new AgentManager({
clients: {
codex: new BufferedLiveClient(),
},
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000127",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
});
const foreground = manager.streamAgent(snapshot.id, "foreground run");
const consumeForeground = (async () => {
for await (const _event of foreground) {
// Drain foreground stream.
}
})();
await manager.waitForAgentRunStart(snapshot.id);
liveEvents.push({ type: "turn_started", provider: "codex" });
liveEvents.push({
type: "timeline",
provider: "codex",
item: { type: "assistant_message", text: "AUTONOMOUS_DURING_FOREGROUND" },
});
liveEvents.push({ type: "turn_completed", provider: "codex" });
releaseForeground.resolve();
await consumeForeground;
const updated = manager.getAgent(snapshot.id);
expect(updated?.lifecycle).toBe("idle");
expect(manager.getTimeline(snapshot.id)).toContainEqual({
type: "assistant_message",
text: "AUTONOMOUS_DURING_FOREGROUND",
});
});
test("restarts live event pump after iterator failure", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-restart-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const liveEvents = new EventPushable<AgentStreamEvent>();
class FlakyLiveSession extends TestAgentSession {
private attempts = 0;
async *streamLiveEvents(): AsyncGenerator<AgentStreamEvent> {
this.attempts += 1;
if (this.attempts === 1) {
throw new Error("simulated live iterator failure");
}
for await (const event of liveEvents) {
yield event;
}
}
override async close(): Promise<void> {
liveEvents.end();
}
}
class FlakyLiveClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new FlakyLiveSession(config);
}
}
const manager = new AgentManager({
clients: {
codex: new FlakyLiveClient(),
},
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000128",
});
const snapshot = await manager.createAgent({
provider: "codex",
cwd: workdir,
});
// Give the first failed stream a chance to restart.
await new Promise((resolve) => setTimeout(resolve, 350));
const assistantSeen = new Promise<void>((resolve, reject) => {
const timeout = setTimeout(() => {
unsubscribe();
reject(new Error("Timed out waiting for restarted live event"));
}, 2_000);
const unsubscribe = manager.subscribe(
(event) => {
if (event.type !== "agent_stream" || event.agentId !== snapshot.id) {
return;
}
if (
event.event.type === "timeline" &&
event.event.item.type === "assistant_message" &&
event.event.item.text === "AUTONOMOUS_AFTER_RESTART"
) {
clearTimeout(timeout);
unsubscribe();
resolve();
}
},
{ agentId: snapshot.id, replayState: false }
);
});
liveEvents.push({ type: "turn_started", provider: "codex" });
liveEvents.push({
type: "timeline",
provider: "codex",
item: { type: "assistant_message", text: "AUTONOMOUS_AFTER_RESTART" },
});
liveEvents.push({ type: "turn_completed", provider: "codex" });
await assistantSeen;
const result = await manager.waitForAgentEvent(snapshot.id);
expect(result.status).toBe("idle");
expect(manager.getTimeline(snapshot.id)).toContainEqual({
type: "assistant_message",
text: "AUTONOMOUS_AFTER_RESTART",
});
});
test("keeps updatedAt monotonic when user message and run start happen in the same millisecond", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
const storagePath = join(workdir, "agents");
@@ -296,6 +296,7 @@ export class AgentManager {
private readonly previousStatuses = new Map<string, AgentLifecycleStatus>();
private readonly backgroundTasks = new Set<Promise<void>>();
private readonly liveEventPumps = new Map<string, Promise<void>>();
private readonly liveEventBacklog = new Map<string, AgentStreamEvent[]>();
private onAgentAttention?: AgentAttentionCallback;
private logger: Logger;
@@ -743,6 +744,7 @@ export class AgentManager {
// Remove the existing agent entry before swapping sessions
this.agents.delete(agentId);
this.liveEventPumps.delete(agentId);
this.liveEventBacklog.delete(agentId);
try {
await existing.session.close();
} catch (error) {
@@ -773,6 +775,7 @@ export class AgentManager {
const agent = this.requireAgent(agentId);
this.agents.delete(agentId);
this.liveEventPumps.delete(agentId);
this.liveEventBacklog.delete(agentId);
// Clean up previousStatus to prevent memory leak
this.previousStatuses.delete(agentId);
const session = agent.session;
@@ -1004,6 +1007,7 @@ export class AgentManager {
);
}
this.emitState(mutableAgent);
this.flushLiveEventBacklog(mutableAgent);
};
const self = this;
@@ -1147,7 +1151,11 @@ export class AgentManager {
async cancelAgentRun(agentId: string): Promise<boolean> {
const agent = this.requireAgent(agentId);
const pendingRun = agent.pendingRun;
if (!pendingRun || typeof pendingRun.return !== "function") {
const hasForegroundPendingRun =
Boolean(pendingRun) && typeof pendingRun?.return === "function";
const isAutonomousRunning = agent.lifecycle === "running" && !hasForegroundPendingRun;
if (!hasForegroundPendingRun && !isAutonomousRunning) {
return false;
}
@@ -1160,16 +1168,18 @@ export class AgentManager {
);
}
try {
// Await the generator's .return() to ensure the finally block runs
// and pendingRun is properly cleared before we return.
await pendingRun.return(undefined as unknown as AgentStreamEvent);
} catch (error) {
this.logger.error(
{ err: error, agentId },
"Failed to cancel run"
);
throw error;
if (hasForegroundPendingRun && pendingRun) {
try {
// Await the generator's .return() to ensure the finally block runs
// and pendingRun is properly cleared before we return.
await pendingRun.return(undefined as unknown as AgentStreamEvent);
} catch (error) {
this.logger.error(
{ err: error, agentId },
"Failed to cancel run"
);
throw error;
}
}
// Clear any pending permissions that weren't cleaned up by handleStreamEvent.
@@ -1564,6 +1574,7 @@ export class AgentManager {
const newInfo = await agent.session.getRuntimeInfo();
const changed =
newInfo.model !== agent.runtimeInfo?.model ||
newInfo.thinkingOptionId !== agent.runtimeInfo?.thinkingOptionId ||
newInfo.sessionId !== agent.runtimeInfo?.sessionId ||
newInfo.modeId !== agent.runtimeInfo?.modeId;
agent.runtimeInfo = newInfo;
@@ -1926,29 +1937,54 @@ export class AgentManager {
if (!supportsLiveEventStream(agent.session)) {
return;
}
const liveSession = agent.session;
if (this.liveEventPumps.has(agent.id)) {
return;
}
const pump = (async () => {
try {
for await (const event of liveSession.streamLiveEvents()) {
const current = this.agents.get(agent.id);
if (!current) {
break;
while (true) {
const current = this.agents.get(agent.id);
if (!current) {
return;
}
if (!supportsLiveEventStream(current.session)) {
return;
}
try {
for await (const event of current.session.streamLiveEvents()) {
const latest = this.agents.get(agent.id);
if (!latest) {
return;
}
// Keep consuming provider events even during an active foreground run,
// then replay them immediately once that run settles.
if (latest.pendingRun) {
this.enqueueLiveEvent(latest.id, event);
continue;
}
this.flushLiveEventBacklog(latest);
this.handleStreamEvent(latest, event);
}
// Foreground streamAgent owns live prompt events while running.
// Background live-event tailing is for autonomous activity while idle.
if (current.pendingRun) {
continue;
}
this.handleStreamEvent(current, event);
this.logger.warn(
{ agentId: agent.id },
"Live event pump stream ended; restarting"
);
} catch (error) {
this.logger.warn(
{ err: error, agentId: agent.id },
"Live event pump failed"
);
}
// Keep pump alive unless the agent is gone.
await new Promise((resolve) => setTimeout(resolve, 250));
const latest = this.agents.get(agent.id);
if (!latest) {
return;
}
if (!latest.pendingRun) {
this.flushLiveEventBacklog(latest);
}
} catch (error) {
this.logger.warn(
{ err: error, agentId: agent.id },
"Live event pump failed"
);
}
})();
this.liveEventPumps.set(agent.id, pump);
@@ -1959,4 +1995,27 @@ export class AgentManager {
}
});
}
private enqueueLiveEvent(agentId: string, event: AgentStreamEvent): void {
const existing = this.liveEventBacklog.get(agentId);
if (existing) {
existing.push(event);
return;
}
this.liveEventBacklog.set(agentId, [event]);
}
private flushLiveEventBacklog(agent: ActiveManagedAgent): void {
if (agent.pendingRun) {
return;
}
const pending = this.liveEventBacklog.get(agent.id);
if (!pending || pending.length === 0) {
return;
}
this.liveEventBacklog.delete(agent.id);
for (const event of pending) {
this.handleStreamEvent(agent, event);
}
}
}
@@ -24,6 +24,23 @@ type ProjectionOptions = {
internal?: boolean;
};
function normalizeThinkingOptionId(value: string | null | undefined): string | null {
if (typeof value !== "string") return null;
const normalized = value.trim();
return normalized.length > 0 ? normalized : null;
}
export function resolveEffectiveThinkingOptionId(options: {
runtimeInfo?: AgentRuntimeInfo | null;
configuredThinkingOptionId?: string | null;
}): string | null {
const runtimeInfo = options.runtimeInfo;
if (runtimeInfo && "thinkingOptionId" in runtimeInfo) {
return normalizeThinkingOptionId(runtimeInfo.thinkingOptionId);
}
return normalizeThinkingOptionId(options.configuredThinkingOptionId);
}
export function toStoredAgentRecord(
agent: ManagedAgent,
options?: ProjectionOptions
@@ -67,6 +84,10 @@ export function toAgentPayload(
): AgentSnapshotPayload {
const runtimeInfo = sanitizeRuntimeInfo(agent.runtimeInfo);
const thinkingOptionId = agent.config.thinkingOptionId ?? null;
const effectiveThinkingOptionId = resolveEffectiveThinkingOptionId({
runtimeInfo,
configuredThinkingOptionId: thinkingOptionId,
});
const payload: AgentSnapshotPayload = {
id: agent.id,
@@ -74,6 +95,7 @@ export function toAgentPayload(
cwd: agent.cwd,
model: agent.config.model ?? null,
thinkingOptionId,
effectiveThinkingOptionId,
runtimeInfo,
createdAt: agent.createdAt.toISOString(),
updatedAt: agent.updatedAt.toISOString(),
@@ -287,6 +309,9 @@ function sanitizeRuntimeInfo(
if (runtimeInfo.model !== undefined) {
sanitized.model = runtimeInfo.model;
}
if (runtimeInfo.thinkingOptionId !== undefined) {
sanitized.thinkingOptionId = runtimeInfo.thinkingOptionId;
}
if (runtimeInfo.modeId !== undefined) {
sanitized.modeId = runtimeInfo.modeId;
}
@@ -278,6 +278,7 @@ export type AgentRuntimeInfo = {
provider: AgentProvider;
sessionId: string | null;
model?: string | null;
thinkingOptionId?: string | null;
modeId?: string | null;
extra?: AgentMetadata;
};
@@ -49,6 +49,7 @@ const STORED_AGENT_SCHEMA = z.object({
provider: z.string(),
sessionId: z.string().nullable(),
model: z.string().nullable().optional(),
thinkingOptionId: z.string().nullable().optional(),
modeId: z.string().nullable().optional(),
extra: z.record(z.unknown()).optional(),
})
@@ -1,7 +1,4 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { appendFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { ClaudeAgentClient } from "./claude-agent.js";
@@ -295,7 +292,7 @@ describe("ClaudeAgentSession interrupt restart regression", () => {
const secondTurnEvents = await secondTurnPromise;
const secondAssistantText = collectAssistantText(secondTurnEvents);
expect(sdkMocks.query).toHaveBeenCalledTimes(2);
expect(sdkMocks.query.mock.calls.length).toBeGreaterThanOrEqual(2);
expect(secondAssistantText).toContain("NEW_TURN_RESPONSE");
expect(secondAssistantText).not.toContain("OLD_TURN_RESPONSE");
@@ -430,6 +427,299 @@ describe("ClaudeAgentSession interrupt restart regression", () => {
await session.close();
});
test("ignores stale task-notification message_start bursts before prompt replay", async () => {
const logger = createTestLogger();
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-message-start-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "task_notification",
task_id: "task-msg-start-1",
status: "completed",
output_file: "/tmp/task-msg-start-1.txt",
summary: "Background task finished",
session_id: "task-notification-message-start-session",
uuid: "task-msg-start-note-1",
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
parent_tool_use_id: null,
event: {
type: "message_start",
message: {
id: "stale-msg-start-1",
role: "assistant",
model: "opus",
usage: { input_tokens: 1, output_tokens: 0 },
},
},
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
content: "STALE_MESSAGE_START_RESPONSE",
},
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
if (step === 5) {
step += 1;
const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid";
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "current prompt" },
parent_tool_use_id: null,
uuid: promptUuid,
session_id: "task-notification-message-start-session",
isReplay: true,
},
};
}
if (step === 6) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
content: "CURRENT_AFTER_MESSAGE_START",
},
},
};
}
if (step === 7) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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 events = await collectUntilTerminal(session.stream("current prompt"));
const assistantText = collectAssistantText(events);
expect(assistantText).toContain("CURRENT_AFTER_MESSAGE_START");
expect(assistantText).not.toContain("STALE_MESSAGE_START_RESPONSE");
await session.close();
});
test("ignores stale user-shaped task-notification message_start bursts before prompt replay", async () => {
const logger = createTestLogger();
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-user-message-start-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "user",
message: {
role: "user",
content:
"<task-notification>\n<task-id>task-msg-start-user-1</task-id>\n</task-notification>",
},
parent_tool_use_id: null,
uuid: "task-msg-start-user-1",
session_id: "task-notification-user-message-start-session",
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
parent_tool_use_id: null,
event: {
type: "message_start",
message: {
id: "stale-user-msg-start-1",
role: "assistant",
model: "opus",
usage: { input_tokens: 1, output_tokens: 0 },
},
},
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
content: "STALE_USER_TASK_NOTIFICATION_RESPONSE",
},
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
if (step === 5) {
step += 1;
const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid";
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "current prompt" },
parent_tool_use_id: null,
uuid: promptUuid,
session_id: "task-notification-user-message-start-session",
isReplay: true,
},
};
}
if (step === 6) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
content: "CURRENT_AFTER_USER_MESSAGE_START",
},
},
};
}
if (step === 7) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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 events = await collectUntilTerminal(session.stream("current prompt"));
const assistantText = collectAssistantText(events);
expect(assistantText).toContain("CURRENT_AFTER_USER_MESSAGE_START");
expect(assistantText).not.toContain("STALE_USER_TASK_NOTIFICATION_RESPONSE");
await session.close();
});
test("does not terminate the current prompt on a stale pre-prompt result event", async () => {
const logger = createTestLogger();
@@ -528,156 +818,408 @@ describe("ClaudeAgentSession interrupt restart regression", () => {
await session.close();
});
test("emits autonomous live events from history tail when Claude wakes itself", async () => {
test("emits autonomous live events from SDK stream when Claude wakes itself", async () => {
const logger = createTestLogger();
const projectDir = mkdtempSync(path.join(tmpdir(), "claude-live-tail-project-"));
const configDir = mkdtempSync(path.join(tmpdir(), "claude-live-tail-config-"));
const previousClaudeConfigDir = process.env.CLAUDE_CONFIG_DIR;
process.env.CLAUDE_CONFIG_DIR = configDir;
let queryCreateCount = 0;
let localPromptUuid: string | null = null;
sdkMocks.query.mockImplementation(() => {
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "live-autonomous-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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;
sdkMocks.query.mockImplementation(
({ prompt }: { prompt: AsyncIterable<unknown> }) => {
queryCreateCount += 1;
if (queryCreateCount === 1) {
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: "live-autonomous-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
localPromptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid";
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "seed prompt" },
parent_tool_use_id: null,
uuid: localPromptUuid,
session_id: "live-autonomous-session",
isReplay: true,
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "SEED_DONE" },
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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;
}
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "user",
message: {
role: "user",
content:
"<task-notification>\n<task-id>bg-1</task-id>\n<status>completed</status>\n</task-notification>",
},
parent_tool_use_id: null,
uuid: "task-note-user-1",
session_id: "live-autonomous-session",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "AUTONOMOUS_WAKE_RESPONSE" },
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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(),
});
try {
const client = new ClaudeAgentClient({ logger });
const session = await client.createSession({
provider: "claude",
cwd: projectDir,
});
await collectUntilTerminal(session.stream("seed prompt"));
expect(session.describePersistence()?.sessionId).toBe(
"live-autonomous-session"
);
const sanitizedCwd = projectDir.replace(/[\\/\.]/g, "-").replace(/_/g, "-");
const historyDir = path.join(configDir, "projects", sanitizedCwd);
mkdirSync(historyDir, { recursive: true });
const historyPath = path.join(historyDir, "live-autonomous-session.jsonl");
writeFileSync(historyPath, "", "utf8");
const liveIterator = (
session as unknown as {
streamLiveEvents: () => AsyncGenerator<AgentStreamEvent>;
}
).streamLiveEvents();
const timedReader = createTimedIteratorReader({ iterator: liveIterator });
const appendAutonomousBurst = (suffix: string) => {
appendFileSync(
historyPath,
`${JSON.stringify({
type: "system",
subtype: "task_notification",
task_id: `task-${suffix}`,
status: "completed",
output_file: "/tmp/out.txt",
summary: "Background task finished",
uuid: `task-note-uuid-${suffix}`,
session_id: "live-autonomous-session",
})}\n`
);
appendFileSync(
historyPath,
`${JSON.stringify({
type: "assistant",
message: { content: "AUTONOMOUS_WAKE_RESPONSE" },
uuid: `assistant-uuid-${suffix}`,
session_id: "live-autonomous-session",
})}\n`
);
appendFileSync(
historyPath,
`${JSON.stringify({
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
uuid: `result-uuid-${suffix}`,
session_id: "live-autonomous-session",
})}\n`
);
};
const liveEvents: AgentStreamEvent[] = [];
let completed = false;
for (let attempt = 0; attempt < 4 && !completed; attempt += 1) {
appendAutonomousBurst(String(attempt));
while (liveEvents.length < 12) {
let next: IteratorResult<AgentStreamEvent>;
try {
next = await timedReader.nextWithTimeout(2_000);
} catch {
break;
}
if (next.done) {
break;
}
liveEvents.push(next.value);
if (next.value.type === "turn_completed") {
completed = true;
break;
}
}
await collectUntilTerminal(session.stream("seed prompt"));
expect(localPromptUuid).toBeTruthy();
expect(session.describePersistence()?.sessionId).toBe("live-autonomous-session");
for (let attempt = 0; attempt < 80; attempt += 1) {
const activeTurnPromise = (
session as unknown as { activeTurnPromise?: Promise<void> | null }
).activeTurnPromise;
if (!activeTurnPromise) {
break;
}
expect(liveEvents.some((event) => event.type === "turn_started")).toBe(true);
expect(
liveEvents.some(
(event) =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text.includes("AUTONOMOUS_WAKE_RESPONSE")
)
).toBe(true);
expect(liveEvents.some((event) => event.type === "turn_completed")).toBe(true);
await session.close();
} finally {
if (previousClaudeConfigDir === undefined) {
delete process.env.CLAUDE_CONFIG_DIR;
} else {
process.env.CLAUDE_CONFIG_DIR = previousClaudeConfigDir;
}
rmSync(projectDir, { recursive: true, force: true });
rmSync(configDir, { recursive: true, force: true });
await new Promise((resolve) => setTimeout(resolve, 25));
}
expect(
(session as unknown as { activeTurnPromise?: Promise<void> | null })
.activeTurnPromise ?? null
).toBeNull();
const liveIterator = (
session as unknown as {
streamLiveEvents: () => AsyncGenerator<AgentStreamEvent>;
}
).streamLiveEvents();
const timedReader = createTimedIteratorReader({ iterator: liveIterator });
const liveEvents: AgentStreamEvent[] = [];
for (let attempt = 0; attempt < 8; attempt += 1) {
const next = await timedReader.nextWithTimeout(5_000);
if (next.done) {
break;
}
liveEvents.push(next.value);
if (next.value.type === "turn_completed") {
break;
}
}
expect(liveEvents.some((event) => event.type === "turn_started")).toBe(true);
expect(
liveEvents.some(
(event) =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text.includes("AUTONOMOUS_WAKE_RESPONSE")
)
).toBe(true);
expect(liveEvents.some((event) => event.type === "turn_completed")).toBe(true);
await liveIterator.return?.();
await session.close();
});
test("releases local-turn suppression when task notifications arrive as user payloads", async () => {
const logger = createTestLogger();
let queryCreateCount = 0;
let localPromptUuid: string | null = null;
sdkMocks.query.mockImplementation(
({ prompt }: { prompt: AsyncIterable<unknown> }) => {
queryCreateCount += 1;
if (queryCreateCount === 1) {
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: "live-task-user-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
localPromptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid";
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "seed prompt" },
parent_tool_use_id: null,
uuid: localPromptUuid,
session_id: "live-task-user-session",
isReplay: true,
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "SEED_DONE" },
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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;
}
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "seed prompt" },
parent_tool_use_id: null,
uuid: localPromptUuid ?? "missing-prompt-uuid",
session_id: "live-task-user-session",
isReplay: true,
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "SHOULD_STAY_SUPPRESSED" },
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "user",
message: {
role: "user",
content:
"<task-notification>\n<task-id>bg-1</task-id>\n<status>completed</status>\n</task-notification>",
},
parent_tool_use_id: null,
uuid: "task-note-user-1",
session_id: "live-task-user-session",
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "AUTONOMOUS_AFTER_TASK_NOTIFICATION" },
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
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(),
});
await collectUntilTerminal(session.stream("seed prompt"));
expect(localPromptUuid).toBeTruthy();
expect(session.describePersistence()?.sessionId).toBe("live-task-user-session");
for (let attempt = 0; attempt < 80; attempt += 1) {
const activeTurnPromise = (
session as unknown as { activeTurnPromise?: Promise<void> | null }
).activeTurnPromise;
if (!activeTurnPromise) {
break;
}
await new Promise((resolve) => setTimeout(resolve, 25));
}
expect(
(session as unknown as { activeTurnPromise?: Promise<void> | null })
.activeTurnPromise ?? null
).toBeNull();
const liveIterator = (
session as unknown as {
streamLiveEvents: () => AsyncGenerator<AgentStreamEvent>;
}
).streamLiveEvents();
const timedReader = createTimedIteratorReader({ iterator: liveIterator });
const liveEvents: AgentStreamEvent[] = [];
for (let attempt = 0; attempt < 12; attempt += 1) {
const next = await timedReader.nextWithTimeout(5_000);
if (next.done) {
break;
}
liveEvents.push(next.value);
if (next.value.type === "turn_completed") {
break;
}
}
expect(
liveEvents.some(
(event) =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text.includes("SHOULD_STAY_SUPPRESSED")
)
).toBe(false);
expect(liveEvents.some((event) => event.type === "turn_started")).toBe(true);
expect(
liveEvents.some(
(event) =>
event.type === "timeline" &&
event.item.type === "assistant_message" &&
event.item.text.includes("AUTONOMOUS_AFTER_TASK_NOTIFICATION")
)
).toBe(true);
await liveIterator.return?.();
await session.close();
});
});
@@ -0,0 +1,800 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { AgentManager } from "../agent-manager.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js";
const sdkMocks = vi.hoisted(() => ({
query: vi.fn(),
}));
vi.mock("@anthropic-ai/claude-agent-sdk", () => ({
query: sdkMocks.query,
}));
type QueryMock = {
next: ReturnType<typeof vi.fn>;
interrupt: ReturnType<typeof vi.fn>;
return: ReturnType<typeof vi.fn>;
setPermissionMode: ReturnType<typeof vi.fn>;
setModel: ReturnType<typeof vi.fn>;
supportedModels: ReturnType<typeof vi.fn>;
supportedCommands: ReturnType<typeof vi.fn>;
rewindFiles: ReturnType<typeof vi.fn>;
};
function buildUsage() {
return {
input_tokens: 1,
cache_read_input_tokens: 0,
output_tokens: 1,
};
}
function createPromptUuidReader(prompt: AsyncIterable<unknown>) {
const iterator = prompt[Symbol.asyncIterator]();
let cached: Promise<string | null> | null = null;
return async () => {
if (!cached) {
cached = iterator.next().then((next) => {
if (next.done) {
return null;
}
const value = next.value as { uuid?: unknown } | undefined;
return typeof value?.uuid === "string" ? value.uuid : null;
});
}
return cached;
};
}
function createBaseQueryMock(nextImpl: QueryMock["next"]): QueryMock {
return {
next: nextImpl,
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 })),
};
}
async function createSession() {
const client = new ClaudeAgentClient({ logger: createTestLogger() });
return client.createSession({
provider: "claude",
cwd: process.cwd(),
});
}
async function collectUntilTerminal(
stream: AsyncGenerator<AgentStreamEvent>
): Promise<AgentStreamEvent[]> {
const events: AgentStreamEvent[] = [];
for await (const event of stream) {
events.push(event);
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
break;
}
}
return events;
}
async function waitForCondition(
check: () => boolean,
timeoutMs: number
): Promise<void> {
const started = Date.now();
while (Date.now() - started < timeoutMs) {
if (check()) {
return;
}
await new Promise((resolve) => setTimeout(resolve, 20));
}
throw new Error("Timed out waiting for condition");
}
describe("ClaudeAgentSession redesign invariants", () => {
beforeEach(() => {
sdkMocks.query.mockReset();
});
afterEach(() => {
sdkMocks.query.mockReset();
});
test("routes by deterministic identifier priority: task_id > parent_message_id > message_id", async () => {
const session = await createSession();
const internal = session as unknown as {
createRun: (owner: "autonomous", queue: null) => { id: string };
runTracker: {
bindIdentifiers: (
run: { id: string },
ids: { taskId: string | null; parentMessageId: string | null; messageId: string | null }
) => void;
};
routeMessage: (input: {
message: AgentStreamEvent | Record<string, unknown>;
identifiers: { taskId: string | null; parentMessageId: string | null; messageId: string | null };
metadataOnly: boolean;
}) => { run: { id: string } | null; reason: string };
turnState: "autonomous";
};
const taskRun = internal.createRun("autonomous", null);
internal.runTracker.bindIdentifiers(taskRun, {
taskId: "task-A",
parentMessageId: null,
messageId: "msg-A",
});
const parentRun = internal.createRun("autonomous", null);
internal.runTracker.bindIdentifiers(parentRun, {
taskId: null,
parentMessageId: "parent-B",
messageId: "msg-B",
});
const messageRun = internal.createRun("autonomous", null);
internal.runTracker.bindIdentifiers(messageRun, {
taskId: null,
parentMessageId: null,
messageId: "msg-C",
});
internal.turnState = "autonomous";
const taskPriorityRoute = internal.routeMessage({
message: { type: "assistant", message: { content: "task-priority" } },
identifiers: {
taskId: "task-A",
parentMessageId: "parent-B",
messageId: "msg-C",
},
metadataOnly: false,
});
expect(taskPriorityRoute.reason).toBe("task_id");
expect(taskPriorityRoute.run?.id).toBe(taskRun.id);
const parentPriorityRoute = internal.routeMessage({
message: { type: "assistant", message: { content: "parent-priority" } },
identifiers: {
taskId: null,
parentMessageId: "parent-B",
messageId: "msg-C",
},
metadataOnly: false,
});
expect(parentPriorityRoute.reason).toBe("parent_message_id");
expect(parentPriorityRoute.run?.id).toBe(parentRun.id);
const messagePriorityRoute = internal.routeMessage({
message: { type: "assistant", message: { content: "message-priority" } },
identifiers: {
taskId: null,
parentMessageId: null,
messageId: "msg-C",
},
metadataOnly: false,
});
expect(messagePriorityRoute.reason).toBe("message_id");
expect(messagePriorityRoute.run?.id).toBe(messageRun.id);
await session.close();
});
test("does not route unbound events to foreground before prompt replay is observed", async () => {
const session = await createSession();
const internal = session as unknown as {
createRun: (owner: "foreground" | "autonomous", queue: unknown) => { id: string };
runTracker: {
bindIdentifiers: (
run: { id: string },
ids: { taskId: string | null; parentMessageId: string | null; messageId: string | null }
) => void;
};
activeForegroundTurn: { runId: string; queue: unknown } | null;
turnState: "foreground";
routeMessage: (input: {
message: Record<string, unknown>;
identifiers: { taskId: string | null; parentMessageId: string | null; messageId: string | null };
metadataOnly: boolean;
}) => { run: { id: string } | null; reason: string };
};
const foregroundQueueStub = {
push: () => undefined,
end: () => undefined,
[Symbol.asyncIterator]: () => ({
next: async () => ({ done: true as const, value: undefined }),
}),
};
const foregroundRun = internal.createRun("foreground", foregroundQueueStub);
internal.turnState = "foreground";
internal.activeForegroundTurn = {
runId: foregroundRun.id,
queue: foregroundQueueStub,
};
const unboundBeforeReplay = internal.routeMessage({
message: { type: "assistant", message: { content: "stale-before-replay" } },
identifiers: {
taskId: null,
parentMessageId: null,
messageId: null,
},
metadataOnly: false,
});
expect(unboundBeforeReplay.reason).toBe("fallback");
expect(unboundBeforeReplay.run?.id).not.toBe(foregroundRun.id);
const promptReplayId = "foreground-replay-id";
internal.runTracker.bindIdentifiers(foregroundRun, {
taskId: null,
parentMessageId: null,
messageId: promptReplayId,
});
const replayRoute = internal.routeMessage({
message: {
type: "user",
message: { role: "user", content: "prompt replay" },
uuid: promptReplayId,
},
identifiers: {
taskId: null,
parentMessageId: null,
messageId: promptReplayId,
},
metadataOnly: false,
});
expect(replayRoute.reason).toBe("message_id");
expect(replayRoute.run?.id).toBe(foregroundRun.id);
await session.close();
});
test("tracks run lifecycle transitions for success, error, and interrupt", async () => {
const session = await createSession();
let streamCase: "success" | "error" | "interrupt" = "success";
sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
const readPromptUuid = createPromptUuidReader(prompt);
let step = 0;
let interruptRequested = false;
const mock = createBaseQueryMock(
vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "redesign-lifecycle-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "prompt replay" },
parent_tool_use_id: null,
uuid: (await readPromptUuid()) ?? "missing-prompt-uuid",
session_id: "redesign-lifecycle-session",
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "assistant output" },
},
};
}
if (streamCase === "interrupt") {
if (!interruptRequested) {
await new Promise<void>((resolve) => setTimeout(resolve, 50));
return {
done: false,
value: {
type: "assistant",
message: { content: "waiting for interrupt" },
},
};
}
return { done: true, value: undefined };
}
if (step === 3) {
step += 1;
if (streamCase === "success") {
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
return {
done: false,
value: {
type: "result",
subtype: "error",
usage: buildUsage(),
errors: ["simulated failure"],
total_cost_usd: 0,
},
};
}
return { done: true, value: undefined };
})
);
mock.interrupt.mockImplementation(async () => {
interruptRequested = true;
});
return mock;
});
streamCase = "success";
const successEvents = await collectUntilTerminal(session.stream("success prompt"));
expect(successEvents.some((event) => event.type === "turn_completed")).toBe(
true
);
expect(successEvents.some((event) => event.type === "turn_failed")).toBe(
false
);
expect(successEvents.some((event) => event.type === "turn_canceled")).toBe(
false
);
streamCase = "error";
const errorEvents = await collectUntilTerminal(session.stream("error prompt"));
expect(errorEvents.some((event) => event.type === "turn_failed")).toBe(true);
expect(errorEvents.some((event) => event.type === "turn_completed")).toBe(
false
);
streamCase = "interrupt";
const interruptStream = session.stream("interrupt prompt");
const interruptEvents: AgentStreamEvent[] = [];
for await (const event of interruptStream) {
interruptEvents.push(event);
if (
event.type === "timeline" &&
event.item.type === "assistant_message"
) {
await session.interrupt();
}
if (event.type === "turn_canceled") {
break;
}
}
expect(interruptEvents.some((event) => event.type === "turn_canceled")).toBe(
true
);
await session.close();
});
test("assembles assistant timeline when message_delta arrives before message_start", async () => {
sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
const readPromptUuid = createPromptUuidReader(prompt);
let step = 0;
return createBaseQueryMock(
vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "redesign-timeline-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "timeline prompt" },
parent_tool_use_id: null,
uuid: (await readPromptUuid()) ?? "missing-prompt-uuid",
session_id: "redesign-timeline-session",
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "content_block_delta",
delta: { type: "text_delta", text: "HELLO " },
},
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "message_start",
message: { id: "message-1", role: "assistant", model: "opus" },
},
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "content_block_delta",
message_id: "message-1",
delta: { type: "text_delta", text: "WORLD" },
},
},
};
}
if (step === 5) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
event: {
type: "message_stop",
message_id: "message-1",
},
},
};
}
if (step === 6) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
return { done: true, value: undefined };
})
);
});
const session = await createSession();
const events = await collectUntilTerminal(session.stream("timeline prompt"));
const assistantText = events
.filter(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" && event.item.type === "assistant_message"
)
.map((event) => event.item.text)
.join("");
expect(assistantText).toContain("HELLO WORLD");
await session.close();
});
test("does not use stream_event uuid as assistant message identity when message_id is missing", async () => {
sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
const readPromptUuid = createPromptUuidReader(prompt);
let step = 0;
return createBaseQueryMock(
vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "redesign-stream-event-uuid-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "uuid fallback prompt" },
parent_tool_use_id: null,
uuid: (await readPromptUuid()) ?? "missing-prompt-uuid",
session_id: "redesign-stream-event-uuid-session",
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
uuid: "stream-event-uuid-1",
event: {
type: "message_start",
message: { role: "assistant", model: "opus" },
},
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
uuid: "stream-event-uuid-2",
event: {
type: "content_block_delta",
delta: { type: "text_delta", text: "HELLO " },
},
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
uuid: "stream-event-uuid-3",
event: {
type: "content_block_delta",
delta: { type: "text_delta", text: "WORLD" },
},
},
};
}
if (step === 5) {
step += 1;
return {
done: false,
value: {
type: "stream_event",
uuid: "stream-event-uuid-4",
event: {
type: "message_stop",
},
},
};
}
if (step === 6) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
return { done: true, value: undefined };
})
);
});
const session = await createSession();
const events = await collectUntilTerminal(session.stream("uuid fallback prompt"));
const assistantText = events
.filter(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" && event.item.type === "assistant_message"
)
.map((event) => event.item.text)
.join("");
expect(assistantText).toContain("HELLO WORLD");
const assembler = session as unknown as {
timelineAssembler: { messages: Map<string, unknown> };
};
expect(assembler.timelineAssembler.messages.size).toBe(1);
await session.close();
});
test("surfaces autonomous running transitions through manager-visible events during foreground overlap", async () => {
sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
const readPromptUuid = createPromptUuidReader(prompt);
let step = 0;
return createBaseQueryMock(
vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "redesign-manager-overlap-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "user",
message: { role: "user", content: "foreground prompt replay" },
parent_tool_use_id: null,
uuid: (await readPromptUuid()) ?? "missing-prompt-uuid",
session_id: "redesign-manager-overlap-session",
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "FOREGROUND_DONE" },
},
};
}
if (step === 3) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
if (step === 4) {
step += 1;
return {
done: false,
value: {
type: "user",
message: {
role: "user",
content:
"<task-notification>\n<task-id>bg-1</task-id>\n<status>completed</status>\n</task-notification>",
},
parent_tool_use_id: null,
uuid: "task-note-overlap-1",
session_id: "redesign-manager-overlap-session",
},
};
}
if (step === 5) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: { content: "AUTONOMOUS_OVERLAP_DONE" },
},
};
}
if (step === 6) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
return { done: true, value: undefined };
})
);
});
const logger = createTestLogger();
const manager = new AgentManager({
clients: {
claude: new ClaudeAgentClient({ logger }),
},
logger,
});
const agent = await manager.createAgent({
provider: "claude",
cwd: process.cwd(),
});
const runningStateEvents: string[] = [];
const turnStartedEvents: AgentStreamEvent[] = [];
const unsubscribe = manager.subscribe(
(event) => {
if (event.type === "agent_state" && event.agent.id === agent.id) {
if (event.agent.lifecycle === "running") {
runningStateEvents.push(event.agent.lifecycle);
}
return;
}
if (
event.type === "agent_stream" &&
event.agentId === agent.id &&
event.event.type === "turn_started"
) {
turnStartedEvents.push(event.event);
}
},
{ agentId: agent.id, replayState: true }
);
const foregroundEvents = await collectUntilTerminal(
manager.streamAgent(agent.id, "foreground prompt")
);
expect(
foregroundEvents.some((event) => event.type === "turn_completed")
).toBe(true);
await waitForCondition(() => turnStartedEvents.length >= 2, 5_000);
await waitForCondition(() => runningStateEvents.length >= 2, 5_000);
const timeline = manager.getTimeline(agent.id);
expect(
timeline.some(
(item) =>
item.type === "assistant_message" &&
item.text.includes("FOREGROUND_DONE")
)
).toBe(true);
expect(
timeline.some(
(item) =>
item.type === "assistant_message" &&
item.text.includes("AUTONOMOUS_OVERLAP_DONE")
)
).toBe(true);
unsubscribe();
await manager.closeAgent(agent.id);
});
});
File diff suppressed because it is too large Load Diff
@@ -96,6 +96,12 @@ const ClaudeToolDetailPass2Schema = z.union([
toolDetailBranchByName("search", ToolSearchInputSchema, z.unknown(), (input) =>
toSearchToolDetail(input)
),
toolDetailBranchByName("Glob", ToolSearchInputSchema, z.unknown(), (input) =>
toSearchToolDetail(input)
),
toolDetailBranchByName("glob", ToolSearchInputSchema, z.unknown(), (input) =>
toSearchToolDetail(input)
),
ClaudeSpeakToolDetailSchema,
]);
@@ -226,6 +226,25 @@ describe("claude tool-call mapper", () => {
});
});
it("maps Glob calls as search detail using pattern input", () => {
const item = expectMapped(
mapClaudeCompletedToolCall({
callId: "claude-glob-1",
name: "Glob",
input: { pattern: "**/.claude/commands/paseo*" },
output: { output: "No files found" },
})
);
expect(item.status).toBe("completed");
expect(item.error).toBeNull();
expect(item.name).toBe("Glob");
expect(item.detail).toEqual({
type: "search",
query: "**/.claude/commands/paseo*",
});
});
it("normalizes claude speak tool names through schema transforms", () => {
const item = expectMapped(
mapClaudeCompletedToolCall({
@@ -75,6 +75,8 @@ const ClaudeSearchToolNameSchema = z.union([
z.literal("WebSearch"),
z.literal("web_search"),
z.literal("search"),
z.literal("Glob"),
z.literal("glob"),
]);
const ClaudeSpeakToolNameSchema = z
.string()
@@ -2229,6 +2229,7 @@ class CodexAppServerAgentSession implements AgentSession {
provider: CODEX_PROVIDER,
sessionId: this.currentThreadId,
model: this.config.model ?? null,
thinkingOptionId: normalizeCodexThinkingOptionId(this.config.thinkingOptionId) ?? null,
modeId: this.currentMode ?? null,
extra: this.resolvedCollaborationMode
? { collaborationMode: this.resolvedCollaborationMode.name }
@@ -2434,22 +2435,39 @@ class CodexAppServerAgentSession implements AgentSession {
if (!this.client) return;
if (this.currentThreadId) return;
// Resolve model - if not specified, query available models and pick default
// Resolve model + thinking defaults when omitted.
let configuredDefaults: CodexConfiguredDefaults = {};
let model = this.config.model;
if (!model) {
const configuredDefaults = await readCodexConfiguredDefaults(this.client, this.logger);
model = configuredDefaults.model;
let thinkingOptionId = normalizeCodexThinkingOptionId(this.config.thinkingOptionId);
if (!model || !thinkingOptionId) {
configuredDefaults = await readCodexConfiguredDefaults(this.client, this.logger);
}
if (!model) {
model = configuredDefaults.model;
}
if (!thinkingOptionId) {
thinkingOptionId = configuredDefaults.thinkingOptionId;
}
if (!model || !thinkingOptionId) {
const modelResponse = (await this.client.request("model/list", {})) as CodexModelListResponse;
const models = modelResponse?.data ?? [];
const defaultModel = models.find((m) => m.isDefault) ?? models[0];
if (!defaultModel) {
throw new Error("No models available from Codex app-server");
}
model = defaultModel.id;
const selectedModel =
(model ? models.find((candidate) => candidate.id === model) : undefined) ?? defaultModel;
if (!model) {
model = selectedModel.id;
}
if (!thinkingOptionId) {
thinkingOptionId = normalizeCodexThinkingOptionId(selectedModel.defaultReasoningEffort);
}
}
this.config.model = model;
this.config.thinkingOptionId = thinkingOptionId;
const preset = MODE_PRESETS[this.currentMode] ?? MODE_PRESETS[DEFAULT_CODEX_MODE_ID];
const approvalPolicy = this.config.approvalPolicy ?? preset.approvalPolicy;
@@ -509,6 +509,7 @@ export const ToolEditOutputSchema = z.union([
export const ToolSearchInputSchema = z.union([
z.object({ query: z.string() }).passthrough().transform((value) => ({ query: value.query })),
z.object({ q: z.string() }).passthrough().transform((value) => ({ query: value.q })),
z.object({ pattern: z.string() }).passthrough().transform((value) => ({ query: value.pattern })),
]);
export type ParsedToolShellInput = z.infer<typeof ToolShellInputSchema>;
@@ -0,0 +1,811 @@
import { describe, expect, test } from "vitest";
import { existsSync, mkdtempSync, readFileSync, realpathSync, rmSync } from "node:fs";
import { homedir, tmpdir } from "node:os";
import path from "node:path";
import pino from "pino";
import WebSocket from "ws";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { DaemonClient } from "../test-utils/daemon-client.js";
import { ClaudeAgentClient } from "../agent/providers/claude-agent.js";
import { getFullAccessConfig } from "./agent-configs.js";
import { isCommandAvailable } from "../agent/provider-launch-config.js";
function tmpCwd(): string {
return mkdtempSync(path.join(tmpdir(), "daemon-real-claude-autonomous-wake-"));
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function sanitizeClaudeProjectPath(cwd: string): string {
return cwd.replace(/[\\/\.]/g, "-").replace(/_/g, "-");
}
function resolveClaudeTranscriptPath(params: {
cwd: string;
sessionId: string;
}): string {
const configDir = process.env.CLAUDE_CONFIG_DIR ?? path.join(homedir(), ".claude");
const cwdCandidates = [params.cwd];
try {
const realpath = realpathSync(params.cwd);
if (!cwdCandidates.includes(realpath)) {
cwdCandidates.push(realpath);
}
} catch {
// ignore realpath errors for temporary cwd teardown races
}
for (const cwd of cwdCandidates) {
const candidate = path.join(
configDir,
"projects",
sanitizeClaudeProjectPath(cwd),
`${params.sessionId}.jsonl`
);
if (existsSync(candidate)) {
return candidate;
}
}
return path.join(
configDir,
"projects",
sanitizeClaudeProjectPath(params.cwd),
`${params.sessionId}.jsonl`
);
}
type ClaudeTranscriptLine = {
type?: unknown;
timestamp?: unknown;
uuid?: unknown;
parentUuid?: unknown;
message?: { content?: unknown } | null;
};
function readTranscriptLines(pathname: string): ClaudeTranscriptLine[] {
if (!existsSync(pathname)) {
return [];
}
const text = readFileSync(pathname, "utf8");
const rows: ClaudeTranscriptLine[] = [];
for (const line of text.split(/\n+/)) {
const trimmed = line.trim();
if (!trimmed) {
continue;
}
try {
rows.push(JSON.parse(trimmed) as ClaudeTranscriptLine);
} catch {
// ignore malformed rows while transcript is still being appended
}
}
return rows;
}
function readUserText(line: ClaudeTranscriptLine): string {
const content = line.message?.content;
if (typeof content === "string") {
return content;
}
if (!Array.isArray(content)) {
return "";
}
return content
.map((chunk) => {
if (!chunk || typeof chunk !== "object") {
return "";
}
const text = (chunk as { text?: unknown }).text;
if (typeof text === "string") {
return text;
}
const nested = (chunk as { content?: unknown }).content;
return typeof nested === "string" ? nested : "";
})
.join(" ")
.trim();
}
function readAssistantText(line: ClaudeTranscriptLine): string {
const content = line.message?.content;
if (typeof content === "string") {
return content;
}
if (!Array.isArray(content)) {
return "";
}
return content
.map((chunk) => {
if (!chunk || typeof chunk !== "object") {
return "";
}
const text = (chunk as { text?: unknown }).text;
if (typeof text === "string") {
return text;
}
return "";
})
.join("")
.trim();
}
function parseTimestamp(line: ClaudeTranscriptLine): number {
const timestamp = line.timestamp;
if (typeof timestamp !== "string") {
return 0;
}
const parsed = Date.parse(timestamp);
return Number.isFinite(parsed) ? parsed : 0;
}
type TranscriptRaceEvidence = {
helloAssistantText: string;
notificationOutcomeAssistantText: string;
};
function extractTranscriptRaceEvidence(
lines: ClaudeTranscriptLine[]
): TranscriptRaceEvidence | null {
const userLines = lines.filter((line) => line.type === "user");
const assistantLines = lines.filter((line) => line.type === "assistant");
const doItAgain = [...userLines]
.reverse()
.find((line) => readUserText(line).toLowerCase().includes("do it again"));
const helloUser = [...userLines]
.reverse()
.find((line) => readUserText(line).trim().toLowerCase() === "hello");
if (!doItAgain || !helloUser || typeof helloUser.uuid !== "string") {
return null;
}
const helloAssistant = assistantLines.find(
(line) =>
typeof line.parentUuid === "string" &&
line.parentUuid === helloUser.uuid &&
readAssistantText(line).length > 0
);
if (!helloAssistant) {
return null;
}
const doItAgainTs = parseTimestamp(doItAgain);
const taskNotificationUser = userLines.find((line) => {
if (parseTimestamp(line) <= doItAgainTs) {
return false;
}
return readUserText(line).includes("<task-notification>");
});
if (!taskNotificationUser || typeof taskNotificationUser.uuid !== "string") {
return null;
}
const helloAssistantUuid =
typeof helloAssistant.uuid === "string" ? helloAssistant.uuid : null;
const taskNotificationTs = parseTimestamp(taskNotificationUser);
const notificationOutcomeAssistant = [...assistantLines]
.sort((a, b) => parseTimestamp(a) - parseTimestamp(b))
.find((line) => {
const lineTs = parseTimestamp(line);
if (lineTs < taskNotificationTs) {
return false;
}
const lineUuid = typeof line.uuid === "string" ? line.uuid : null;
if (helloAssistantUuid && lineUuid === helloAssistantUuid) {
return false;
}
return readAssistantText(line).length > 0;
});
if (!notificationOutcomeAssistant) {
return null;
}
return {
helloAssistantText: readAssistantText(helloAssistant),
notificationOutcomeAssistantText: readAssistantText(notificationOutcomeAssistant),
};
}
function normalizeText(value: string): string {
return value.replace(/\s+/g, " ").trim().toLowerCase();
}
function summarizeTranscriptTail(lines: ClaudeTranscriptLine[], limit = 14): string {
const tail = lines.slice(-limit);
return tail
.map((line) => {
const ts = typeof line.timestamp === "string" ? line.timestamp : "";
const type = typeof line.type === "string" ? line.type : "unknown";
if (type === "user") {
return `${ts} user ${readUserText(line).slice(0, 160)}`;
}
if (type === "assistant") {
return `${ts} assistant ${readAssistantText(line).slice(0, 160)}`;
}
if (type === "queue-operation") {
const op = (line as { operation?: unknown }).operation;
const content = (line as { content?: unknown }).content;
return `${ts} queue ${typeof op === "string" ? op : "unknown"} ${typeof content === "string" ? content.slice(0, 120) : ""}`;
}
return `${ts} ${type}`;
})
.join("\n");
}
async function runPreHelloNoise(params: {
wsUrl: string;
durationMs: number;
}): Promise<void> {
const deadline = Date.now() + params.durationMs;
while (Date.now() < deadline) {
await new Promise<void>((resolve) => {
const ws = new WebSocket(params.wsUrl);
const fallback = setTimeout(() => {
try {
ws.close();
} catch {
// ignore
}
resolve();
}, 250);
ws.once("open", () => {
try {
// Intentionally send a session message before hello to create the same
// pre-hello churn seen in production logs.
ws.send(
JSON.stringify({
type: "session",
payload: {
type: "fetch_agents_request",
requestId: `noise-${Date.now()}`,
},
})
);
} catch {
// ignore
}
setTimeout(() => {
try {
ws.close();
} catch {
// ignore
}
}, 5);
});
ws.once("close", () => {
clearTimeout(fallback);
resolve();
});
ws.once("error", () => {
clearTimeout(fallback);
resolve();
});
});
}
}
function summarizeTimelineEntry(entry: {
item:
| { type: "user_message"; text: string }
| { type: "assistant_message"; text: string }
| { type: "tool_call"; name: string; status: string }
| { type: string };
}): string {
const item = entry.item;
if (item.type === "user_message") {
return `[user] ${item.text}`;
}
if (item.type === "assistant_message") {
return `[assistant] ${item.text}`;
}
if (item.type === "tool_call") {
return `[tool:${item.name}:${item.status}]`;
}
return `[${item.type}]`;
}
describe("daemon E2E (real claude) - autonomous wake from background task", () => {
test.runIf(isCommandAvailable("claude"))(
"returns to running after background sleep completes without a second prompt",
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-wake-real" },
});
const agent = await client.createAgent({
cwd,
title: "claude-autonomous-wake-real",
...getFullAccessConfig("claude"),
});
await client.sendMessage(
agent.id,
[
"Use a background task to run exactly: sleep 5",
"Do not wait for it to finish.",
"After launching it, reply with exactly: SPAWNED",
].join(" ")
);
const firstFinish = await client.waitForFinish(agent.id, 240_000);
expect(firstFinish.status).toBe("idle");
const timelineBeforeWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const summarized = timelineBeforeWake.entries.map(summarizeTimelineEntry);
// Required by reproduction request: log timeline at idle edge before autonomous wake.
// eslint-disable-next-line no-console
console.log("TIMELINE_BEFORE_AUTONOMOUS_WAKE\n" + summarized.join("\n"));
expect(summarized.some((line) => line.includes("SPAWNED"))).toBe(true);
// No new user prompt here: we expect autonomous transition caused by
// background task completion notification.
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
} finally {
await client.close();
await daemon.close();
rmSync(cwd, { recursive: true, force: true });
}
},
420_000
);
test.runIf(isCommandAvailable("claude"))(
"accepts a new prompt after background sleep finishes and replies HELLO",
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-followup-real" },
});
const agent = await client.createAgent({
cwd,
title: "claude-autonomous-followup-real",
...getFullAccessConfig("claude"),
});
await client.sendMessage(
agent.id,
[
"Use a background task to run exactly: sleep 5",
"Do not wait for it to finish.",
"After launching it, reply with exactly: SPAWNED",
].join(" ")
);
const firstFinish = await client.waitForFinish(agent.id, 240_000);
expect(firstFinish.status).toBe("idle");
await new Promise((resolve) => setTimeout(resolve, 6_000));
await client.sendMessage(agent.id, "say exactly HELLO");
const secondFinish = await client.waitForFinish(agent.id, 240_000);
expect(secondFinish.status).toBe("idle");
expect(secondFinish.lastMessage?.trim()).toBe("HELLO");
} finally {
await client.close();
await daemon.close();
rmSync(cwd, { recursive: true, force: true });
}
},
420_000
);
test.runIf(isCommandAvailable("claude"))(
"repro: do-it-again + immediate hello can hang after autonomous wake under churn",
async () => {
const logger = pino({ level: "silent" });
const cwd = tmpCwd();
const daemon = await createTestPaseoDaemon({
agentClients: { claude: new ClaudeAgentClient({ logger }) },
logger,
});
const wsUrl = `ws://127.0.0.1:${daemon.port}/ws`;
const client = new DaemonClient({ url: wsUrl });
try {
await client.connect();
await client.fetchAgents({
subscribe: { subscriptionId: "claude-hang-repro-real" },
});
const agent = await client.createAgent({
cwd,
title: "claude-hang-repro-real",
...getFullAccessConfig("claude"),
});
for (let cycle = 0; cycle < 20; cycle += 1) {
const noise = runPreHelloNoise({ wsUrl, durationMs: 9_000 });
await client.sendMessage(agent.id, "sleep for 5, in the background");
const firstKickoff = await client.waitForFinish(agent.id, 180_000);
expect(firstKickoff.status).toBe("idle");
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
const firstCompletion = await client.waitForFinish(agent.id, 90_000);
expect(firstCompletion.status).toBe("idle");
await client.sendMessage(agent.id, "do it again");
const secondKickoff = await client.waitForFinish(agent.id, 180_000);
expect(secondKickoff.status).toBe("idle");
await client.sendMessage(agent.id, "hello");
const helloReply = await client.waitForFinish(agent.id, 180_000);
expect(helloReply.status).toBe("idle");
expect((helloReply.lastMessage ?? "").toLowerCase()).toContain("hello");
const runningSnapshot = await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
const timelineAtWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
// eslint-disable-next-line no-console
console.log(
JSON.stringify({
cycle,
wakeUpdatedAt: runningSnapshot.updatedAt,
entriesAtWake: timelineAtWake.entries.length,
})
);
await sleep(6_000);
const timelineAfterWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
expect(timelineAfterWake.entries.length).toBeGreaterThan(
timelineAtWake.entries.length
);
const secondCompletion = await client.waitForFinish(agent.id, 20_000);
expect(secondCompletion.status).toBe("idle");
await noise;
}
} finally {
await client.close();
await daemon.close();
rmSync(cwd, { recursive: true, force: true });
}
},
900_000
);
test.runIf(isCommandAvailable("claude"))(
"repro: second background sleep completion after HELLO should settle back to idle",
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-background-repro-real" },
});
const agent = await client.createAgent({
cwd,
title: "claude-background-repro-real",
...getFullAccessConfig("claude"),
});
await client.sendMessage(agent.id, "sleep for 5, in the background");
const firstKickoff = await client.waitForFinish(agent.id, 180_000);
expect(firstKickoff.status).toBe("idle");
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
const firstCompletion = await client.waitForFinish(agent.id, 60_000);
expect(firstCompletion.status).toBe("idle");
await client.sendMessage(agent.id, "do it again");
const secondKickoff = await client.waitForFinish(agent.id, 180_000);
expect(secondKickoff.status).toBe("idle");
await client.sendMessage(agent.id, "hello");
const helloReply = await client.waitForFinish(agent.id, 180_000);
expect(helloReply.status).toBe("idle");
expect((helloReply.lastMessage ?? "").toLowerCase()).toContain("hello");
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
// This is the failure point seen in production: agent flips to running
// on task completion and never settles back to idle.
const secondCompletion = await client.waitForFinish(agent.id, 20_000);
expect(secondCompletion.status).toBe("idle");
} finally {
await client.close();
await daemon.close();
rmSync(cwd, { recursive: true, force: true });
}
},
600_000
);
test.runIf(isCommandAvailable("claude"))(
"stress: immediate HELLO before task notification should not leave autonomous run stuck",
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-race-stress-real" },
});
const agent = await client.createAgent({
cwd,
title: "claude-autonomous-race-stress-real",
...getFullAccessConfig("claude"),
});
for (let cycle = 0; cycle < 60; cycle += 1) {
const helloToken = `HELLO_CYCLE_${cycle}`;
await client.sendMessage(
agent.id,
[
"Use the Task tool (not a Bash background process).",
"Start a background task that runs exactly: sleep 1",
"Do not wait for the task result.",
"Immediately reply with exactly: SPAWNED",
].join(" ")
);
const firstFinish = await client.waitForFinish(agent.id, 240_000);
expect(firstFinish.status).toBe("idle");
// Race path under test: immediately send the next user prompt before
// the background-task completion notification wakes Claude again.
await client.sendMessage(agent.id, `say exactly ${helloToken}`);
const secondFinish = await client.waitForFinish(agent.id, 240_000);
expect(secondFinish.status).toBe("idle");
expect(secondFinish.lastMessage?.trim()).toBe(helloToken);
const wakeSnapshot = await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
20_000
);
const timelineAtWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
// eslint-disable-next-line no-console
console.log(
JSON.stringify({
cycle,
autonomousWakeUpdatedAt: wakeSnapshot.updatedAt,
timelineEntriesAtWake: timelineAtWake.entries.length,
})
);
await sleep(3_000);
const timelineAfterWake = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
expect(timelineAfterWake.entries.length).toBeGreaterThan(
timelineAtWake.entries.length
);
await client.cancelAgent(agent.id);
const afterCancel = await client.waitForFinish(agent.id, 10_000);
expect(afterCancel.status).toBe("idle");
}
} finally {
await client.close();
await daemon.close();
rmSync(cwd, { recursive: true, force: true });
}
},
900_000
);
test.runIf(isCommandAvailable("claude"))(
"repro: transcript/timeline parity after do-it-again + hello race (hang + interrupt + drop)",
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-transcript-parity-race" },
});
const agent = await client.createAgent({
cwd,
title: "claude-transcript-parity-race",
...getFullAccessConfig("claude"),
});
await client.sendMessage(agent.id, "sleep for 5, in the background");
const firstKickoff = await client.waitForFinish(agent.id, 180_000);
expect(firstKickoff.status).toBe("idle");
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
const firstAutonomousCompletion = await client.waitForFinish(agent.id, 60_000);
expect(firstAutonomousCompletion.status).toBe("idle");
await client.sendMessage(agent.id, "do it again");
const secondKickoff = await client.waitForFinish(agent.id, 180_000);
expect(secondKickoff.status).toBe("idle");
await client.waitForAgentUpsert(
agent.id,
(snapshot) => snapshot.status === "running",
30_000
);
await sleep(500);
await client.sendMessage(agent.id, "hello");
let waitError: Error | null = null;
try {
await client.waitForFinish(agent.id, 20_000);
} catch (error) {
waitError =
error instanceof Error ? error : new Error(String(error ?? "wait_for_finish failed"));
}
let afterWait = await client.fetchAgent(agent.id);
let cancelRecovered = true;
if (afterWait?.status === "running") {
await client.cancelAgent(agent.id);
try {
const afterCancel = await client.waitForFinish(agent.id, 15_000);
cancelRecovered = afterCancel.status === "idle";
} catch {
cancelRecovered = false;
}
afterWait = await client.fetchAgent(agent.id);
}
expect(waitError).toBeNull();
expect(cancelRecovered).toBe(true);
const sessionId =
afterWait?.persistence?.sessionId ??
afterWait?.runtimeInfo?.sessionId ??
null;
expect(typeof sessionId === "string" && sessionId.length > 0).toBe(true);
const transcriptCwd = afterWait?.cwd ?? cwd;
const transcriptPath = resolveClaudeTranscriptPath({
cwd: transcriptCwd,
sessionId: sessionId as string,
});
let evidence: TranscriptRaceEvidence | null = null;
const transcriptDeadline = Date.now() + 20_000;
while (Date.now() < transcriptDeadline) {
const lines = readTranscriptLines(transcriptPath);
evidence = extractTranscriptRaceEvidence(lines);
if (evidence) {
break;
}
await sleep(250);
}
if (!evidence) {
const transcriptDump = summarizeTranscriptTail(readTranscriptLines(transcriptPath));
throw new Error(
[
"Failed to extract transcript race evidence (hello assistant and notification-turn assistant).",
`transcriptPath=${transcriptPath}`,
`waitError=${waitError ? waitError.message : "null"}`,
`afterWaitStatus=${afterWait?.status ?? "unknown"}`,
`cancelRecovered=${cancelRecovered}`,
"transcriptTail:",
transcriptDump,
].join("\n")
);
}
const timeline = await client.fetchAgentTimeline(agent.id, {
direction: "tail",
limit: 0,
projection: "canonical",
});
const assistantTexts = timeline.entries
.filter(
(
entry
): entry is {
item: { type: "assistant_message"; text: string };
} => entry.item.type === "assistant_message"
)
.map((entry) => normalizeText(entry.item.text));
const helloAssistant = normalizeText(evidence.helloAssistantText);
const notificationAssistant = normalizeText(
evidence.notificationOutcomeAssistantText
);
expect(
assistantTexts.some((text) => text.includes(helloAssistant))
).toBe(true);
expect(
assistantTexts.some((text) => text.includes(notificationAssistant))
).toBe(true);
} finally {
await client.close();
await daemon.close();
rmSync(cwd, { recursive: true, force: true });
}
},
600_000
);
});
+12 -12
View File
@@ -291,25 +291,25 @@ export function startRelayTransport({
}, delayMs);
};
const ensureClientDataSocket = (routingId: string): void => {
const ensureClientDataSocket = (connectionId: string): void => {
if (stopped) return;
if (!routingId) return;
if (dataSockets.has(routingId)) return;
if (!connectionId) return;
if (dataSockets.has(connectionId)) return;
const url = buildRelayWebSocketUrl({
endpoint: relayEndpoint,
serverId,
role: "server",
connectionId: routingId,
connectionId,
});
const socket = new WebSocket(url, { handshakeTimeout: 10_000, perMessageDeflate: false });
dataSockets.set(routingId, socket);
dataSockets.set(connectionId, socket);
let attached = false;
const openTimeout = setTimeout(() => {
if (stopped) return;
if (socket.readyState === WebSocket.OPEN) return;
relayLogger.warn({ url, routingId }, "relay_data_open_timeout_terminating");
relayLogger.warn({ url, connectionId }, "relay_data_open_timeout_terminating");
try {
socket.terminate();
} catch {
@@ -319,7 +319,7 @@ export function startRelayTransport({
socket.on("open", () => {
clearTimeout(openTimeout);
relayLogger.info({ url, routingId }, "relay_data_connected");
relayLogger.info({ url, connectionId }, "relay_data_connected");
if (attached) return;
attached = true;
const externalMetadata: ExternalSocketMetadata = {
@@ -329,7 +329,7 @@ export function startRelayTransport({
void attachEncryptedSocket(
socket,
daemonKeyPair,
relayLogger.child({ routingId }),
relayLogger.child({ connectionId }),
attachSocket,
externalMetadata
);
@@ -341,16 +341,16 @@ export function startRelayTransport({
socket.on("close", (code, reason) => {
clearTimeout(openTimeout);
relayLogger.warn(
{ code, reason: reason?.toString?.(), url, routingId },
{ code, reason: reason?.toString?.(), url, connectionId },
"relay_data_disconnected"
);
if (dataSockets.get(routingId) === socket) {
dataSockets.delete(routingId);
if (dataSockets.get(connectionId) === socket) {
dataSockets.delete(connectionId);
}
});
socket.on("error", (err) => {
relayLogger.warn({ err, url, routingId }, "relay_data_error");
relayLogger.warn({ err, url, connectionId }, "relay_data_error");
});
};
+37 -1
View File
@@ -63,7 +63,7 @@ import type {
ManagedAgent,
} from './agent/agent-manager.js'
import { scheduleAgentMetadataGeneration } from './agent/agent-metadata-generator.js'
import { toAgentPayload } from './agent/agent-projections.js'
import { resolveEffectiveThinkingOptionId, toAgentPayload } from './agent/agent-projections.js'
import {
appendTimelineItemIfAgentKnown,
emitLiveTimelineItemIfAgentKnown,
@@ -953,12 +953,33 @@ export class Session {
const lastUserMessageAt = record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null
const provider = coerceAgentProvider(this.sessionLogger, record.provider, record.id)
const runtimeInfo = record.runtimeInfo
? {
provider: coerceAgentProvider(this.sessionLogger, record.runtimeInfo.provider, record.id),
sessionId: record.runtimeInfo.sessionId,
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, 'model')
? { model: record.runtimeInfo.model ?? null }
: {}),
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, 'thinkingOptionId')
? { thinkingOptionId: record.runtimeInfo.thinkingOptionId ?? null }
: {}),
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, 'modeId')
? { modeId: record.runtimeInfo.modeId ?? null }
: {}),
...(record.runtimeInfo.extra ? { extra: record.runtimeInfo.extra } : {}),
}
: undefined
return {
id: record.id,
provider,
cwd: record.cwd,
model: record.config?.model ?? null,
thinkingOptionId: record.config?.thinkingOptionId ?? null,
effectiveThinkingOptionId: resolveEffectiveThinkingOptionId({
runtimeInfo,
configuredThinkingOptionId: record.config?.thinkingOptionId ?? null,
}),
...(runtimeInfo ? { runtimeInfo } : {}),
createdAt: createdAt.toISOString(),
updatedAt: updatedAt.toISOString(),
lastUserMessageAt: lastUserMessageAt ? lastUserMessageAt.toISOString() : null,
@@ -1055,6 +1076,21 @@ export class Session {
return false
}
if (filter?.thinkingOptionId !== undefined) {
const expectedThinkingOptionId = resolveEffectiveThinkingOptionId({
configuredThinkingOptionId: filter.thinkingOptionId ?? null,
})
const resolvedThinkingOptionId =
agent.effectiveThinkingOptionId ??
resolveEffectiveThinkingOptionId({
runtimeInfo: agent.runtimeInfo,
configuredThinkingOptionId: agent.thinkingOptionId ?? null,
})
if (resolvedThinkingOptionId !== expectedThinkingOptionId) {
return false
}
}
if (filter?.statuses && filter.statuses.length > 0) {
const statuses = new Set(filter.statuses)
if (!statuses.has(agent.status)) {
+3
View File
@@ -366,6 +366,7 @@ const AgentRuntimeInfoSchema: z.ZodType<AgentRuntimeInfo> = z.object({
provider: AgentProviderSchema,
sessionId: z.string().nullable(),
model: z.string().nullable().optional(),
thinkingOptionId: z.string().nullable().optional(),
modeId: z.string().nullable().optional(),
extra: z.record(z.unknown()).optional(),
})
@@ -376,6 +377,7 @@ export const AgentSnapshotPayloadSchema = z.object({
cwd: z.string(),
model: z.string().nullable(),
thinkingOptionId: z.string().nullable().optional(),
effectiveThinkingOptionId: z.string().nullable().optional(),
createdAt: z.string(),
updatedAt: z.string(),
lastUserMessageAt: z.string().nullable(),
@@ -426,6 +428,7 @@ const AgentDirectoryFilterSchema = z.object({
statuses: z.array(AgentStatusSchema).optional(),
includeArchived: z.boolean().optional(),
requiresAttention: z.boolean().optional(),
thinkingOptionId: z.string().nullable().optional(),
})
export const DeleteAgentRequestMessageSchema = z.object({