mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 16:03:49 +00:00
Fix Claude structured native chat publication
This commit is contained in:
@@ -61,13 +61,13 @@ describe('Claude resume process proof', () => {
|
||||
).rejects.toThrow('PID-reuse-safe start time')
|
||||
})
|
||||
|
||||
it('matches adjacent resume and session-id argv tokens exactly', () => {
|
||||
it('matches only adjacent resume argv tokens exactly', () => {
|
||||
expect(
|
||||
isClaudeResumeProcessCommandLine(`claude --resume ${SESSION_ID}`, SESSION_ID, 'darwin')
|
||||
).toBe(true)
|
||||
expect(
|
||||
isClaudeResumeProcessCommandLine(`claude --session-id ${SESSION_ID}`, SESSION_ID, 'darwin')
|
||||
).toBe(true)
|
||||
).toBe(false)
|
||||
expect(
|
||||
isClaudeResumeProcessCommandLine(`claude --resume ${SESSION_ID}-other`, SESSION_ID, 'darwin')
|
||||
).toBe(false)
|
||||
|
||||
@@ -11,10 +11,7 @@ export function isClaudeResumeProcessCommandLine(
|
||||
if (!tokens || !sessionId) {
|
||||
return false
|
||||
}
|
||||
return tokens.some(
|
||||
(token, index) =>
|
||||
(token === '--resume' || token === '--session-id') && tokens[index + 1] === sessionId
|
||||
)
|
||||
return tokens.some((token, index) => token === '--resume' && tokens[index + 1] === sessionId)
|
||||
}
|
||||
|
||||
type StructuredTuiIdentityInput = Parameters<typeof readStructuredTuiProcessIdentity>[0]
|
||||
|
||||
@@ -1,7 +1,20 @@
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { PassThrough } from 'node:stream'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { spawnProcess } from '../../shared/child-process/run-process'
|
||||
|
||||
const { forceTerminateProcessTreeMock, waitForProcessExitUntilMock } = vi.hoisted(() => ({
|
||||
forceTerminateProcessTreeMock: vi.fn(),
|
||||
waitForProcessExitUntilMock: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('../../shared/child-process/process-tree-termination', () => ({
|
||||
forceTerminateProcessTree: forceTerminateProcessTreeMock
|
||||
}))
|
||||
|
||||
vi.mock('../codex/codex-process-exit-deadline', () => ({
|
||||
waitForProcessExitUntil: waitForProcessExitUntilMock
|
||||
}))
|
||||
import {
|
||||
openClaudeStreamJsonConnection,
|
||||
type ClaudeControlRequest
|
||||
@@ -43,6 +56,11 @@ function writtenFrames(child: FakeChild): Promise<Record<string, unknown>[]> {
|
||||
}
|
||||
|
||||
describe('Claude stream-json connection', () => {
|
||||
beforeEach(() => {
|
||||
forceTerminateProcessTreeMock.mockReset().mockResolvedValue(true)
|
||||
waitForProcessExitUntilMock.mockReset().mockResolvedValue(undefined)
|
||||
})
|
||||
|
||||
afterEach(() => vi.unstubAllEnvs())
|
||||
|
||||
it('spawns in the pinned workspace and routes acknowledged control requests', async () => {
|
||||
@@ -194,4 +212,19 @@ describe('Claude stream-json connection', () => {
|
||||
])
|
||||
await expect(connection.close()).resolves.toBe(true)
|
||||
})
|
||||
|
||||
it('retries process termination after an unproven close', async () => {
|
||||
forceTerminateProcessTreeMock.mockResolvedValueOnce(false).mockResolvedValueOnce(true)
|
||||
const process = fakeSpawn()
|
||||
const connection = await openClaudeStreamJsonConnection(
|
||||
{ command: 'claude', args: [], cwd: '/work' },
|
||||
{},
|
||||
process.spawnImpl
|
||||
)
|
||||
|
||||
await expect(connection.close()).resolves.toBe(false)
|
||||
process.child.emit('exit', 0, null)
|
||||
await expect(connection.close()).resolves.toBe(true)
|
||||
expect(forceTerminateProcessTreeMock).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -253,7 +253,7 @@ export async function openClaudeStreamJsonConnection(
|
||||
return promise
|
||||
}
|
||||
|
||||
const close = (): Promise<boolean> => {
|
||||
const close = async (): Promise<boolean> => {
|
||||
closePromise ??= (async () => {
|
||||
closing = true
|
||||
try {
|
||||
@@ -267,7 +267,11 @@ export async function openClaudeStreamJsonConnection(
|
||||
failPending(new Error('claude stream-json connection closed'))
|
||||
return descendantsExited && exited
|
||||
})()
|
||||
return closePromise
|
||||
const provenExited = await closePromise
|
||||
if (!provenExited) {
|
||||
closePromise = null
|
||||
}
|
||||
return provenExited
|
||||
}
|
||||
|
||||
return {
|
||||
|
||||
@@ -26,7 +26,7 @@ function userMessage(blocks: AgentJournalMessageItem['blocks']): AgentJournalMes
|
||||
}
|
||||
|
||||
describe('Claude structured text dispatch', () => {
|
||||
it('accepts a slash command from its result receipt when Claude omits the user replay', async () => {
|
||||
it('leaves a result-only slash command receipt unconfirmed', async () => {
|
||||
const session = sessionFor()
|
||||
const dispatched = dispatchClaudeTurn(
|
||||
session,
|
||||
@@ -43,12 +43,8 @@ describe('Claude structured text dispatch', () => {
|
||||
})
|
||||
|
||||
await expect(dispatched).resolves.toEqual({
|
||||
state: 'accepted',
|
||||
providerIdentity: {
|
||||
provider: 'claude',
|
||||
sessionId: 'provider-session',
|
||||
uuid: 'command-result-uuid'
|
||||
}
|
||||
state: 'unknown',
|
||||
reason: 'claude accepted a message but did not replay its uuid in time'
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -9,18 +9,10 @@ export function resolveClaudeReplayWaiter(
|
||||
message: Record<string, unknown>
|
||||
): void {
|
||||
const isUserReplay = message.type === 'user' && message.parent_tool_use_id === null
|
||||
const isCompletedCommand = message.type === 'result'
|
||||
if (
|
||||
(!isUserReplay && !isCompletedCommand) ||
|
||||
readClaudeFrameString(message, 'session_id') !== session.providerSessionId
|
||||
) {
|
||||
if (!isUserReplay || readClaudeFrameString(message, 'session_id') !== session.providerSessionId) {
|
||||
return
|
||||
}
|
||||
const uuid = readClaudeFrameString(message, 'uuid')
|
||||
const current = session.dispatchWaiters[0]
|
||||
if (isCompletedCommand && !current?.acceptsResult) {
|
||||
return
|
||||
}
|
||||
const waiter = uuid ? session.dispatchWaiters.shift() : undefined
|
||||
if (waiter && uuid) {
|
||||
clearTimeout(waiter.timer)
|
||||
@@ -46,14 +38,9 @@ async function messageContent(body: AgentJournalMessageItem): Promise<unknown[]>
|
||||
return content
|
||||
}
|
||||
|
||||
function waitForReplay(
|
||||
session: ClaudeSession,
|
||||
timeoutMs: number,
|
||||
acceptsResult: boolean
|
||||
): Promise<string | null> {
|
||||
function waitForReplay(session: ClaudeSession, timeoutMs: number): Promise<string | null> {
|
||||
return new Promise((resolve) => {
|
||||
const waiter = {
|
||||
acceptsResult,
|
||||
resolve,
|
||||
timer: setTimeout(() => {
|
||||
const index = session.dispatchWaiters.indexOf(waiter)
|
||||
@@ -79,10 +66,7 @@ export async function dispatchClaudeTurn(
|
||||
} catch (error) {
|
||||
return { state: 'rejected', reason: (error as Error).message }
|
||||
}
|
||||
const acceptsResult = input.body.blocks.some(
|
||||
(block) => block.type === 'text' && block.text.trimStart().startsWith('/')
|
||||
)
|
||||
const replayed = waitForReplay(session, timeoutMs, acceptsResult)
|
||||
const replayed = waitForReplay(session, timeoutMs)
|
||||
try {
|
||||
await session.connection.send({
|
||||
type: 'user',
|
||||
|
||||
@@ -62,6 +62,7 @@ function fakeClaude(
|
||||
settings?: unknown
|
||||
replayUuid?: string | null
|
||||
routes?: Record<string, Route>
|
||||
closeResults?: boolean[]
|
||||
} = {}
|
||||
): {
|
||||
connections: FakeConnection[]
|
||||
@@ -135,7 +136,7 @@ function fakeClaude(
|
||||
close: async () => {
|
||||
connection.closeCount += 1
|
||||
connection.closed = true
|
||||
return true
|
||||
return options.closeResults?.shift() ?? true
|
||||
}
|
||||
}
|
||||
connections.push(connection)
|
||||
@@ -540,6 +541,35 @@ describe('ClaudeStructuredSessionAdapter turns and controls', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeStructuredSessionAdapter shutdown', () => {
|
||||
it('returns false and retains a published session until exit is proven', async () => {
|
||||
const claude = fakeClaude({ closeResults: [false, true] })
|
||||
const adapter = await acquired(claude)
|
||||
|
||||
await expect(adapter.closeSession('session-1')).resolves.toBe(false)
|
||||
await expect(adapter.closeSession('session-1')).resolves.toBe(true)
|
||||
expect(claude.connections[0].closeCount).toBe(2)
|
||||
})
|
||||
|
||||
it('retries shutdown until a later attempt proves child exit', async () => {
|
||||
const claude = fakeClaude({ closeResults: [false, true] })
|
||||
const adapter = await acquired(claude)
|
||||
|
||||
await expect(adapter.closeAll()).resolves.toBeUndefined()
|
||||
expect(claude.connections[0].closeCount).toBe(2)
|
||||
})
|
||||
|
||||
it('bounds persistent unproven shutdown attempts', async () => {
|
||||
const claude = fakeClaude({ closeResults: [false, false, false] })
|
||||
const adapter = await acquired(claude)
|
||||
|
||||
await expect(adapter.closeAll()).rejects.toThrow(
|
||||
'claude structured session teardown could not prove provider-child exit'
|
||||
)
|
||||
expect(claude.connections[0].closeCount).toBe(3)
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeStructuredSessionAdapter prompts', () => {
|
||||
it('turns can_use_tool into an addressable durable approval callback', async () => {
|
||||
const claude = fakeClaude()
|
||||
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
rethrowAfterAgentSessionAcquisitionCleanup as rethrowAfterCleanup
|
||||
} from '../native-chat/agent-session-wire/structured-agent-session-adapter'
|
||||
import type { AgentSessionExecutionLocation } from '../../shared/agent-session-record'
|
||||
import { closeProcessRegistry } from '../../shared/child-process/close-process-registry'
|
||||
import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host'
|
||||
import { openClaudeStreamJsonConnection } from './claude-stream-json-connection'
|
||||
import { answerClaudePrompt, cancelClaudeTurn } from './claude-structured-control-actions'
|
||||
@@ -255,11 +256,13 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
|
||||
|
||||
async closeAll(): Promise<void> {
|
||||
this.acquisitions.close()
|
||||
const ids = new Set([...this.sessions.keys(), ...this.acquisitions.sessionIds()])
|
||||
const outcomes = await Promise.all([...ids].map((sessionId) => this.closeSession(sessionId)))
|
||||
if (outcomes.some((exited) => !exited)) {
|
||||
throw new Error('claude structured session teardown could not prove provider-child exit')
|
||||
}
|
||||
await closeProcessRegistry({
|
||||
attempts: 3,
|
||||
hasEntries: () => this.sessions.size > 0 || this.acquisitions.size > 0,
|
||||
entryIds: () => new Set([...this.sessions.keys(), ...this.acquisitions.sessionIds()]),
|
||||
closeEntry: (sessionId) => this.closeSession(sessionId),
|
||||
failureMessage: 'claude structured session teardown could not prove provider-child exit'
|
||||
})
|
||||
}
|
||||
|
||||
private session(sessionId: string): ClaudeSession {
|
||||
|
||||
@@ -71,7 +71,6 @@ export function observeClaudeTopLevelLeaf(
|
||||
export type ClaudeDispatchWaiter = {
|
||||
resolve: (uuid: string | null) => void
|
||||
timer: ReturnType<typeof setTimeout>
|
||||
acceptsResult: boolean
|
||||
}
|
||||
|
||||
export type ClaudeSession = {
|
||||
|
||||
@@ -123,6 +123,44 @@ describe('registerRuntimeHandlers', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('publishes Claude structured tabs to the local desktop renderer', async () => {
|
||||
const claudeTab = {
|
||||
type: 'agent-session' as const,
|
||||
id: 'agent-session:claude-1',
|
||||
title: 'Claude Chat',
|
||||
sessionId: 'claude-1',
|
||||
agent: 'claude' as const,
|
||||
isActive: true
|
||||
}
|
||||
const runtime = {
|
||||
getRuntimeId: vi.fn().mockReturnValue('runtime-1'),
|
||||
restoreStructuredAgentSessionTabs: vi.fn().mockResolvedValue(undefined),
|
||||
supportsAuthoritativeSessionTabsInventory: vi.fn().mockReturnValue(false),
|
||||
listAllMobileSessionTabs: vi.fn().mockResolvedValue([
|
||||
{
|
||||
worktree: 'worktree-1',
|
||||
publicationEpoch: 'structured-1',
|
||||
snapshotVersion: 1,
|
||||
activeGroupId: 'group-1',
|
||||
activeTabId: claudeTab.id,
|
||||
activeTabType: 'agent-session',
|
||||
tabGroups: [{ id: 'group-1', activeTabId: claudeTab.id, tabOrder: [claudeTab.id] }],
|
||||
tabs: [claudeTab]
|
||||
}
|
||||
])
|
||||
}
|
||||
|
||||
registerRuntimeHandlers(runtime as never)
|
||||
|
||||
const handler = handleMock.mock.calls.find(([channel]) => channel === 'runtime:call')![1]
|
||||
const result = await handler({ sender: {} }, { method: 'session.tabs.listAll', params: {} })
|
||||
|
||||
expect(result).toMatchObject({
|
||||
ok: true,
|
||||
result: { snapshots: [{ tabs: [claudeTab] }] }
|
||||
})
|
||||
})
|
||||
|
||||
it('registers project group runtime RPC methods for local desktop callers', async () => {
|
||||
const runtime = {
|
||||
syncWindowGraph: vi.fn(),
|
||||
|
||||
+11
-3
@@ -10,10 +10,18 @@ import type {
|
||||
import type { RuntimeRpcResponse } from '../../shared/runtime-rpc-envelope'
|
||||
import type { ClientHostedBrowserRowsEvent } from '../../shared/client-hosted-browser-rows'
|
||||
import { TERMINAL_FIT_RESTORE_DEADLINE_MS } from '../../shared/terminal-fit-restore-deadline'
|
||||
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../shared/protocol-version'
|
||||
import {
|
||||
CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
|
||||
} from '../../shared/protocol-version'
|
||||
import { RpcDispatcher } from '../runtime/rpc/dispatcher'
|
||||
import { ALL_RPC_METHODS } from '../runtime/rpc/methods'
|
||||
|
||||
const LOCAL_RUNTIME_CLIENT_CAPABILITIES = [
|
||||
STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
|
||||
CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY
|
||||
] as const
|
||||
|
||||
function boundTerminalFitRestore(pending: Promise<boolean>): Promise<boolean> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
const deadline = new Promise<boolean>((resolve) => {
|
||||
@@ -71,7 +79,7 @@ export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
|
||||
{
|
||||
clientId: 'desktop-renderer',
|
||||
clientKind: 'runtime',
|
||||
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY]
|
||||
clientCapabilities: LOCAL_RUNTIME_CLIENT_CAPABILITIES
|
||||
}
|
||||
)) as RuntimeRpcResponse<unknown>
|
||||
}
|
||||
@@ -113,7 +121,7 @@ export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
|
||||
signal: controller.signal,
|
||||
clientId: 'desktop-renderer',
|
||||
clientKind: 'runtime',
|
||||
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY]
|
||||
clientCapabilities: LOCAL_RUNTIME_CLIENT_CAPABILITIES
|
||||
}
|
||||
)
|
||||
.finally(stop)
|
||||
|
||||
Reference in New Issue
Block a user