feat: wire Claude structured sessions

This commit is contained in:
Brennan Benson
2026-08-10 04:25:54 -07:00
parent 2a0d5bc32f
commit 23ce5e5a67
24 changed files with 1547 additions and 63 deletions
@@ -0,0 +1,158 @@
import type {
AgentJournalItemBody,
AgentJournalItemIdentity,
AgentJournalMessageItem
} from '../../shared/agent-session-journal-types'
import type { NativeChatBlock } from '../../shared/native-chat-types'
import {
boundInlineText,
DEFAULT_JOURNAL_PAYLOAD_LIMITS
} from '../native-chat/agent-session-journal/journal-payload-bounds'
export type ClaudeMessageEnvelope = {
sessionId: string
uuid: string
role: 'assistant' | 'user'
content: unknown[]
}
export type ClaudeToolUse = { id: string; name: string; input: unknown }
export type ClaudeToolResult = { toolUseId: string; output: string; failed: boolean }
export function claudeRecord(value: unknown): Record<string, unknown> | null {
return typeof value === 'object' && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: null
}
export function claudeText(value: unknown): string | null {
return typeof value === 'string' && value.length > 0 ? value : null
}
export function readClaudeMessageEnvelope(
frame: Record<string, unknown>
): ClaudeMessageEnvelope | null {
if (frame.type !== 'assistant' && frame.type !== 'user') {
return null
}
const message = claudeRecord(frame.message)
const sessionId = claudeText(frame.session_id)
const uuid = claudeText(frame.uuid)
const role = message?.role
return sessionId && uuid && (role === 'assistant' || role === 'user')
? {
sessionId,
uuid,
role,
content: Array.isArray(message?.content) ? message.content : []
}
: null
}
export function claudeMessageIdentity(
envelope: Pick<ClaudeMessageEnvelope, 'sessionId' | 'uuid'>
): AgentJournalItemIdentity {
return { provider: 'claude', sessionId: envelope.sessionId, uuid: envelope.uuid }
}
function messageBlocks(envelope: ClaudeMessageEnvelope): NativeChatBlock[] {
const blocks: NativeChatBlock[] = []
for (const value of envelope.content) {
const part = claudeRecord(value)
const text = claudeText(part?.text)
if (part?.type === 'text' && text) {
blocks.push({ type: 'text', text })
continue
}
const source = claudeRecord(part?.source)
const url = claudeText(source?.url)
if (part?.type === 'image' && source?.type === 'url' && url) {
blocks.push({ type: 'image-ref', url })
}
}
return blocks
}
export function claudeMessageBody(envelope: ClaudeMessageEnvelope): AgentJournalMessageItem | null {
const blocks = messageBlocks(envelope)
return blocks.length > 0 ? { kind: 'message', role: envelope.role, blocks } : null
}
export function claudeToolUses(envelope: ClaudeMessageEnvelope): ClaudeToolUse[] {
return envelope.content.flatMap((value) => {
const part = claudeRecord(value)
const id = claudeText(part?.id)
const name = claudeText(part?.name)
return part?.type === 'tool_use' && id && name ? [{ id, name, input: part.input ?? null }] : []
})
}
function resultText(value: unknown): string {
if (typeof value === 'string') {
return value
}
if (!Array.isArray(value)) {
return value === undefined ? '' : JSON.stringify(value)
}
return value
.flatMap((entry) => {
if (typeof entry === 'string') {
return [entry]
}
const part = claudeRecord(entry)
return part?.type === 'text' && typeof part.text === 'string' ? [part.text] : []
})
.join('\n')
}
export function claudeToolResults(envelope: ClaudeMessageEnvelope): ClaudeToolResult[] {
return envelope.content.flatMap((value) => {
const part = claudeRecord(value)
const toolUseId = claudeText(part?.tool_use_id)
return part?.type === 'tool_result' && toolUseId
? [
{
toolUseId,
output: resultText(part.content),
failed: part.is_error === true
}
]
: []
})
}
export function claudeThinkingText(envelope: ClaudeMessageEnvelope): string | null {
const parts = envelope.content.flatMap((value) => {
const part = claudeRecord(value)
const thinking = claudeText(part?.thinking)
return part?.type === 'thinking' && thinking ? [thinking] : []
})
return parts.length > 0 ? parts.join('\n') : null
}
export function claudeToolBody(input: {
tool: ClaudeToolUse
result?: ClaudeToolResult
}): AgentJournalItemBody {
return {
kind: 'tool-call',
name: input.tool.name,
input: input.tool.input,
state: input.result ? (input.result.failed ? 'failed' : 'completed') : 'running',
...(input.result
? { output: boundInlineText(input.result.output, DEFAULT_JOURNAL_PAYLOAD_LIMITS).bounded }
: {})
}
}
export function claudeStreamingMessageBody(text: string): AgentJournalMessageItem {
return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] }
}
export function claudeToolIdentity(sessionId: string, toolUseId: string): AgentJournalItemIdentity {
return { provider: 'orca', clientMessageId: `claude-tool:${sessionId}:${toolUseId}` }
}
export function claudeThinkingIdentity(sessionId: string, uuid: string): AgentJournalItemIdentity {
return { provider: 'orca', clientMessageId: `claude-thinking:${sessionId}:${uuid}` }
}
@@ -0,0 +1,212 @@
import { describe, expect, it, vi } from 'vitest'
import type {
AgentJournalItemBody,
AgentJournalItemIdentity
} from '../../shared/agent-session-journal-types'
import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key'
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
import type { ClaudePendingPrompt } from './claude-structured-prompt-replies'
import { createClaudeJournalTranslator } from './claude-structured-journal-translation'
function sinkState() {
const items: { identity: AgentJournalItemIdentity; body: AgentJournalItemBody }[] = []
const tombstones: AgentJournalItemIdentity[] = []
const sink: StructuredAgentSessionEventSink = {
appendItem: (identity, body) => items.push({ identity, body }),
appendTombstone: (identity) => tombstones.push(identity),
publish: vi.fn()
}
return { sink, items, tombstones }
}
function message(
type: 'assistant' | 'user',
uuid: string,
content: unknown[],
parentToolUseId: string | null = null
) {
return {
type: 'message' as const,
sessionId: 'orca-session',
message: {
type,
uuid,
session_id: 'claude-session',
parent_tool_use_id: parentToolUseId,
message: { role: type, content }
}
}
}
describe('Claude structured journal translation', () => {
it('reuses the shared coalescer and finalizes the provider-keyed message row', () => {
const state = sinkState()
let scheduled: (() => void) | null = null
const translator = createClaudeJournalTranslator({
sink: state.sink,
schedule: (run, delay) => {
expect(delay).toBe(60)
scheduled = run
return () => {
scheduled = null
}
}
})
translator.handle({
type: 'message',
sessionId: 'orca-session',
message: {
type: 'stream_event',
uuid: 'assistant-1',
session_id: 'claude-session',
event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'Hel' } }
}
})
translator.handle({
type: 'message',
sessionId: 'orca-session',
message: {
type: 'stream_event',
uuid: 'assistant-1',
session_id: 'claude-session',
event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'lo' } }
}
})
expect(state.items).toEqual([])
const run = scheduled as (() => void) | null
run?.()
expect(state.items.at(-1)).toEqual({
identity: { provider: 'claude', sessionId: 'claude-session', uuid: 'assistant-1' },
body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'Hello' }] }
})
translator.handle(message('assistant', 'assistant-1', [{ type: 'text', text: 'Hello!' }]))
expect(state.items.at(-1)).toEqual({
identity: { provider: 'claude', sessionId: 'claude-session', uuid: 'assistant-1' },
body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'Hello!' }] }
})
})
it('journals turn lifecycle and updates one tool row through its result', () => {
const state = sinkState()
const translator = createClaudeJournalTranslator({ sink: state.sink })
translator.handle(message('user', 'user-1', [{ type: 'text', text: 'List files' }]))
translator.handle(
message('assistant', 'assistant-tool', [
{ type: 'tool_use', id: 'tool-1', name: 'Bash', input: { command: 'ls' } }
])
)
translator.handle(
message(
'user',
'tool-result-1',
[{ type: 'tool_result', tool_use_id: 'tool-1', content: 'a.ts\nb.ts' }],
'tool-1'
)
)
const keyed = new Map(
state.items.map((item) => [agentJournalItemKey(item.identity), item.body])
)
expect(keyed.get('claude:claude-session:user-1')).toMatchObject({
kind: 'message',
role: 'user'
})
expect(keyed.get('orca:claude-tool%3Aclaude-session%3Atool-1')).toMatchObject({
kind: 'tool-call',
name: 'Bash',
state: 'completed',
output: { head: 'a.ts\nb.ts', truncated: false }
})
expect(
state.items.some(
(item) => item.body.kind === 'status' && item.body.turnLifecycle?.turnId === 'user-1'
)
).toBe(true)
translator.handle({
type: 'message',
sessionId: 'orca-session',
message: { type: 'result', session_id: 'claude-session', uuid: 'result-1' }
})
expect(state.tombstones.at(-1)).toMatchObject({
provider: 'legacy',
agent: 'claude',
recordId: 'turn-lifecycle:user-1'
})
})
it('creates addressable approval and multi-question cards and cancels them durably', () => {
const state = sinkState()
const bindings: unknown[][] = []
const translator = createClaudeJournalTranslator({
sink: state.sink,
bindPromptItemId: (...args) => bindings.push(args)
})
const approval = prompt({
requestId: 'permission-1',
promptKey: 'permission-1',
toolUseId: 'tool-1',
toolName: 'Bash',
kind: 'approval',
input: { command: 'git status' },
questionIds: []
})
translator.handle({ type: 'prompt', sessionId: 'orca-session', prompt: approval })
expect(state.items.at(-1)?.body).toMatchObject({
kind: 'approval',
title: 'Allow Bash?',
options: expect.arrayContaining([{ id: 'allow', label: 'Allow' }])
})
expect(bindings[0]).toEqual([
'orca:claude-prompt%3Aorca-session%3Apermission-1',
'permission-1'
])
const questions = prompt({
requestId: 'questions-1',
promptKey: 'questions-1',
toolUseId: 'tool-q',
toolName: 'AskUserQuestion',
kind: 'question',
input: {
questions: [
{ question: 'Library?', options: [{ label: 'Luxon' }] },
{ question: 'Ship?', options: [{ label: 'Yes' }] }
]
},
questionIds: ['Library?', 'Ship?']
})
translator.handle({ type: 'prompt', sessionId: 'orca-session', prompt: questions })
expect(state.items.filter((item) => item.body.kind === 'question')).toHaveLength(2)
expect(bindings.at(-1)).toEqual([
'orca:claude-prompt%3Aorca-session%3Aquestions-1%3AShip%3F',
'questions-1',
'Ship?'
])
translator.handle({
type: 'prompt-cancelled',
sessionId: 'orca-session',
promptKey: 'questions-1'
})
expect(state.tombstones).toHaveLength(2)
})
})
function prompt(
input: Pick<
ClaudePendingPrompt,
'requestId' | 'promptKey' | 'toolUseId' | 'toolName' | 'kind' | 'input' | 'questionIds'
>
): ClaudePendingPrompt {
return {
...input,
suggestions: [],
answers: new Map(),
request: { subtype: 'can_use_tool' }
}
}
@@ -0,0 +1,275 @@
import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types'
import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key'
import {
createAgentSessionDeltaCoalescer,
type AgentSessionDeltaCoalescerDeps
} from '../native-chat/agent-session-wire/agent-session-delta-coalescer'
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
import type { ClaudeStructuredSessionEvent } from './claude-structured-session-state'
import {
claudeMessageBody,
claudeMessageIdentity,
claudeRecord,
claudeStreamingMessageBody,
claudeText,
claudeThinkingIdentity,
claudeThinkingText,
claudeToolBody,
claudeToolIdentity,
claudeToolResults,
claudeToolUses,
readClaudeMessageEnvelope,
type ClaudeToolUse
} from './claude-structured-item-translation'
import {
claudeApprovalItem,
claudePromptIdentity,
claudeQuestionItems
} from './claude-structured-prompt-items'
import type { ClaudePromptRegistry } from './claude-structured-prompt-replies'
export type ClaudeJournalTranslatorDeps = {
sink: StructuredAgentSessionEventSink
bindPromptItemId?: (journalItemId: string, promptKey: string, questionId?: string) => void
coalesceMs?: number
schedule?: AgentSessionDeltaCoalescerDeps['schedule']
}
export type ClaudeJournalTranslator = {
handle: (event: ClaudeStructuredSessionEvent) => void
flush: () => void
dispose: () => void
}
export function createClaudeSessionJournalTranslator(
sink: StructuredAgentSessionEventSink | undefined,
prompts: ClaudePromptRegistry
): ClaudeJournalTranslator | null {
return sink
? createClaudeJournalTranslator({
sink,
bindPromptItemId: (itemId, promptKey, questionId) =>
prompts.bindJournalItemId(itemId, promptKey, questionId)
})
: null
}
function lifecycleIdentity(sessionId: string, turnId: string): AgentJournalItemIdentity {
return {
provider: 'legacy',
agent: 'claude',
sessionId,
recordId: `turn-lifecycle:${turnId}`
}
}
function streamDelta(message: Record<string, unknown>): string | null {
if (message.type !== 'stream_event') {
return null
}
const event = claudeRecord(message.event)
if (event?.type === 'content_block_start') {
return claudeText(claudeRecord(event.content_block)?.text)
}
if (event?.type !== 'content_block_delta') {
return null
}
const delta = claudeRecord(event.delta)
return delta?.type === 'text_delta' ? claudeText(delta.text) : null
}
export function createClaudeJournalTranslator(
deps: ClaudeJournalTranslatorDeps
): ClaudeJournalTranslator {
const tools = new Map<string, ClaudeToolUse>()
const promptItems = new Map<string, AgentJournalItemIdentity[]>()
const streamIdentities = new Map<string, AgentJournalItemIdentity>()
const latestStreamText = new Map<string, string>()
const checkpointLengths = new Map<string, number>()
let currentTurn: { sessionId: string; turnId: string } | null = null
const publishLifecycle = (sessionId: string, turnId: string, running: boolean): void => {
const identity = lifecycleIdentity(sessionId, turnId)
if (running) {
deps.sink.appendItem(identity, {
kind: 'status',
text: 'Claude is working…',
turnLifecycle: { turnId, state: 'running' }
})
} else {
deps.sink.appendTombstone(identity)
}
deps.sink.publish()
}
const persistStream = (key: string, text: string, force: boolean): void => {
latestStreamText.set(key, text)
const checkpointLength = checkpointLengths.get(key) ?? 0
const nextLength = Math.max(checkpointLength + 32, Math.ceil(checkpointLength * 1.125))
if (!force && checkpointLength > 0 && text.length < nextLength) {
return
}
const identity = streamIdentities.get(key)
if (!identity) {
return
}
checkpointLengths.set(key, text.length)
deps.sink.appendItem(identity, claudeStreamingMessageBody(text))
deps.sink.publish()
}
const coalescer = createAgentSessionDeltaCoalescer({
windowMs: deps.coalesceMs,
schedule: deps.schedule,
emit: (key, text) => persistStream(key, text, false)
})
const flushStreams = (): void => {
coalescer.flushAll()
for (const [key, text] of latestStreamText) {
if (checkpointLengths.get(key) !== text.length) {
persistStream(key, text, true)
}
}
}
const handleStream = (message: Record<string, unknown>): boolean => {
const delta = streamDelta(message)
const sessionId = claudeText(message.session_id)
const uuid = claudeText(message.uuid)
if (!delta || !sessionId || !uuid) {
return false
}
const key = agentJournalItemKey({ provider: 'claude', sessionId, uuid })
streamIdentities.set(key, { provider: 'claude', sessionId, uuid })
coalescer.append(key, delta)
return true
}
const handleMessage = (message: Record<string, unknown>): void => {
const envelope = readClaudeMessageEnvelope(message)
if (!envelope) {
return
}
const identity = claudeMessageIdentity(envelope)
const key = agentJournalItemKey(identity)
coalescer.forget(key)
latestStreamText.delete(key)
checkpointLengths.delete(key)
streamIdentities.delete(key)
let changed = false
const body = claudeMessageBody(envelope)
if (body) {
deps.sink.appendItem(identity, body)
changed = true
}
for (const tool of claudeToolUses(envelope)) {
tools.set(tool.id, tool)
deps.sink.appendItem(
claudeToolIdentity(envelope.sessionId, tool.id),
claudeToolBody({ tool })
)
changed = true
}
for (const result of claudeToolResults(envelope)) {
const tool = tools.get(result.toolUseId) ?? {
id: result.toolUseId,
name: 'tool',
input: null
}
deps.sink.appendItem(
claudeToolIdentity(envelope.sessionId, result.toolUseId),
claudeToolBody({ tool, result })
)
changed = true
}
const thinking = claudeThinkingText(envelope)
if (thinking) {
deps.sink.appendItem(claudeThinkingIdentity(envelope.sessionId, envelope.uuid), {
kind: 'status',
text: thinking
})
changed = true
}
if (envelope.role === 'user' && body && message.parent_tool_use_id === null) {
if (currentTurn) {
publishLifecycle(currentTurn.sessionId, currentTurn.turnId, false)
}
currentTurn = { sessionId: envelope.sessionId, turnId: envelope.uuid }
publishLifecycle(envelope.sessionId, envelope.uuid, true)
}
if (changed) {
deps.sink.publish()
}
}
const handlePrompt = (event: Extract<ClaudeStructuredSessionEvent, { type: 'prompt' }>): void => {
const identities: AgentJournalItemIdentity[] = []
if (event.prompt.kind === 'question') {
for (const question of claudeQuestionItems({
sessionId: event.sessionId,
prompt: event.prompt
})) {
identities.push(question.identity)
deps.sink.appendItem(question.identity, question.body)
deps.bindPromptItemId?.(
agentJournalItemKey(question.identity),
event.prompt.promptKey,
question.questionId
)
}
} else {
const identity = claudePromptIdentity({
sessionId: event.sessionId,
promptKey: event.prompt.promptKey
})
identities.push(identity)
deps.sink.appendItem(identity, claudeApprovalItem(event.prompt))
deps.bindPromptItemId?.(agentJournalItemKey(identity), event.prompt.promptKey)
}
promptItems.set(event.prompt.promptKey, identities)
deps.sink.publish()
}
return {
handle: (event) => {
if (event.type === 'ended') {
flushStreams()
if (currentTurn) {
publishLifecycle(currentTurn.sessionId, currentTurn.turnId, false)
currentTurn = null
}
return
}
if (event.type === 'message' && handleStream(event.message)) {
return
}
flushStreams()
if (event.type === 'prompt') {
handlePrompt(event)
} else if (event.type === 'prompt-cancelled') {
for (const identity of promptItems.get(event.promptKey) ?? []) {
deps.sink.appendTombstone(identity)
}
promptItems.delete(event.promptKey)
deps.sink.publish()
} else if (event.type === 'message' && event.message.type === 'result') {
if (currentTurn) {
publishLifecycle(currentTurn.sessionId, currentTurn.turnId, false)
currentTurn = null
}
} else if (event.type === 'message') {
handleMessage(event.message)
}
},
flush: flushStreams,
dispose: () => {
coalescer.dispose()
tools.clear()
promptItems.clear()
streamIdentities.clear()
latestStreamText.clear()
checkpointLengths.clear()
}
}
}
@@ -0,0 +1,108 @@
import type {
AgentJournalApprovalItem,
AgentJournalItemIdentity,
AgentJournalPromptOption,
AgentJournalQuestionItem
} from '../../shared/agent-session-journal-types'
import {
boundInlineText,
DEFAULT_JOURNAL_PAYLOAD_LIMITS
} from '../native-chat/agent-session-journal/journal-payload-bounds'
import { claudeRecord, claudeText } from './claude-structured-item-translation'
import {
CLAUDE_APPROVAL_DECISIONS,
encodeClaudeQuestionOptionId,
type ClaudeApprovalDecision,
type ClaudePendingPrompt
} from './claude-structured-prompt-replies'
const APPROVAL_LABELS: Record<ClaudeApprovalDecision, string> = {
allow: 'Allow',
allowForSession: 'Allow for this session',
deny: 'Deny',
cancel: 'Stop'
}
const PENDING = {
state: 'pending',
selectedOptionId: null,
resolvedBy: null,
resolvedAt: null
} as const
export function claudePromptIdentity(input: {
sessionId: string
promptKey: string
questionId?: string
}): AgentJournalItemIdentity {
const suffix = input.questionId ? `:${input.questionId}` : ''
return {
provider: 'orca',
clientMessageId: `claude-prompt:${input.sessionId}:${input.promptKey}${suffix}`
}
}
export function claudeApprovalItem(prompt: ClaudePendingPrompt): AgentJournalApprovalItem {
const serialized = JSON.stringify(prompt.input)
return {
kind: 'approval',
title: `Allow ${prompt.toolName}?`,
detail: serialized ? boundInlineText(serialized, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text : null,
options: CLAUDE_APPROVAL_DECISIONS.map((decision) => ({
id: decision,
label: APPROVAL_LABELS[decision]
})),
resolution: { ...PENDING }
}
}
export type ClaudeQuestionItem = {
questionId: string
identity: AgentJournalItemIdentity
body: AgentJournalQuestionItem
}
function questionOptions(
question: Record<string, unknown>,
questionId: string
): AgentJournalPromptOption[] {
if (!Array.isArray(question.options)) {
return []
}
return question.options.flatMap((value) => {
const option = claudeRecord(value)
const label = claudeText(option?.label)
return label ? [{ id: encodeClaudeQuestionOptionId(questionId, label), label }] : []
})
}
export function claudeQuestionItems(input: {
sessionId: string
prompt: ClaudePendingPrompt
}): ClaudeQuestionItem[] {
const questions = Array.isArray(input.prompt.input.questions) ? input.prompt.input.questions : []
return questions.flatMap((value, index) => {
const question = claudeRecord(value)
const questionId = input.prompt.questionIds[index]
const text = claudeText(question?.question) ?? claudeText(question?.header)
return question && questionId && text
? [
{
questionId,
identity: claudePromptIdentity({
sessionId: input.sessionId,
promptKey: input.prompt.promptKey,
questionId
}),
body: {
kind: 'question',
question: text,
options: questionOptions(question, questionId),
freeTextQuestionId: questionId,
resolution: { ...PENDING }
}
}
]
: []
})
}
@@ -24,6 +24,7 @@ import { setClaudeStructuredOption } from './claude-structured-options'
import { ClaudePromptRegistry } from './claude-structured-prompt-replies'
import { readClaudeStructuredSessionOptions } from './claude-structured-session-options'
import { createClaudeSessionPublication } from './claude-structured-session-publication'
import { createClaudeSessionJournalTranslator } from './claude-structured-journal-translation'
import {
cancelClaudeAcquisitionAttempt,
ClaudeAcquisitionRegistry,
@@ -34,7 +35,7 @@ import {
} from './claude-structured-session-state'
import {
closeClaudePublishedSession,
settleClaudeDispatchWaiters
settleClaudeExitedSession
} from './claude-structured-session-close'
export type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution'
@@ -63,6 +64,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
}): Promise<AgentSessionAcquisition> {
const sessionId = input.identity.sessionId
const prompts = new ClaudePromptRegistry()
const translator = createClaudeSessionJournalTranslator(input.events, prompts)
const { previous, attempt } = this.acquisitions.start(sessionId, prompts)
let liveSession: ClaudeSession | null = null
let observedLeafUuid: string | null = null
@@ -183,6 +185,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
fence: input.fence,
resumed: launch.resumed,
prompts,
translator,
events: input.events,
process,
...(this.deps.mintLinkId ? { linkId: this.deps.mintLinkId() } : {}),
@@ -202,6 +205,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
initDeadline.clear()
this.acquisitions.deleteIfCurrent(sessionId, attempt)
if (this.sessions.get(sessionId)?.connection !== attempt.connection) {
translator?.dispose()
prompts.clear()
await attempt.connection?.close()
}
@@ -227,9 +231,8 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
return
}
this.sessions.delete(sessionId)
settleClaudeDispatchWaiters(session)
session.prompts.clear()
this.emit(session, session.events, { type: 'ended', sessionId, reason: error.message })
settleClaudeExitedSession(session)
}
private emit(
@@ -237,6 +240,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
_events: StructuredAgentSessionEventSink | undefined,
event: ClaudeStructuredSessionEvent
): void {
_session?.translator?.handle(event)
this.deps.onEvent?.(event)
}
@@ -7,6 +7,12 @@ export function settleClaudeDispatchWaiters(session: ClaudeSession): void {
}
}
export function settleClaudeExitedSession(session: ClaudeSession): void {
settleClaudeDispatchWaiters(session)
session.prompts.clear()
session.translator?.dispose()
}
export async function closeClaudePublishedSession(input: {
sessions: Map<string, ClaudeSession>
sessionId: string
@@ -53,11 +59,14 @@ export async function closeClaudePublishedSession(input: {
} catch (error) {
persistenceError = error
} finally {
input.onEvent?.({
const ended = {
type: 'ended',
sessionId: input.sessionId,
reason: 'claude session closed'
})
} as const
session.translator?.handle(ended)
input.onEvent?.(ended)
session.translator?.dispose()
await session.connection.close()
}
if (persistenceError) {
@@ -2,6 +2,7 @@ import type { AgentSessionAcquisition } from '../native-chat/agent-session-wire/
import { readClaudeFrameString, type ClaudeInitObservation } from './claude-structured-init-proof'
import { claudeProviderHandleLink } from './claude-structured-owner-identity'
import type { ClaudePromptRegistry } from './claude-structured-prompt-replies'
import type { ClaudeJournalTranslator } from './claude-structured-journal-translation'
import type { ClaudeSession } from './claude-structured-session-state'
export function createClaudeSessionPublication(input: {
@@ -11,6 +12,7 @@ export function createClaudeSessionPublication(input: {
fence: number
resumed: boolean
prompts: ClaudePromptRegistry
translator: ClaudeJournalTranslator | null
events: ClaudeSession['events']
process: AgentSessionAcquisition['process']
linkId?: string
@@ -42,6 +44,7 @@ export function createClaudeSessionPublication(input: {
...(model ? { model } : {}),
...(effort ? { effort } : {})
},
translator: input.translator,
events: input.events
}
}
@@ -5,6 +5,7 @@ import type {
openClaudeStreamJsonConnection
} from './claude-stream-json-connection'
import type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution'
import type { ClaudeJournalTranslator } from './claude-structured-journal-translation'
import type { ClaudePendingPrompt, ClaudePromptRegistry } from './claude-structured-prompt-replies'
export type ClaudeAuthDiagnostic = {
@@ -63,6 +64,7 @@ export type ClaudeSession = {
dispatchWaiters: ClaudeDispatchWaiter[]
options: Map<string, string>
reportedOptions: { model?: string; effort?: string }
translator: ClaudeJournalTranslator | null
events: StructuredAgentSessionEventSink | undefined
}
@@ -0,0 +1,89 @@
import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types'
import type { AgentSessionExecutionLocation } from '../../../shared/agent-session-record'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
type RoutedAgent = 'claude' | 'codex'
export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessionAdapter {
private readonly owners = new Map<string, StructuredAgentSessionAdapter>()
constructor(
private readonly adapters: Record<RoutedAgent, StructuredAgentSessionAdapter>,
private readonly closeAdapters: () => Promise<void>
) {}
supportsCreate = (location: AgentSessionExecutionLocation, agent: string): boolean => {
const adapter = this.adapterForAgent(agent)
return adapter ? (adapter.supportsLocation?.(location) ?? false) : false
}
supportsLocation = (location: AgentSessionExecutionLocation): boolean =>
Object.values(this.adapters).some((adapter) => adapter.supportsLocation?.(location) ?? false)
async acquire(input: Parameters<StructuredAgentSessionAdapter['acquire']>[0]) {
const adapter = this.requireAgent(input.identity)
const acquired = await adapter.acquire(input)
this.owners.set(input.identity.sessionId, adapter)
return acquired
}
async releaseAcquisition(input: { sessionId: string }): Promise<void> {
const adapter = this.owners.get(input.sessionId)
if (adapter) {
await adapter.releaseAcquisition?.(input)
this.owners.delete(input.sessionId)
return
}
for (const candidate of Object.values(this.adapters)) {
await candidate.releaseAcquisition?.(input)
}
}
dispatch: StructuredAgentSessionAdapter['dispatch'] = (input) =>
this.owner(input.sessionId).dispatch(input)
cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) =>
this.owner(input.sessionId).cancelTurn(input)
answerPrompt: StructuredAgentSessionAdapter['answerPrompt'] = (input) =>
this.owner(input.sessionId).answerPrompt(input)
setOption: StructuredAgentSessionAdapter['setOption'] = (input) =>
this.owner(input.sessionId).setOption(input)
readOptions = (input: { sessionId: string; fence: number }) => {
const reader = this.owner(input.sessionId).readOptions
if (!reader) {
throw new Error(`structured session ${input.sessionId} does not report options`)
}
return reader(input)
}
historyFilePath = (input: { identity: AgentSessionJournalIdentity }) =>
this.requireAgent(input.identity).historyFilePath?.(input) ?? Promise.resolve(null)
async closeAll(): Promise<void> {
this.owners.clear()
await this.closeAdapters()
}
private owner(sessionId: string): StructuredAgentSessionAdapter {
const adapter = this.owners.get(sessionId)
if (!adapter) {
throw new Error(`no live structured adapter owns ${sessionId}`)
}
return adapter
}
private requireAgent(identity: AgentSessionJournalIdentity): StructuredAgentSessionAdapter {
const adapter = this.adapterForAgent(identity.agent)
if (!adapter) {
throw new Error(`structured sessions do not support ${identity.agent}`)
}
return adapter
}
private adapterForAgent(agent: string): StructuredAgentSessionAdapter | null {
return agent === 'claude' || agent === 'codex' ? this.adapters[agent] : null
}
}
@@ -34,6 +34,8 @@ export type AgentSessionDispatchOutcome =
| { state: 'unknown'; reason: string }
export type StructuredAgentSessionAdapter = {
/** Provider-aware capability check for hosts that route more than one adapter. */
supportsCreate?(location: AgentSessionExecutionLocation, agent: string): boolean
/** Provider/runtime support, kept here so remote enablement changes adapter data, not UI logic. */
supportsLocation?(location: AgentSessionExecutionLocation): boolean
/** Makes the reservation real. Called once per reservation, with the spawn
@@ -59,6 +59,7 @@ import {
} from './structured-agent-session-subscribers'
import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue'
import { restoreStructuredAgentSessionsOnRestart } from './structured-agent-session-restart-restore'
import * as providerRouting from './structured-agent-session-provider-routing'
export type StructuredAgentSessionCaller = {
/** Stable per-client identity; scopes the operation ledger and records who
@@ -109,18 +110,14 @@ export class StructuredAgentSessionHost {
}
supportsCreate(location: AgentSessionExecutionLocation, agent: string): boolean {
return agent === 'codex' && (this.deps.adapter.supportsLocation?.(location) ?? false)
return providerRouting.adapterSupportsAgentSessionCreate(this.deps.adapter, location, agent)
}
listSessionTabs(): {
sessionId: string
workspaceId: string
agent: 'codex'
}[] {
listSessionTabs(): providerRouting.StructuredAgentSessionTab[] {
return [...this.sessions.entries()].map(([sessionId, session]) => ({
sessionId,
workspaceId: session.params.location.workspaceId,
agent: 'codex' as const
agent: providerRouting.structuredAgentSessionTabAgent(session.params.agent)
}))
}
@@ -128,7 +125,11 @@ export class StructuredAgentSessionHost {
await restoreStructuredAgentSessionsOnRestart({
store: this.deps.store,
journalRoot: this.deps.journalRoot,
records: this.deps.store.listRecords().filter((record) => record.provider === 'codex'),
records: this.deps.store
.listRecords()
.filter((record) =>
providerRouting.adapterSupportsAgentSessionRecord(this.deps.adapter, record)
),
reconcile: this.reconcileLeases,
operationId: () =>
`${Math.trunc(this.now()).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`,
@@ -0,0 +1,35 @@
import type {
AgentSessionExecutionLocation,
AgentSessionRecord
} from '../../../shared/agent-session-record'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
export type StructuredAgentSessionTab = {
sessionId: string
workspaceId: string
agent: 'claude' | 'codex'
}
export function adapterSupportsAgentSessionCreate(
adapter: StructuredAgentSessionAdapter,
location: AgentSessionExecutionLocation,
agent: string
): boolean {
return (
adapter.supportsCreate?.(location, agent) ??
(agent === 'codex' && (adapter.supportsLocation?.(location) ?? false))
)
}
export function adapterSupportsAgentSessionRecord(
adapter: StructuredAgentSessionAdapter,
record: AgentSessionRecord
): boolean {
return adapter.supportsCreate
? adapter.supportsCreate(record.location, record.provider)
: record.provider === 'codex'
}
export function structuredAgentSessionTabAgent(agent: string): 'claude' | 'codex' {
return agent === 'claude' ? 'claude' : 'codex'
}
@@ -25,7 +25,7 @@ export async function restoreStructuredAgentSessionRead(
sessionId: string
): Promise<RestoredStructuredAgentSessionRead | null> {
const record = store.getRecord(sessionId)
if (!record || record.provider !== 'codex') {
if (!record) {
return null
}
const params = attachParamsForRecord(record, {
@@ -59,8 +59,8 @@ export function attachParamsForRecord(
payloadFingerprint: ''
},
location: record.location,
provider: 'codex',
agent: 'codex',
provider: record.provider,
agent: record.provider,
accountHome: record.accountHome,
runtimeKind: record.lease.runtimeKind
}
@@ -0,0 +1,46 @@
import { classifyObservedAgentSessionSpawnToken } from '../../shared/agent-session-lease-adjudication'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import type { AgentSessionStoreState } from './agent-session-record-store-file'
export function isVerifiable(
state: AgentSessionStoreState,
keyId: string,
now: number,
retentionMs: number
): boolean {
const retired = state.retiredClaimKeys.find((entry) => entry.keyId === keyId)
return !retired || now - retired.retiredAt <= retentionMs
}
export function markConflicted(record: AgentSessionRecord, now: number): AgentSessionRecord {
return {
...record,
updatedAt: now,
// A conflicted key must remain conflicted after its observing process exits.
lease: { ...record.lease, claimStatus: 'conflicted', handoffStage: 'manual-recovery' }
}
}
export function retire(
state: AgentSessionStoreState,
keyId: string,
now: number,
retentionMs: number
): void {
if (!state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) {
state.retiredClaimKeys.push({ keyId, retiredAt: now })
}
state.retiredClaimKeys = state.retiredClaimKeys.filter(
(entry) => now - entry.retiredAt <= retentionMs
)
}
export function listOrphanSpawnTokens(
records: readonly AgentSessionRecord[],
observedTokens: readonly string[]
): string[] {
const leases = records.map((record) => record.lease)
return observedTokens.filter(
(spawnToken) => classifyObservedAgentSessionSpawnToken({ spawnToken, leases }) === 'orphan'
)
}
@@ -0,0 +1,48 @@
import { describe, expect, it } from 'vitest'
import {
agentSessionLeaseFixture,
agentSessionRecordFixture
} from '../../shared/agent-session-record.test-fixture'
import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
import { recordAgentSessionProviderHandle } from './agent-session-provider-handle-transition'
function resumedLink(fence: number): AgentSessionProviderHandleLink {
return {
linkId: 'link-2',
handle: { provider: 'claude', sessionId: 'session-alpha-1', leafUuid: 'leaf-2' },
origin: 'resumed',
mintedAtFence: fence,
observedAt: 4_000
}
}
describe('recordAgentSessionProviderHandle', () => {
it('advances a live Claude chain head and its proof', () => {
const record = agentSessionRecordFixture()
const next = recordAgentSessionProviderHandle({
record,
fence: record.lease.runtimeFence,
link: resumedLink(record.lease.runtimeFence),
now: 4_000
})
expect(next.providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'leaf-2' })
expect(next.lease.provenHandleLinkId).toBe('link-2')
})
it('records a leaf during proof without granting ownership', () => {
const lease = agentSessionLeaseFixture({
runtimeFence: 8,
claimStatus: 'reserved',
handoffStage: 'new-owner-proving',
provenHandleLinkId: null
})
const next = recordAgentSessionProviderHandle({
record: agentSessionRecordFixture(lease),
fence: lease.runtimeFence,
link: resumedLink(lease.runtimeFence),
now: 4_000
})
expect(next.providerHandleChain.at(-1)?.handle).toMatchObject({ leafUuid: 'leaf-2' })
expect(next.lease).toMatchObject({ claimStatus: 'reserved', provenHandleLinkId: null })
})
})
@@ -0,0 +1,41 @@
import {
appendAgentSessionProviderHandleLink,
type AgentSessionProviderHandleLink
} from '../../shared/agent-session-provider-handle'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
export function recordAgentSessionProviderHandle(args: {
record: AgentSessionRecord
fence: number
link: AgentSessionProviderHandleLink
now: number
}): AgentSessionRecord {
const { record } = args
if (record.lease.runtimeFence !== args.fence) {
throw new Error('agent_session_stale_fence')
}
if (args.link.handle.provider !== record.provider || args.link.mintedAtFence !== args.fence) {
throw new Error('agent_session_provider_handle_invalid')
}
if (record.lease.claimStatus !== 'live' && record.lease.handoffStage !== 'new-owner-proving') {
throw new Error('agent_session_ownership_unknown')
}
const providerHandleChain = appendAgentSessionProviderHandleLink(
record.providerHandleChain,
args.link
)
const head = providerHandleChain.at(-1)
if (!head) {
throw new Error('agent_session_provider_handle_invalid')
}
return {
...record,
providerHandleChain,
lease: {
...record.lease,
...(record.lease.claimStatus === 'live' ? { provenHandleLinkId: head.linkId } : {}),
lastRenewedAt: args.now
},
updatedAt: args.now
}
}
+19 -22
View File
@@ -21,7 +21,6 @@ import {
type AgentSessionOperationAdmission
} from './agent-session-operation-admission'
import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
import { classifyObservedAgentSessionSpawnToken } from '../../shared/agent-session-lease-adjudication'
import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
import {
agentSessionScopeKey,
@@ -41,6 +40,7 @@ import {
setAgentSessionHandoffStage,
setAgentSessionJournalCheckpoint
} from './agent-session-lease-transitions'
import { recordAgentSessionProviderHandle } from './agent-session-provider-handle-transition'
import {
applyAgentSessionReservation,
evaluateAgentSessionReserveOperation,
@@ -58,6 +58,7 @@ import {
AgentSessionStoreTransactionQueue,
markAgentSessionStoreLeasesUnreconciled
} from './agent-session-store-transaction-queue'
import * as claimKeyState from './agent-session-claim-key-state'
export type {
AgentSessionReserveRequest,
@@ -131,16 +132,12 @@ export class AgentSessionRecordStore {
}
isClaimKeyVerifiable(keyId: string, now: number): boolean {
const retired = this.state.retiredClaimKeys.find((entry) => entry.keyId === keyId)
return !retired || now - retired.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS
return claimKeyState.isVerifiable(this.state, keyId, now, AGENT_SESSION_CLAIM_KEY_RETENTION_MS)
}
/** Spawn tokens observed on the host with no matching lease. Stop them; never adopt them. */
listOrphanSpawnTokens(observedTokens: readonly string[]): string[] {
const leases = this.listRecords().map((record) => record.lease)
return observedTokens.filter(
(spawnToken) => classifyObservedAgentSessionSpawnToken({ spawnToken, leases }) === 'orphan'
)
return claimKeyState.listOrphanSpawnTokens(this.listRecords(), observedTokens)
}
/**
@@ -200,6 +197,17 @@ export class AgentSessionRecordStore {
)
}
async recordProviderHandle(args: {
sessionId: string
fence: number
link: AgentSessionProviderHandleLink
now: number
}): Promise<AgentSessionRecord> {
return this.mutate(args.sessionId, (record) =>
recordAgentSessionProviderHandle({ ...args, record })
)
}
async renewLease(args: {
sessionId: string
fence: number
@@ -307,24 +315,13 @@ export class AgentSessionRecordStore {
}
async markClaimConflicted(sessionId: string, now: number): Promise<AgentSessionRecord> {
return this.mutate(sessionId, (record) => ({
...record,
updatedAt: now,
// Why: a conflicted key must stay conflicted across a restart; it cannot resolve to free
// merely because the process that observed the conflict is gone.
lease: { ...record.lease, claimStatus: 'conflicted', handoffStage: 'manual-recovery' }
}))
return this.mutate(sessionId, (record) => claimKeyState.markConflicted(record, now))
}
async retireClaimKey(keyId: string, now: number): Promise<void> {
await this.transact(() => {
if (!this.state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) {
this.state.retiredClaimKeys.push({ keyId, retiredAt: now })
}
this.state.retiredClaimKeys = this.state.retiredClaimKeys.filter(
(entry) => now - entry.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS
)
})
await this.transact(() =>
claimKeyState.retire(this.state, keyId, now, AGENT_SESSION_CLAIM_KEY_RETENTION_MS)
)
}
private async mutate(
@@ -0,0 +1,358 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope'
import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types'
import type { AgentSessionSubscribeEvent } from '../../shared/agent-session-wire'
import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../shared/protocol-version'
import type {
ClaudeStreamJsonConnection,
ClaudeStreamJsonConnectionHandlers,
ClaudeStreamJsonLaunch,
openClaudeStreamJsonConnection
} from '../claude/claude-stream-json-connection'
import { claudeSessionIdForOrcaSession } from '../claude/claude-structured-launch-resolution'
import { attachFingerprintFields } from '../native-chat/agent-session-wire/structured-agent-session-attach'
import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry'
import type { OrcaRuntimeService } from './orca-runtime'
import type { RpcRequest, RpcResponse } from './rpc/core'
import { RpcDispatcher } from './rpc/dispatcher'
import { STRUCTURED_AGENT_SESSION_METHODS } from './rpc/methods/structured-agent-session'
import {
ensureStructuredAgentSessionHost,
stopStructuredAgentSessionRuntime
} from './structured-agent-session-runtime'
const SESSION = 'claude-integration-1'
const PROVIDER_SESSION = claudeSessionIdForOrcaSession(SESSION)
const WORKSPACE = 'workspace-claude'
const CLIENT = {
clientKind: 'mobile' as const,
clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY]
}
type FakeClaudeConnection = Omit<ClaudeStreamJsonConnection, 'closed'> & {
closed: boolean
launch: ClaudeStreamJsonLaunch
handlers: ClaudeStreamJsonConnectionHandlers
calls: { subtype: string; params?: Record<string, unknown> }[]
sent: Record<string, unknown>[]
replies: { requestId: string; response: unknown }[]
}
function fakeClaude() {
const connections: FakeClaudeConnection[] = []
const openConnection = (async (launch, handlers = {}) => {
const connection: FakeClaudeConnection = {
launch,
handlers,
calls: [],
sent: [],
replies: [],
pid: 4321 + connections.length,
closed: false,
request: async (subtype, params) => {
connection.calls.push({ subtype, params })
if (subtype === 'initialize') {
handlers.onMessage?.({
type: 'system',
subtype: 'init',
session_id: PROVIDER_SESSION,
...(connections.length === 0 ? { uuid: 'init-leaf' } : {}),
model: 'claude-sonnet-5',
apiKeySource: 'none'
})
return { models: [{ value: 'sonnet', displayName: 'Sonnet' }] }
}
return subtype === 'get_settings' ? { env: {} } : {}
},
send: async (message) => {
connection.sent.push(message)
if (message.type === 'user') {
handlers.onMessage?.({ ...message, uuid: 'user-1' })
}
},
respond: async (requestId, response) => {
connection.replies.push({ requestId, response })
},
respondWithError: async () => {},
close: async () => {
connection.closed = true
}
}
connections.push(connection)
return connection
}) as typeof openClaudeStreamJsonConnection
const live = (): FakeClaudeConnection => {
const connection = connections.at(-1)
if (!connection) {
throw new Error('no Claude connection')
}
return connection
}
return { connections, openConnection, live }
}
let operations = 0
function operationId(): string {
operations += 1
return `${Date.now()}-${operations.toString(16).padStart(32, '0')}`
}
function envelope(method: string, fields: Record<string, unknown>, fence: number | null) {
return {
sessionId: SESSION,
clientOperationId: operationId(),
expectedRuntimeFence: fence,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method,
sessionId: SESSION,
fields
})
}
}
function createIntentParams() {
const worktree = `id:${WORKSPACE}`
const fields = { worktree, agent: 'claude' }
return { envelope: envelope('agentSession.create', fields, null), ...fields }
}
function ensureParams(fence: number) {
const params = {
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: WORKSPACE,
workspaceKind: 'git-worktree' as const
},
provider: 'claude' as const,
agent: 'claude',
accountHome: { variable: 'CLAUDE_CONFIG_DIR' as const, path: '/accounts/claude' },
runtimeKind: 'native' as const,
providerHandle: {
kind: 'claude' as const,
sessionId: PROVIDER_SESSION,
leafUuid: 'assistant-leaf'
}
}
const base = {
sessionId: SESSION,
clientOperationId: operationId(),
expectedRuntimeFence: fence,
payloadFingerprint: ''
}
return {
...params,
envelope: {
...base,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method: 'agentSession.attach',
sessionId: SESSION,
fields: attachFingerprintFields({ ...params, envelope: base } as never)
})
}
}
}
let claude: ReturnType<typeof fakeClaude>
let root: string
let dispatcher: RpcDispatcher
let cleanups: Map<string, () => void>
async function call(method: string, params: unknown): Promise<RpcResponse> {
const replies: RpcResponse[] = []
const request: RpcRequest = { id: `req-${operations}`, authToken: 'token', method, params }
await dispatcher.dispatchStreaming(request, (raw) => replies.push(JSON.parse(raw)), CLIENT)
if (!replies[0]) {
throw new Error(`no reply for ${method}`)
}
return replies[0]
}
async function ok<T>(method: string, params: unknown): Promise<T> {
const response = await call(method, params)
expect(response, JSON.stringify(response)).toMatchObject({ ok: true })
const result = (response as { result: { ok: boolean; value?: T } }).result
expect(result).toMatchObject({ ok: true })
return result.value as T
}
async function subscribe(): Promise<AgentSessionSubscribeEvent[]> {
const frames: AgentSessionSubscribeEvent[] = []
await dispatcher.dispatchStreaming(
{
id: 'subscribe-1',
authToken: 'token',
method: 'agentSession.subscribe',
params: { sessionId: SESSION }
},
(raw) => {
const response = JSON.parse(raw) as { ok: boolean; result?: AgentSessionSubscribeEvent }
if (response.ok && response.result) {
frames.push(response.result)
}
},
CLIENT
)
return frames
}
function itemsOf(frames: AgentSessionSubscribeEvent[]): AgentJournalRenderItem[] {
const items = new Map<string, AgentJournalRenderItem>()
for (const frame of frames) {
const rows =
frame.type === 'snapshot' || frame.type === 'reset'
? frame.snapshot.items
: frame.type === 'batch'
? frame.batch.items
: []
for (const row of rows) {
items.set(row.itemId, row)
}
}
return [...items.values()]
}
function textOf(item: AgentJournalRenderItem): string {
return item.body?.kind === 'message'
? item.body.blocks.map((block) => (block.type === 'text' ? block.text : '')).join('')
: ''
}
beforeEach(async () => {
operations = 0
root = await mkdtemp(join(tmpdir(), 'orca-claude-structured-integration-'))
claude = fakeClaude()
cleanups = new Map()
const runtime = {
getRuntimeId: () => 'runtime-1',
getStructuredAgentSessionCreateSupport: async () => ({ supported: true }),
resolveStructuredAgentSessionCreateIntent: async (input: { envelope: unknown }) => ({
...ensureParams(1),
envelope: input.envelope,
providerHandle: undefined
}),
publishStructuredAgentSessionTab: vi.fn(),
ensureStructuredAgentSessionHost: () =>
ensureStructuredAgentSessionHost({
stateDirectory: root,
hostId: 'local',
claimKeyId: 'key-1',
resolveWorkspacePath: async (workspaceId) => `/repos/${workspaceId}`,
resolveCodexCommand: () => '/usr/local/bin/codex',
resolveClaudeCommand: () => '/usr/local/bin/claude',
openClaudeConnection: claude.openConnection
}).then(() => undefined),
registerSubscriptionCleanup: (id: string, dispose: () => void) => cleanups.set(id, dispose),
cleanupSubscription: (id: string) => cleanups.get(id)?.(),
cleanupSubscriptionsByPrefix: () => {}
}
dispatcher = new RpcDispatcher({
runtime: runtime as unknown as OrcaRuntimeService,
methods: STRUCTURED_AGENT_SESSION_METHODS
})
})
afterEach(async () => {
await stopStructuredAgentSessionRuntime()
await rm(root, { recursive: true, force: true })
})
describe('a structured Claude session over agentSession.*', () => {
it('creates, sends, streams, approves, interrupts, and resumes from the chain head', async () => {
const created = await ok<{ fence: number }>('agentSession.create', createIntentParams())
expect(claude.live().launch.args).toContain('--session-id')
expect(claude.live().launch.args).toContain(PROVIDER_SESSION)
const stream = await subscribe()
const body = { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'List files' }] }
const sent = await ok<{
submission: { dispatchState: string; providerItemId: string | null }
}>('agentSession.send', {
envelope: envelope('agentSession.send', { body }, created.fence),
body
})
expect(sent.submission).toMatchObject({
dispatchState: 'accepted',
providerItemId: `claude:${PROVIDER_SESSION}:user-1`
})
claude.live().handlers.onMessage?.({
type: 'stream_event',
session_id: PROVIDER_SESSION,
uuid: 'assistant-leaf',
event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'Two files.' } }
})
claude.live().handlers.onMessage?.({
type: 'assistant',
session_id: PROVIDER_SESSION,
uuid: 'assistant-leaf',
parent_tool_use_id: null,
message: { role: 'assistant', content: [{ type: 'text', text: 'Two files.' }] }
})
await getStructuredAgentSessionHost()?.flushStreamedEvents(SESSION)
expect(itemsOf(stream).find((item) => textOf(item) === 'Two files.')?.itemId).toBe(
`claude:${PROVIDER_SESSION}:assistant-leaf`
)
claude.live().handlers.onControlRequest?.({
type: 'control_request',
request_id: 'permission-1',
request: {
subtype: 'can_use_tool',
tool_name: 'Bash',
tool_use_id: 'tool-1',
input: { command: 'ls' }
}
})
await getStructuredAgentSessionHost()?.flushStreamedEvents(SESSION)
const approval = itemsOf(stream).find((item) => item.body?.kind === 'approval')
expect(approval?.body).toMatchObject({ title: 'Allow Bash?', detail: '{"command":"ls"}' })
await ok('agentSession.respondToApproval', {
envelope: envelope(
'agentSession.respondTo:approval',
{
itemId: approval?.itemId,
expectedRevision: approval?.revision,
optionId: 'allow'
},
created.fence
),
itemId: approval?.itemId,
expectedRevision: approval?.revision,
optionId: 'allow'
})
expect(claude.live().replies.at(-1)).toMatchObject({
requestId: 'permission-1',
response: { behavior: 'allow', toolUseID: 'tool-1' }
})
await expect(
ok('agentSession.cancel', {
envelope: envelope('agentSession.cancel', { turnId: 'user-1' }, created.fence),
turnId: 'user-1'
})
).resolves.toMatchObject({ turnId: 'user-1', cancelled: true })
expect(claude.live().calls.at(-1)).toMatchObject({ subtype: 'interrupt' })
const old = claude.live()
const resumed = await ok<{ fence: number }>('agentSession.ensure', ensureParams(created.fence))
expect(resumed.fence).toBe(created.fence + 1)
expect(old.closed).toBe(true)
expect(claude.live().launch.args.slice(-2)).toEqual(['--resume', PROVIDER_SESSION])
const host = getStructuredAgentSessionHost() as unknown as {
deps: { store: { getRecord: (sessionId: string) => { providerHandleChain: unknown[] } } }
}
expect(host.deps.store.getRecord(SESSION).providerHandleChain.at(-1)).toMatchObject({
handle: {
provider: 'claude',
sessionId: PROVIDER_SESSION,
leafUuid: 'assistant-leaf'
},
origin: 'resumed'
})
})
})
@@ -52,4 +52,56 @@ describe('structured agent-session create intent', () => {
path: '/accounts/selected/home'
})
})
it('pins the selected Claude config directory without changing its auth environment', async () => {
const runtime = new OrcaRuntimeService({
getSettings: () => ({
agentDefaultEnv: {
claude: {
CLAUDE_CONFIG_DIR: '/accounts/claude/home',
ANTHROPIC_AUTH_TOKEN: 'inherited-by-launch'
}
}
})
} as never)
vi.spyOn(runtime, 'getStructuredAgentSessionCreateSupport').mockResolvedValue({
supported: true
})
const internal = runtime as unknown as {
resolveStructuredAgentSessionLocation: (selector: string) => Promise<{
executionHostId: string
wslDistro: null
workspaceId: string
workspaceKind: 'folder'
}>
resolveRuntimeFileTarget: (selector: string) => Promise<{
worktree: { path: string }
}>
}
internal.resolveStructuredAgentSessionLocation = vi.fn(async () => ({
executionHostId: 'local',
wslDistro: null,
workspaceId: 'folder-1',
workspaceKind: 'folder' as const
}))
internal.resolveRuntimeFileTarget = vi.fn(async () => ({
worktree: { path: '/folders/one' }
}))
const intent = await runtime.resolveStructuredAgentSessionCreateIntent({
envelope: { sessionId: 'session-claude', clientOperationId: 'operation-claude' },
worktree: 'id:folder-1',
agent: 'claude'
})
expect(intent).toMatchObject({
provider: 'claude',
agent: 'claude',
location: { workspaceKind: 'folder' },
accountHome: {
variable: 'CLAUDE_CONFIG_DIR',
path: '/accounts/claude/home'
}
})
})
})
+23 -18
View File
@@ -9201,11 +9201,8 @@ export class OrcaRuntimeService {
async getStructuredAgentSessionCreateSupport(
worktreeSelector: string,
agent: 'codex'
agent: 'claude' | 'codex'
): Promise<{ supported: boolean; reason?: 'agent' | 'remote' | 'wsl' }> {
if (agent !== 'codex') {
return { supported: false, reason: 'agent' }
}
const location = await this.resolveStructuredAgentSessionLocation(worktreeSelector)
await this.ensureStructuredAgentSessionHost()
if (getStructuredAgentSessionHost()?.supportsCreate(location, agent)) {
@@ -9246,18 +9243,24 @@ export class OrcaRuntimeService {
async resolveStructuredAgentSessionCreateIntent(input: {
envelope: { sessionId: string; clientOperationId: string }
worktree: string
agent: 'codex'
agent: 'claude' | 'codex'
}): Promise<AgentSessionAttachParams> {
const support = await this.getStructuredAgentSessionCreateSupport(input.worktree, input.agent)
if (!support.supported) {
throw new Error('structured_agent_session_unsupported')
}
const settings = this.requireStore().getSettings()
const launchEnv = resolveTuiAgentLaunchEnv('codex', settings.agentDefaultEnv)
const launchEnv = resolveTuiAgentLaunchEnv(input.agent, settings.agentDefaultEnv)
const location = await this.resolveStructuredAgentSessionLocation(input.worktree)
const workspacePath = (await this.resolveRuntimeFileTarget(input.worktree)).worktree.path
const preparedHome = this.prepareCodexStructuredLaunchFn?.({ workspacePath, launchEnv })
const configuredHome = launchEnv.CODEX_HOME
const preparedHome =
input.agent === 'codex'
? this.prepareCodexStructuredLaunchFn?.({ workspacePath, launchEnv })
: null
const configuredHome =
input.agent === 'codex' ? launchEnv.CODEX_HOME : launchEnv.CLAUDE_CONFIG_DIR
const variable =
input.agent === 'codex' ? ('CODEX_HOME' as const) : ('CLAUDE_CONFIG_DIR' as const)
return {
envelope: {
sessionId: input.envelope.sessionId,
@@ -9266,16 +9269,18 @@ export class OrcaRuntimeService {
payloadFingerprint: ''
},
location,
provider: 'codex',
agent: 'codex',
provider: input.agent,
agent: input.agent,
accountHome: {
variable: 'CODEX_HOME',
variable,
path:
preparedHome?.trim() ||
(this.prepareCodexStructuredLaunchFn
? getSystemCodexHomePath()
: configuredHome?.trim()) ||
getSystemCodexHomePath()
input.agent === 'codex'
? preparedHome?.trim() ||
(this.prepareCodexStructuredLaunchFn
? getSystemCodexHomePath()
: configuredHome?.trim()) ||
getSystemCodexHomePath()
: configuredHome?.trim() || join(homedir(), '.claude')
},
runtimeKind: 'native'
}
@@ -9293,7 +9298,7 @@ export class OrcaRuntimeService {
publishStructuredAgentSessionTab(input: {
workspaceId: string
sessionId: string
agent: 'codex'
agent: 'claude' | 'codex'
activate: boolean
notify?: boolean
}): void {
@@ -9305,7 +9310,7 @@ export class OrcaRuntimeService {
const tab: RuntimeMobileSessionAgentTab = {
type: 'agent-session',
id,
title: 'Codex Chat',
title: input.agent === 'claude' ? 'Claude Chat' : 'Codex Chat',
sessionId: input.sessionId,
agent: input.agent,
isActive: input.activate
@@ -98,7 +98,7 @@ export const CreateIntentParams = z
.object({
envelope: MutationEnvelope,
worktree: Identifier('Invalid worktree selector'),
agent: z.literal('codex')
agent: z.enum(['claude', 'codex'])
})
.strict()
@@ -107,7 +107,7 @@ export const CreateParams = z.union([AttachParams, CreateIntentParams])
export const CreateSupportParams = z
.object({
worktree: Identifier('Invalid worktree selector'),
agent: z.literal('codex')
agent: z.enum(['claude', 'codex'])
})
.strict()
@@ -137,7 +137,7 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [
ctx.runtime.publishStructuredAgentSessionTab({
workspaceId: resolved.location.workspaceId,
sessionId: result.value.sessionId,
agent: 'codex',
agent: resolved.agent === 'claude' ? 'claude' : 'codex',
activate: true
})
}
@@ -10,12 +10,19 @@
import { join } from 'node:path'
import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import { createClaudeStructuredLaunchResolver } from '../claude/claude-structured-launch-resolution'
import {
ClaudeStructuredSessionAdapter,
type ClaudeStructuredSessionAdapterDeps
} from '../claude/claude-structured-session-adapter'
import { claudeProviderHandleLink } from '../claude/claude-structured-owner-identity'
import { createCodexStructuredLaunchResolver } from '../codex/codex-structured-launch-resolution'
import {
CodexStructuredSessionAdapter,
type CodexStructuredSessionAdapterDeps
} from '../codex/codex-structured-session-adapter'
import { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-host'
import { StructuredAgentSessionAdapterRouter } from '../native-chat/agent-session-wire/structured-agent-session-adapter-router'
import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry'
import { AgentSessionRecordStore } from './agent-session-record-store'
import { probeAgentSessionProcessIdentity } from './agent-session-process-identity-probe'
@@ -34,15 +41,17 @@ export type StructuredAgentSessionRuntimeDeps = {
claimKeyId: string
resolveWorkspacePath: (workspaceId: string) => Promise<string>
resolveCodexCommand?: () => string
resolveClaudeCommand?: () => string
/** Transport for the Codex child. Overridden only to drive the whole runtime
* against a scripted app-server; production spawns the real one. */
openCodexConnection?: CodexStructuredSessionAdapterDeps['openConnection']
openClaudeConnection?: ClaudeStructuredSessionAdapterDeps['openConnection']
onError?: (input: { scope: string; error: unknown }) => void
}
type InstalledRuntime = {
host: StructuredAgentSessionHost
adapter: CodexStructuredSessionAdapter
adapter: StructuredAgentSessionAdapterRouter
}
let installing: Promise<InstalledRuntime> | null = null
@@ -58,7 +67,7 @@ export function ensureStructuredAgentSessionHost(
return installing.then((installed) => installed.host)
}
/** Drops the host and reaps every Codex child under it. Runtime teardown and
/** Drops the host and reaps every structured child under it. Runtime teardown and
* test isolation take the same path, so neither can leave a live app-server. */
export async function stopStructuredAgentSessionRuntime(): Promise<void> {
const pending = installing
@@ -83,7 +92,7 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
directory: join(deps.stateDirectory, RECORD_STORE_DIR_NAME),
hostId: deps.hostId
})
const adapter = new CodexStructuredSessionAdapter({
const codex = new CodexStructuredSessionAdapter({
resolveLaunch: createCodexStructuredLaunchResolver({
store,
resolveWorkspacePath: deps.resolveWorkspacePath,
@@ -91,6 +100,36 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
}),
...(deps.openCodexConnection ? { openConnection: deps.openCodexConnection } : {})
})
const claude = new ClaudeStructuredSessionAdapter({
resolveLaunch: createClaudeStructuredLaunchResolver({
store,
resolveWorkspacePath: deps.resolveWorkspacePath,
...(deps.resolveClaudeCommand ? { resolveCommand: deps.resolveClaudeCommand } : {})
}),
...(deps.openClaudeConnection ? { openConnection: deps.openClaudeConnection } : {}),
persistHandle: async (observed) => {
const record = store.getRecord(observed.sessionId)
if (!record || record.provider !== 'claude') {
return
}
const fence = record.lease.runtimeFence
await store.recordProviderHandle({
sessionId: observed.sessionId,
fence,
link: claudeProviderHandleLink({
sessionId: observed.providerSessionId,
leafUuid: observed.leafUuid,
resumed: record.providerHandleChain.length > 0,
fence,
observedAt: Date.now()
}),
now: Date.now()
})
}
})
const adapter = new StructuredAgentSessionAdapterRouter({ claude, codex }, async () =>
Promise.all([claude.closeAll(), codex.closeAll()]).then(() => undefined)
)
const host = new StructuredAgentSessionHost({
store,
adapter,
+1 -1
View File
@@ -256,7 +256,7 @@ export type RuntimeMobileSessionAgentTab = {
id: string
title: string
sessionId: string
agent: 'codex'
agent: 'claude' | 'codex'
color?: string | null
isPinned?: boolean
isActive: boolean