mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 08:03:37 +00:00
fix(agent-status): keep the turn identity a disconnect clear removed (STA-3524)
An SSH teardown calls clearStatusEntriesForConnection, deleting every row the connection owned. The reconnect replay then re-ingests each cached status with no `previous` entry, so attachStatusTiming mints a fresh stateStartedAt. That timestamp is the completion-notification dedupe key, so the finished turn looks brand new and the task-complete notification fires again on every flap. Keep the deleted row's turn identity until the pane's next status. A replay that re-delivers the same connection/state/agentType/prompt reclaims its stateStartedAt; anything else -- a live event, another host, or a completion that landed while disconnected -- keeps live timing. receivedAt still advances so the row sorts after the clear watermark. Fixes #12747
This commit is contained in:
@@ -0,0 +1,195 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { AgentHookServer } from './server'
|
||||
import { makePaneKey } from '../../shared/stable-pane-id'
|
||||
|
||||
vi.mock('../telemetry/client', () => ({ track: vi.fn() }))
|
||||
vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: vi.fn(() => ({})) }))
|
||||
|
||||
const LEAF = '11111111-1111-4111-8111-111111111111'
|
||||
const PANE = makePaneKey('tab-1', LEAF)
|
||||
const CONNECTION = 'ssh-host-1'
|
||||
const T0 = 1_700_000_000_000
|
||||
const PROMPT = 'ship the release notes'
|
||||
|
||||
function ingest(
|
||||
server: AgentHookServer,
|
||||
options: {
|
||||
state: 'working' | 'done'
|
||||
hookEventName: string
|
||||
isReplay?: boolean
|
||||
connectionId?: string
|
||||
prompt?: string
|
||||
}
|
||||
): void {
|
||||
server.ingestRemote(
|
||||
{
|
||||
paneKey: PANE,
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
source: 'claude',
|
||||
hookEventName: options.hookEventName,
|
||||
...(options.isReplay ? { isReplay: true } : {}),
|
||||
payload: { state: options.state, prompt: options.prompt ?? PROMPT, agentType: 'claude' }
|
||||
},
|
||||
options.connectionId ?? CONNECTION
|
||||
)
|
||||
}
|
||||
|
||||
function paneRow(server: AgentHookServer): {
|
||||
state: string
|
||||
receivedAt: number
|
||||
startedAt: number
|
||||
} {
|
||||
const row = server.getStatusSnapshot()[0]
|
||||
if (!row) {
|
||||
throw new Error('expected a cached status row')
|
||||
}
|
||||
return { state: row.state, receivedAt: row.receivedAt, startedAt: row.stateStartedAt }
|
||||
}
|
||||
|
||||
/** Completion-notification dedupe keys on `state:agentType:stateStartedAt`
|
||||
* (agent-completion-coordinator.ts). Mirror that key so these assertions are about the
|
||||
* thing the renderer actually compares, not an incidental field. */
|
||||
function turnIdentity(server: AgentHookServer): string {
|
||||
const row = server.getStatusSnapshot()[0]
|
||||
if (!row) {
|
||||
throw new Error('expected a cached status row')
|
||||
}
|
||||
return [row.state, row.agentType ?? '', String(row.stateStartedAt)].join(':')
|
||||
}
|
||||
|
||||
function completeATurn(server: AgentHookServer): void {
|
||||
ingest(server, { state: 'working', hookEventName: 'UserPromptSubmit' })
|
||||
vi.setSystemTime(T0 + 5_000)
|
||||
ingest(server, { state: 'done', hookEventName: 'Stop' })
|
||||
}
|
||||
|
||||
describe('reconnect clear + replay keeps one completion turn identity (STA-3524)', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
vi.setSystemTime(T0)
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('reclaims the cleared turn identity when the reconnect replays that same turn', () => {
|
||||
const server = new AgentHookServer()
|
||||
completeATurn(server)
|
||||
const beforeDisconnect = paneRow(server)
|
||||
const notifiedIdentity = turnIdentity(server)
|
||||
|
||||
// A dropped SSH connection clears every row it owned; reconnect replays the host cache.
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
expect(server.getStatusSnapshot()).toEqual([])
|
||||
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, { state: 'done', hookEventName: 'Stop', isReplay: true })
|
||||
|
||||
expect(turnIdentity(server)).toBe(notifiedIdentity)
|
||||
expect(paneRow(server).startedAt).toBe(beforeDisconnect.startedAt)
|
||||
// receivedAt must still advance past the clear watermark or the renderer drops the row.
|
||||
expect(paneRow(server).receivedAt).toBeGreaterThan(beforeDisconnect.receivedAt)
|
||||
})
|
||||
|
||||
it('mints a new turn identity for a completion that landed while disconnected', () => {
|
||||
const server = new AgentHookServer()
|
||||
ingest(server, { state: 'working', hookEventName: 'UserPromptSubmit' })
|
||||
const workingIdentity = turnIdentity(server)
|
||||
|
||||
// The turn ended on the host while the relay was down, so this `done` is news.
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, { state: 'done', hookEventName: 'Stop', isReplay: true })
|
||||
|
||||
expect(turnIdentity(server)).not.toBe(workingIdentity)
|
||||
expect(paneRow(server).startedAt).toBe(T0 + 61_000)
|
||||
})
|
||||
|
||||
it('mints a new turn identity when the replay carries a different prompt', () => {
|
||||
const server = new AgentHookServer()
|
||||
completeATurn(server)
|
||||
const firstTurnStartedAt = paneRow(server).startedAt
|
||||
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, {
|
||||
state: 'done',
|
||||
hookEventName: 'Stop',
|
||||
isReplay: true,
|
||||
prompt: 'a different task the host finished meanwhile'
|
||||
})
|
||||
|
||||
expect(paneRow(server).startedAt).toBe(T0 + 61_000)
|
||||
expect(paneRow(server).startedAt).not.toBe(firstTurnStartedAt)
|
||||
})
|
||||
|
||||
it('does not let another host reclaim the cleared turn', () => {
|
||||
const server = new AgentHookServer()
|
||||
completeATurn(server)
|
||||
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, {
|
||||
state: 'done',
|
||||
hookEventName: 'Stop',
|
||||
isReplay: true,
|
||||
connectionId: 'ssh-host-2'
|
||||
})
|
||||
|
||||
expect(paneRow(server).startedAt).toBe(T0 + 61_000)
|
||||
})
|
||||
|
||||
it('does not reclaim the cleared turn for a live (non-replay) event', () => {
|
||||
const server = new AgentHookServer()
|
||||
completeATurn(server)
|
||||
const firstTurnStartedAt = paneRow(server).startedAt
|
||||
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, { state: 'done', hookEventName: 'Stop' })
|
||||
|
||||
expect(paneRow(server).startedAt).toBe(T0 + 61_000)
|
||||
expect(paneRow(server).startedAt).not.toBe(firstTurnStartedAt)
|
||||
})
|
||||
|
||||
it('does not bleed the cleared turn into a genuinely new turn after the clear', () => {
|
||||
const server = new AgentHookServer()
|
||||
completeATurn(server)
|
||||
const firstTurnStartedAt = paneRow(server).startedAt
|
||||
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
|
||||
// A live prompt lands first; the stale timing is spent and must not reach the next done.
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, { state: 'working', hookEventName: 'UserPromptSubmit' })
|
||||
vi.setSystemTime(T0 + 62_000)
|
||||
ingest(server, { state: 'done', hookEventName: 'Stop' })
|
||||
|
||||
expect(paneRow(server).startedAt).toBe(T0 + 62_000)
|
||||
expect(paneRow(server).startedAt).not.toBe(firstTurnStartedAt)
|
||||
})
|
||||
|
||||
it('does not reclaim a turn across a real pane teardown', () => {
|
||||
const server = new AgentHookServer()
|
||||
completeATurn(server)
|
||||
const firstTurnStartedAt = paneRow(server).startedAt
|
||||
|
||||
vi.setSystemTime(T0 + 60_000)
|
||||
server.clearStatusEntriesForConnection(CONNECTION)
|
||||
server.clearPaneState(PANE)
|
||||
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
ingest(server, { state: 'done', hookEventName: 'Stop', isReplay: true })
|
||||
|
||||
expect(paneRow(server).startedAt).toBe(T0 + 61_000)
|
||||
expect(paneRow(server).startedAt).not.toBe(firstTurnStartedAt)
|
||||
})
|
||||
})
|
||||
@@ -190,6 +190,32 @@ type AgentPromptSentDedupeEntry = {
|
||||
promptInteractionKey?: string
|
||||
}
|
||||
|
||||
/** Turn identity of a row a transient connection clear deleted, kept only until the
|
||||
* pane's next status so a reconnect replay of that same turn can reclaim it. */
|
||||
type ClearedStatusTiming = {
|
||||
connectionId: string
|
||||
state: ParsedAgentStatusPayload['state']
|
||||
agentType: ParsedAgentStatusPayload['agentType']
|
||||
prompt: string
|
||||
stateStartedAt: number
|
||||
}
|
||||
|
||||
/** True when a replay re-delivers the exact turn a disconnect clear removed. A replay
|
||||
* carrying anything else is news this client missed and must keep live timing. */
|
||||
function replayReclaimsClearedTurn(
|
||||
cleared: ClearedStatusTiming | undefined,
|
||||
payload: AgentHookEventPayload
|
||||
): cleared is ClearedStatusTiming {
|
||||
return (
|
||||
cleared !== undefined &&
|
||||
payload.isReplay === true &&
|
||||
cleared.connectionId === payload.connectionId &&
|
||||
cleared.state === payload.payload.state &&
|
||||
cleared.agentType === payload.payload.agentType &&
|
||||
cleared.prompt === payload.payload.prompt
|
||||
)
|
||||
}
|
||||
|
||||
function agentTypeToPromptSentAgentKind(agentType: AgentType | undefined): AgentKind {
|
||||
const normalized = agentType?.trim().toLowerCase()
|
||||
if (!normalized || normalized === 'unknown') {
|
||||
@@ -634,6 +660,9 @@ export class AgentHookServer {
|
||||
private closedAgentStatusTabIds = new Set<string>()
|
||||
private closedAgentStatusPaneKeys = new Set<string>()
|
||||
private connectionTimestampWatermarkById = new Map<string, number>()
|
||||
// Why: a transient disconnect clear deletes the row the reconnect replay is about to
|
||||
// re-deliver; keep its turn identity so the replay is recognized as the same turn (STA-3524).
|
||||
private clearedStatusTimingByPaneKey = new Map<string, ClearedStatusTiming>()
|
||||
// Why: skip disk writes when the JSON exactly matches the last write; guards against re-firing trailing timers when nothing changed.
|
||||
private lastWrittenJson: string | null = null
|
||||
|
||||
@@ -1054,12 +1083,23 @@ export class AgentHookServer {
|
||||
previousPromptInteractionKey: previous.promptInteractionKey,
|
||||
incomingPromptInteractionKey: payload.promptInteractionKey
|
||||
})
|
||||
// Why: the pane's next status supersedes the clear either way — reclaimed below or
|
||||
// outdated by a live event — so the tombstone is one-shot.
|
||||
const cleared = this.clearedStatusTimingByPaneKey.get(payload.paneKey)
|
||||
this.clearedStatusTimingByPaneKey.delete(payload.paneKey)
|
||||
// Why: a disconnect clear removed `previous`, so a reconnect replay of that same turn
|
||||
// would otherwise mint a new stateStartedAt — re-firing the task-complete notification
|
||||
// the user already got, since completion dedupe keys on it (STA-3524).
|
||||
const reclaimedStateStartedAt =
|
||||
!previous && replayReclaimsClearedTurn(cleared, payload) ? cleared.stateStartedAt : undefined
|
||||
const stateStartedAt =
|
||||
previous && previous.payload.state === payload.payload.state && !commandCodeNewTurn
|
||||
? previous.stateStartedAt
|
||||
: now
|
||||
: (reclaimedStateStartedAt ?? now)
|
||||
return {
|
||||
...payload,
|
||||
// Why: receivedAt must stay live — it has to sort after the clear watermark or the
|
||||
// renderer drops the replayed row as a resurrected pre-disconnect status.
|
||||
receivedAt: now,
|
||||
stateStartedAt
|
||||
}
|
||||
@@ -2241,6 +2281,7 @@ export class AgentHookServer {
|
||||
this.closedAgentStatusTabIds.clear()
|
||||
this.closedAgentStatusPaneKeys.clear()
|
||||
this.connectionTimestampWatermarkById.clear()
|
||||
this.clearedStatusTimingByPaneKey.clear()
|
||||
this.legacyPaneKeyAliases.clear()
|
||||
clearAllListenerCaches(this.state)
|
||||
this.notifyStatusChangeListeners()
|
||||
@@ -2289,6 +2330,13 @@ export class AgentHookServer {
|
||||
const deleted = this.deleteStatusEntry(paneKey, { preserveAuthority: true })
|
||||
if (deleted) {
|
||||
statusChanged = true
|
||||
this.clearedStatusTimingByPaneKey.set(paneKey, {
|
||||
connectionId: normalizedConnectionId,
|
||||
state: deleted.payload.state,
|
||||
agentType: deleted.payload.agentType,
|
||||
prompt: deleted.payload.prompt,
|
||||
stateStartedAt: deleted.stateStartedAt
|
||||
})
|
||||
if (deleted.payload.agentType === 'codex') {
|
||||
// Why: a replacement remote process may reuse the pane; don't merge it with the lost connection's children.
|
||||
this.state.codexSubagentRosterByPaneKey.delete(paneKey)
|
||||
@@ -2334,6 +2382,9 @@ export class AgentHookServer {
|
||||
if (!options?.preserveAuthority) {
|
||||
this.hydratedLaunchTokenHashByPaneKey.delete(resolvedPaneKey)
|
||||
this.persistedAuthorityCommitmentsByPaneKey.delete(resolvedPaneKey)
|
||||
// Why: only a transient disconnect expects a replay to reclaim the turn; real
|
||||
// teardown must not leave a tombstone a reused pane could match against.
|
||||
this.clearedStatusTimingByPaneKey.delete(resolvedPaneKey)
|
||||
}
|
||||
this.clearAssistantMessageRetry(resolvedPaneKey)
|
||||
this.clearCodexSubagentPoll(resolvedPaneKey)
|
||||
@@ -2414,6 +2465,7 @@ export class AgentHookServer {
|
||||
this.runtimeObservedStatusPaneKeys.delete(paneKey)
|
||||
this.currentAuthorityObservations.delete(paneKey)
|
||||
this.promptSentDedupeByPaneKey.delete(paneKey)
|
||||
this.clearedStatusTimingByPaneKey.delete(paneKey)
|
||||
}
|
||||
if (aliasChanged) {
|
||||
this.notifyPaneKeyAliasPersistenceListener()
|
||||
@@ -2434,6 +2486,7 @@ export class AgentHookServer {
|
||||
clearPaneCacheState(this.state, resolvedPaneKey)
|
||||
this.currentAuthorityObservations.delete(resolvedPaneKey)
|
||||
this.promptSentDedupeByPaneKey.delete(resolvedPaneKey)
|
||||
this.clearedStatusTimingByPaneKey.delete(resolvedPaneKey)
|
||||
let clearedAlias = false
|
||||
for (const [legacyPaneKey, stablePaneKey] of this.legacyPaneKeyAliases) {
|
||||
if (stablePaneKey.stablePaneKey === resolvedPaneKey) {
|
||||
@@ -2443,6 +2496,7 @@ export class AgentHookServer {
|
||||
clearPaneCacheState(this.state, legacyPaneKey)
|
||||
this.currentAuthorityObservations.delete(legacyPaneKey)
|
||||
this.promptSentDedupeByPaneKey.delete(legacyPaneKey)
|
||||
this.clearedStatusTimingByPaneKey.delete(legacyPaneKey)
|
||||
clearedAlias = true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
createAgentCompletionCoordinator,
|
||||
resetAgentCompletionCoordinatorIdentitiesForTest
|
||||
} from './agent-completion-coordinator'
|
||||
import type { AgentCompletionStatusSnapshot } from './agent-completion-coordinator-types'
|
||||
import { makePaneKey } from '../../../../shared/stable-pane-id'
|
||||
|
||||
const PANE = makePaneKey('tab-1', '11111111-1111-4111-8111-111111111111')
|
||||
const T0 = 1_700_000_000_000
|
||||
const PROMPT = 'ship the release notes'
|
||||
|
||||
function row(state: 'working' | 'done', stateStartedAt: number): AgentCompletionStatusSnapshot {
|
||||
return { state, prompt: PROMPT, agentType: 'claude', stateStartedAt }
|
||||
}
|
||||
|
||||
function trackCompletions(): {
|
||||
observe: (snapshot: AgentCompletionStatusSnapshot) => void
|
||||
count: () => number
|
||||
dispose: () => void
|
||||
} {
|
||||
const completions: string[] = []
|
||||
const coordinator = createAgentCompletionCoordinator({
|
||||
paneKey: PANE,
|
||||
getPtyId: () => 'pty-1',
|
||||
getSettings: () => null,
|
||||
inspectProcess: async () => ({ foregroundProcess: null, hasChildProcesses: false }),
|
||||
dispatchCompletion: (title) => {
|
||||
completions.push(title)
|
||||
},
|
||||
isLive: () => true
|
||||
})
|
||||
return {
|
||||
observe: (snapshot) => coordinator.observeHookStatus(snapshot),
|
||||
count: () => completions.length,
|
||||
dispose: () => coordinator.dispose()
|
||||
}
|
||||
}
|
||||
|
||||
/** The renderer's only defence against a repeated `done` is the turn identity
|
||||
* `state:agentType:stateStartedAt`. These cases pin what a reconnect replay must and must
|
||||
* not carry for that defence to hold — the contract `reconnect-clear-replay-turn-identity.test.ts`
|
||||
* enforces on the main-process side (STA-3524). */
|
||||
describe('reconnect replay and the completion turn identity', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
vi.setSystemTime(T0)
|
||||
resetAgentCompletionCoordinatorIdentitiesForTest()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
resetAgentCompletionCoordinatorIdentitiesForTest()
|
||||
})
|
||||
|
||||
it('re-fires the notification when a replayed done arrives with a re-minted stateStartedAt', () => {
|
||||
const notifications = trackCompletions()
|
||||
notifications.observe(row('working', T0))
|
||||
vi.setSystemTime(T0 + 5_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
expect(notifications.count()).toBe(1)
|
||||
|
||||
// What a clear + replay produced before the fix: the same finished turn, restamped.
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
notifications.observe(row('done', T0 + 61_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
|
||||
expect(notifications.count()).toBe(2)
|
||||
notifications.dispose()
|
||||
})
|
||||
|
||||
it('stays silent when the replayed done keeps the turn identity it was notified under', () => {
|
||||
const notifications = trackCompletions()
|
||||
notifications.observe(row('working', T0))
|
||||
vi.setSystemTime(T0 + 5_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
expect(notifications.count()).toBe(1)
|
||||
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
|
||||
expect(notifications.count()).toBe(1)
|
||||
notifications.dispose()
|
||||
})
|
||||
|
||||
it('stays silent across repeated reconnect cycles replaying the same turn', () => {
|
||||
const notifications = trackCompletions()
|
||||
notifications.observe(row('working', T0))
|
||||
vi.setSystemTime(T0 + 5_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
|
||||
// The reported symptom is a flap: dozens of replays in a few minutes.
|
||||
for (let cycle = 1; cycle <= 12; cycle += 1) {
|
||||
vi.setSystemTime(T0 + 60_000 + cycle * 10_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
}
|
||||
|
||||
expect(notifications.count()).toBe(1)
|
||||
notifications.dispose()
|
||||
})
|
||||
|
||||
it('still notifies a completion first seen on the replay', () => {
|
||||
const notifications = trackCompletions()
|
||||
notifications.observe(row('working', T0))
|
||||
expect(notifications.count()).toBe(0)
|
||||
|
||||
// The turn ended while the client was disconnected, so this done is new to the pane.
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
notifications.observe(row('done', T0 + 61_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
|
||||
expect(notifications.count()).toBe(1)
|
||||
notifications.dispose()
|
||||
})
|
||||
|
||||
it('notifies again for a real next turn after the replayed one', () => {
|
||||
const notifications = trackCompletions()
|
||||
notifications.observe(row('working', T0))
|
||||
vi.setSystemTime(T0 + 5_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.setSystemTime(T0 + 61_000)
|
||||
notifications.observe(row('done', T0 + 5_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
expect(notifications.count()).toBe(1)
|
||||
|
||||
vi.setSystemTime(T0 + 70_000)
|
||||
notifications.observe(row('working', T0 + 70_000))
|
||||
vi.setSystemTime(T0 + 80_000)
|
||||
notifications.observe(row('done', T0 + 80_000))
|
||||
vi.advanceTimersByTime(5_000)
|
||||
|
||||
expect(notifications.count()).toBe(2)
|
||||
notifications.dispose()
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user