mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
fix(connector): one turn id from the host's stamp everywhere (#33909)
The host's turn-start timestamp is the only turn id the connector uses: snapshots, turn ends and replies all carry it, and the desktop's separate turn cache is gone, so they can no longer disagree. The unused turn-bound cancel was removed (Stop is parked). The hook server reads the main agent state through the same shared accessor.
This commit is contained in:
@@ -12,7 +12,7 @@ import type {
|
||||
AgentStatusObservationOrigin
|
||||
} from '../../../shared/agent-status-observation'
|
||||
import { AGENT_STATUS_2A_CURRENT_PRODUCER_MODE } from '../../../shared/agent-status-legacy-adapter'
|
||||
import type { AgentStatusState, ParsedAgentStatusPayload } from '../../../shared/agent-status-types'
|
||||
import { mainAgentState } from '../../../shared/main-agent-status'
|
||||
import { admitLegacyAgentStatus } from '../../../shared/agent-hook-listener/listener-state'
|
||||
import type { EnrichedAgentHookEventPayload } from './server-types'
|
||||
import { agentTypeToPromptSentAgentKind } from './server-status-identity'
|
||||
@@ -21,10 +21,6 @@ import { AgentHookServerStatusDisposition } from './server-status-disposition'
|
||||
// Evicting remembered evidence makes later replay fall back to its arrival time.
|
||||
const MAX_REMEMBERED_EVIDENCE_OBSERVATIONS = 1024
|
||||
|
||||
function mainAgentState(payload: ParsedAgentStatusPayload): AgentStatusState {
|
||||
return payload.mainAgent?.state ?? payload.state
|
||||
}
|
||||
|
||||
/** Only live main-agent turn boundaries establish a start; replay and child events preserve it. */
|
||||
function resolveTurnStartedAt(
|
||||
previous: EnrichedAgentHookEventPayload | undefined,
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
} from './external-conversation-links'
|
||||
import type { AgentPromptExpectedBinding } from '../runtime/runtime-terminal-contracts'
|
||||
import { sameBinding } from './external-conversation-send'
|
||||
import { workingStampedTurn } from './external-conversation-turn-id'
|
||||
import type { ConversationKey } from './external-conversation-types'
|
||||
|
||||
type ConversationPane = { conversation: ConversationKey; worktreeId: string; paneKey: string }
|
||||
@@ -76,7 +77,10 @@ export const HOST_CLOCK: Pick<ConversationEventsDeps, 'setTimer' | 'now'> = {
|
||||
/** Watch-scoped turn news shares mobile push detection and automation completion verdicts. */
|
||||
export class ConversationEvents {
|
||||
private readonly listeners = new Set<ConversationWatchListener>()
|
||||
private readonly turnStartedAt = new Map<string, number>()
|
||||
// A started turn's id only where the host stamps none; the host's stamp is the id everywhere else.
|
||||
private readonly unstampedTurn = new Map<string, number>()
|
||||
// The stamped turn last announced per pane, only so a start is told once; never read as an id.
|
||||
private readonly announcedTurn = new Map<string, number>()
|
||||
// Prompts our cancel rejected, by pane: Claude's hooks report no end for that turn.
|
||||
private readonly rejected = new Map<
|
||||
string,
|
||||
@@ -99,6 +103,11 @@ export class ConversationEvents {
|
||||
const unsubscribe = this.deps.onAgentPaneChange((change) => {
|
||||
this.settleRejected(change.paneKey)
|
||||
news.observe(change)
|
||||
// A working row's stamp is told once, whether or not news reported its start.
|
||||
const stamp = workingStampedTurn(this.deps.readHookRow(change.paneKey))
|
||||
if (stamp !== null && change.workspace) {
|
||||
this.announceStamp(change.paneKey, change.workspace.worktreeId, stamp)
|
||||
}
|
||||
})
|
||||
this.release = () => {
|
||||
unsubscribe()
|
||||
@@ -150,7 +159,8 @@ export class ConversationEvents {
|
||||
this.release?.()
|
||||
this.release = null
|
||||
this.listeners.clear()
|
||||
this.turnStartedAt.clear()
|
||||
this.unstampedTurn.clear()
|
||||
this.announcedTurn.clear()
|
||||
this.rejected.clear()
|
||||
}
|
||||
|
||||
@@ -210,11 +220,8 @@ export class ConversationEvents {
|
||||
agent: link.agent,
|
||||
working,
|
||||
prompt: this.deps.pendingPrompt(link),
|
||||
// The same id started() and takeTurnId() give this turn.
|
||||
turnId:
|
||||
this.turnStartedAt.get(link.paneKey) ??
|
||||
row?.turnStartedAt ??
|
||||
(row && working ? row.stateStartedAt : null),
|
||||
// Unstamped and unseen: the working phase's start, as before stamps.
|
||||
turnId: this.turnIdOf(link.paneKey, row) ?? (row && working ? row.stateStartedAt : null),
|
||||
activeAt: Math.max(link.createdAt, row?.receivedAt ?? 0)
|
||||
}
|
||||
})
|
||||
@@ -226,9 +233,24 @@ export class ConversationEvents {
|
||||
if (!link) {
|
||||
return
|
||||
}
|
||||
const at = item.row.turnStartedAt ?? item.row.stateStartedAt
|
||||
this.turnStartedAt.set(item.paneKey, at)
|
||||
this.emit({ kind: 'turn-started', ...pane(link, item.worktreeId), at })
|
||||
if (item.row.turnStartedAt !== undefined) {
|
||||
this.announceStamp(item.paneKey, item.worktreeId, item.row.turnStartedAt)
|
||||
return
|
||||
}
|
||||
this.unstampedTurn.set(item.paneKey, item.row.stateStartedAt)
|
||||
this.emit({ kind: 'turn-started', ...pane(link, item.worktreeId), at: item.row.stateStartedAt })
|
||||
}
|
||||
|
||||
/** Told once per stamp; the cloud applies a start only if it is newer than its own anyway. */
|
||||
private announceStamp(paneKey: string, worktreeId: string, at: number): void {
|
||||
const link = this.linkedFor(paneKey)?.link
|
||||
if (!link || this.announcedTurn.get(paneKey) === at) {
|
||||
return
|
||||
}
|
||||
this.announcedTurn.set(paneKey, at)
|
||||
// The host stamps this pane, so a start recorded before its first stamp names nothing.
|
||||
this.unstampedTurn.delete(paneKey)
|
||||
this.emit({ kind: 'turn-started', ...pane(link, worktreeId), at })
|
||||
}
|
||||
|
||||
private told(item: TerminalAgentTurnNewsItem): void {
|
||||
@@ -247,7 +269,7 @@ export class ConversationEvents {
|
||||
const verdict = judgeIdleAgentTurn(
|
||||
{
|
||||
rows: [item.row],
|
||||
sentAt: this.turnStartedAt.get(item.paneKey) ?? 0,
|
||||
sentAt: this.turnIdOf(item.paneKey, item.row) ?? 0,
|
||||
tail: [],
|
||||
readAgentCommands: () => []
|
||||
},
|
||||
@@ -280,13 +302,19 @@ export class ConversationEvents {
|
||||
if (!link) {
|
||||
return null
|
||||
}
|
||||
const turnId = this.turnStartedAt.get(paneKey) ?? this.deps.readHookRow(paneKey)?.turnStartedAt
|
||||
return { key: { source: link.source, externalId: link.externalId }, turnId: turnId ?? null }
|
||||
const turnId = this.turnIdOf(paneKey, this.deps.readHookRow(paneKey))
|
||||
return { key: { source: link.source, externalId: link.externalId }, turnId }
|
||||
}
|
||||
|
||||
/** The turn a pane is in: the host's stamp, else the unstamped start news reported. */
|
||||
private turnIdOf(paneKey: string, row: AgentStatusIpcPayload | null): number | null {
|
||||
return row?.turnStartedAt ?? this.unstampedTurn.get(paneKey) ?? null
|
||||
}
|
||||
|
||||
/** An ending turn's id; an unstamped one's start is forgotten with it, as before. */
|
||||
private takeTurnId(paneKey: string, row: AgentStatusIpcPayload): number | null {
|
||||
const turnId = this.turnStartedAt.get(paneKey) ?? row.turnStartedAt ?? null
|
||||
this.turnStartedAt.delete(paneKey)
|
||||
const turnId = this.turnIdOf(paneKey, row)
|
||||
this.unstampedTurn.delete(paneKey)
|
||||
return turnId
|
||||
}
|
||||
|
||||
|
||||
@@ -205,6 +205,8 @@ describe('ExternalConversationSeam watch', () => {
|
||||
hookRow({
|
||||
state: 'done',
|
||||
stateStartedAt: 300,
|
||||
// The host keeps the turn's stamp on its done row.
|
||||
turnStartedAt: 100,
|
||||
lastAssistantMessage: 'Done.',
|
||||
mainAgent: { state: 'done', stateStartedAt: 300, outcome: 'success' }
|
||||
})
|
||||
@@ -258,12 +260,72 @@ describe('ExternalConversationSeam watch', () => {
|
||||
})
|
||||
hookRows.set(PANE, hookRow({ state: 'working', turnStartedAt: 500 }))
|
||||
paneChange(moved('rest', 'working', 'prompt-submitted'))
|
||||
hookRows.set(PANE, hookRow({ state: 'done', stateStartedAt: 400, sessionBoundary: false }))
|
||||
hookRows.set(
|
||||
PANE,
|
||||
hookRow({ state: 'done', stateStartedAt: 400, turnStartedAt: 500, sessionBoundary: false })
|
||||
)
|
||||
paneChange(moved('working', 'done', 'turn-finished'))
|
||||
paneChange({ ...moved('rest', 'working', 'prompt-submitted'), paneKey: 'tab-9:other' })
|
||||
expect(seen.map((event) => event.kind)).toEqual(['turn-started'])
|
||||
})
|
||||
|
||||
it('names a turn a new prompt opened after an approval by the host’s stamp everywhere', async () => {
|
||||
const { seam, host, hookRows, paneChange } = await linked()
|
||||
const seen: (ConversationSnapshot | ConversationEvent)[] = []
|
||||
seam.watch(host, (event) => seen.push(event))
|
||||
hookRows.set(PANE, hookRow({ state: 'working', turnStartedAt: 100 }))
|
||||
paneChange(moved('rest', 'working', 'prompt-submitted'))
|
||||
hookRows.set(PANE, { ...APPROVAL, turnStartedAt: 100 })
|
||||
paneChange(moved('working', 'needs-you', 'needs-input'))
|
||||
// The approval was dismissed and a new prompt sent: no prompt-submitted news, a new stamp.
|
||||
hookRows.set(PANE, hookRow({ state: 'working', stateStartedAt: 300, turnStartedAt: 300 }))
|
||||
paneChange(moved('needs-you', 'working', null))
|
||||
expect(seen.at(-1)).toMatchObject({ kind: 'turn-started', at: 300 })
|
||||
expect(seam.replyTarget(PANE)).toEqual({ key: KEY, turnId: 300 })
|
||||
seam.watch(host, (event) => seen.push(event))
|
||||
expect(seen.at(-1)).toMatchObject({ kind: 'snapshot', conversations: [{ turnId: 300 }] })
|
||||
hookRows.set(PANE, { ...APPROVAL, stateStartedAt: 400, turnStartedAt: 300 })
|
||||
paneChange(moved('working', 'needs-you', 'needs-input'))
|
||||
hookRows.set(
|
||||
PANE,
|
||||
hookRow({
|
||||
state: 'done',
|
||||
stateStartedAt: 500,
|
||||
turnStartedAt: 300,
|
||||
mainAgent: { state: 'done', stateStartedAt: 500, outcome: 'success' }
|
||||
})
|
||||
)
|
||||
paneChange(moved('needs-you', 'done', 'turn-finished'))
|
||||
expect(seen.at(-1)).toMatchObject({ kind: 'turn-ended', turnId: 300 })
|
||||
})
|
||||
|
||||
it('tells a stamp that lands on a row already working, once, and names the turn by it', async () => {
|
||||
const { seam, host, hookRows, paneChange } = await linked()
|
||||
const seen: ConversationEvent[] = []
|
||||
seam.watch(host, (event) => {
|
||||
if (event.kind !== 'snapshot') {
|
||||
seen.push(event)
|
||||
}
|
||||
})
|
||||
hookRows.set(PANE, hookRow({ state: 'working', stateStartedAt: 100 }))
|
||||
paneChange(moved('rest', 'working', 'prompt-submitted'))
|
||||
const observed: AgentPaneChange = {
|
||||
kind: 'observed',
|
||||
paneKey: PANE,
|
||||
workspace: { worktreeId: WORKTREE, sshHostId: null },
|
||||
phase: 'working',
|
||||
at: 2
|
||||
}
|
||||
hookRows.set(PANE, hookRow({ state: 'working', stateStartedAt: 100, turnStartedAt: 200 }))
|
||||
paneChange(observed)
|
||||
paneChange(observed)
|
||||
expect(seen.map((event) => event.kind === 'turn-started' && event.at)).toEqual([100, 200])
|
||||
expect(seam.replyTarget(PANE)).toEqual({ key: KEY, turnId: 200 })
|
||||
// A session boundary clears the stamp; the start seen before the first stamp doesn't come back.
|
||||
hookRows.set(PANE, hookRow({ state: 'done', stateStartedAt: 300, sessionBoundary: true }))
|
||||
expect(seam.replyTarget(PANE)).toEqual({ key: KEY, turnId: null })
|
||||
})
|
||||
|
||||
it('tells a watcher of a conversation linked while it watches', async () => {
|
||||
const h = harness()
|
||||
const host = {
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
// A conversation's turn is named by the host's stamp on its agent's hook row, wherever that stamp exists.
|
||||
import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload'
|
||||
import { hookRowAgentPhase } from '../../shared/agent-transitions'
|
||||
|
||||
/**
|
||||
* The stamped turn a working row shows, however it got there: prompt news, a new prompt after a dismissed
|
||||
* approval, or a stamp landing on a row already working. Null for other rows.
|
||||
*/
|
||||
export function workingStampedTurn(row: AgentStatusIpcPayload | null): number | null {
|
||||
if (!row || row.structuredHost || hookRowAgentPhase(row) !== 'working') {
|
||||
return null
|
||||
}
|
||||
return row.turnStartedAt ?? null
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { AgentStatusIpcPayload } from '../../shared/agent-status-ipc-payload'
|
||||
import { mainAgentState } from '../../shared/main-agent-status'
|
||||
import { writeFencedAgentKeys } from './terminal-agent-fenced-keys'
|
||||
import type { TerminalAgentPromptAnswerRuntime } from './terminal-agent-prompt-answer'
|
||||
|
||||
@@ -38,7 +39,7 @@ export async function interruptWorkingAgent(
|
||||
throw new Error('agent_not_reachable')
|
||||
}
|
||||
// Escape stops only the main agent's turn; background work alone leaves an idle composer.
|
||||
if ((row.mainAgent?.state ?? row.state) !== 'working') {
|
||||
if (mainAgentState(row) !== 'working') {
|
||||
return null
|
||||
}
|
||||
const { ptyId, processIncarnation } = runtime.getTerminalPromptRequestBinding(handle)
|
||||
|
||||
@@ -14,6 +14,14 @@ export type AgentMainAgentStatus = {
|
||||
stateStartedAt: number
|
||||
}
|
||||
|
||||
/** The main agent's own state where the row carries one; older rows only have the combined state. */
|
||||
export function mainAgentState(row: {
|
||||
state: AgentStatusState
|
||||
mainAgent?: AgentMainAgentStatus
|
||||
}): AgentStatusState {
|
||||
return row.mainAgent?.state ?? row.state
|
||||
}
|
||||
|
||||
export function mainAgentStatusEqual(
|
||||
a: AgentMainAgentStatus | undefined,
|
||||
b: AgentMainAgentStatus | undefined
|
||||
|
||||
Reference in New Issue
Block a user