starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-09-22 00:02:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(native-chat): count a turn from the send that opened it (#21086)

Browse Source 
* fix(native-chat): count a turn from the send that opened it

The live turn indicator switched on at the submission but anchored its clock at
the provider turn-open, so it jumped back by exactly the dispatch latency the
moment the turn opened. Measured on a real Claude session: the counter climbed to
"Working for 25s", reset to "Working for 0s", then settled "Worked for 26s" —
three readings of one turn, from two different instants.

The host now resolves the send that opened a turn and publishes it as an additive
optional `requestedAt` on the turn lifecycle row. `startedAt` keeps its exact
meaning, the provider turn-open, and is never rewritten, so clients that cannot be
upgraded see no change to any value they already read. Both providers write it;
it is omitted when no send can be named (provider-resumed turns, replayed history).

Readers take one origin, `requestedAt ?? startedAt`, for both the live counter and
the settled host interval, so the two cannot disagree. The provider's own reported
duration keeps outranking the host interval, unchanged.

The host-to-local clock conversion is now latched once per turn rather than
re-derived per render. `receivedAt - hostNow` carries that sample's one-way
delivery latency as well as skew, and the reducer replaces the sample on every
frame, so re-deriving imported fresh jitter and could move the anchor later — the
same class of backwards jump this change removes. With the conversion fixed, an
origin that improves moves the anchor earlier by exactly that much, so displayed
elapsed only grows. No monotonicity guard is added; the ordering is structural.

Desktop and mobile drove byte-identical copies of the timing hook, so both are
collapsed onto one React-free helper in shared.

Regression tests drive the origin resolution rather than an already-resolved
anchor, assert in milliseconds because second-flooring hides the sub-second case,
and include a deliberate host/client skew so a raw timestamp assignment cannot
pass on a machine where the two clocks agree.

* fix(native-chat): correlate Codex turn origins by echo

* fix(native-chat): preserve causal turn timing ownership

* fix(native-chat): keep settled turn timing continuous
This commit is contained in:
Brennan Benson
2026-09-16 15:54:04 -07:00
committed by GitHub
parent 1c4f271478
commit 533b0bd02e
41 changed files with 1330 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile/src/session/use-mobile-structured-agent-turn-timing.ts
+24 -40
View File
@@ -4,38 +4,19 @@ import type {
AgentJournalSubmission
} from '../../../src/shared/agent-session-journal-types'
import type { NativeChatSettledTurns } from '../../../src/shared/native-chat-turn-status'
import type { StructuredAgentHostClock } from '../../../src/shared/structured-agent-session-reducer'
import {
selectStructuredAgentRunningTurnTiming,
selectStructuredAgentSettledTurns,
structuredAgentTurnLocalStartedAt
selectStructuredAgentSettledTurns
} from '../../../src/shared/structured-agent-session-turn-timing'
type TurnAnchor = { turnId: string; startedAt: number | null }
/** The host's clock as last published, paired with the client clock at receipt. */
type HostClock = { hostNow: number; receivedAt: number }
/** The live turn's local-clock anchor. Null when its row carries no host start
* (older hosts), so local observation applies. */
function anchorRunningTurn(
items: readonly AgentJournalRenderItem[],
turnId: string,
hostClock: HostClock | null | undefined
): TurnAnchor {
const timing = selectStructuredAgentRunningTurnTiming(items, turnId)
if (!timing) {
return { turnId, startedAt: null }
}
const now = Date.now()
// Advance the published host clock by the client time since receipt; both
// terms stay single-clock, so a mid-turn attach counts from the real start.
const hostNow = hostClock ? hostClock.hostNow + (now - hostClock.receivedAt) : undefined
return { turnId, startedAt: structuredAgentTurnLocalStartedAt(timing, now, hostNow) }
}
import {
stepStructuredAgentTurnClock,
type StructuredAgentTurnClockLatch
} from '../../../src/shared/structured-agent-turn-clock-anchor'
/** Host-recorded turn timing for the structured lane: settled durations straight
* off the journal, and a skew-free start for the live counter stamped once per
* turn so re-renders never move it. */
* off the journal, and a skew-free start for the live counter whose host-to-local
* conversion is latched once per turn. */
export function useMobileStructuredAgentTurnTiming(
{
items,
@@ -44,7 +25,7 @@ export function useMobileStructuredAgentTurnTiming(
}: {
items: readonly AgentJournalRenderItem[]
submissions: readonly AgentJournalSubmission[]
hostClock?: HostClock | null
hostClock?: StructuredAgentHostClock | null
},
turnId: string | null
): { settledTurns: NativeChatSettledTurns; workingStartedAt: number | null } {
@@ -52,19 +33,22 @@ export function useMobileStructuredAgentTurnTiming(
() => selectStructuredAgentSettledTurns(items, submissions),
[items, submissions]
)
const [anchor, setAnchor] = useState<TurnAnchor | null>(null)
const [latch, setLatch] = useState<StructuredAgentTurnClockLatch | null>(null)
const runningTiming = useMemo(
() => (turnId === null ? null : selectStructuredAgentRunningTurnTiming(items, turnId)),
[items, turnId]
)
// Stamp during render (React's derive-from-props pattern) so the first paint of
// a new turn already counts from the right instant.
if (turnId === null) {
if (anchor !== null) {
setAnchor(null)
}
return { settledTurns, workingStartedAt: null }
const step = stepStructuredAgentTurnClock({
timing: runningTiming,
turnId,
now: Date.now,
hostClock,
latch
})
if (step.latch !== latch) {
setLatch(step.latch)
}
if (anchor?.turnId !== turnId) {
const next = anchorRunningTurn(items, turnId, hostClock)
setAnchor(next)
return { settledTurns, workingStartedAt: next.startedAt }
}
return { settledTurns, workingStartedAt: anchor.startedAt }
return { settledTurns, workingStartedAt: step.workingStartedAt }
}
src/main/claude/claude-structured-control-actions.test.ts
+1
View File
@@ -68,6 +68,7 @@ describe('cancelClaudeTurn', () => {
clientMessageId: `client-${index}`,
sentUuid,
dispatchSequence: index + 1,
requestedAt: null,
replayContentKey: `content-${index}`,
resolve: resolutions[index]!
}))
src/main/claude/claude-structured-dispatch-admission.test.ts
+37 -1
View File
@@ -2,7 +2,7 @@
// completes, and nothing about elapsed time ever puts a message in doubt.
import { describe, expect, it, vi } from 'vitest'
import { dispatchClaudeTurn, resolveClaudeReplayWaiter } from './claude-structured-dispatch'
import { dispatchClaudeTurn, resolveClaudeReplayTurn } from './claude-structured-dispatch'
import {
childExited,
sessionFor,
@@ -10,7 +10,43 @@ import {
userReplayFrame
} from './claude-structured-dispatch-test-support'
function resolveClaudeReplayWaiter(...args: Parameters<typeof resolveClaudeReplayTurn>): boolean {
return resolveClaudeReplayTurn(...args) !== null
}
describe('Claude structured dispatch admission', () => {
it('opens queued exact replays with the origin owned by each send', async () => {
const session = sessionFor()
await dispatchClaudeTurn(session, {
clientMessageId: 'client-a',
body: userMessage([{ type: 'text', text: 'a' }]),
requestedAt: 100
})
const aUuid = session.dispatchWaiters[0]!.sentUuid
expect(resolveClaudeReplayTurn(session, userReplayFrame(aUuid, 'a'))).toEqual({
requestedAt: 100
})
await dispatchClaudeTurn(session, {
clientMessageId: 'client-b',
body: userMessage([{ type: 'text', text: 'b' }]),
requestedAt: 200
})
await dispatchClaudeTurn(session, {
clientMessageId: 'client-c',
body: userMessage([{ type: 'text', text: 'c' }]),
requestedAt: 300
})
const [b, c] = session.dispatchWaiters
expect(resolveClaudeReplayTurn(session, userReplayFrame(b!.sentUuid, 'b'))).toEqual({
requestedAt: 200
})
expect(resolveClaudeReplayTurn(session, userReplayFrame(c!.sentUuid, 'c'))).toEqual({
requestedAt: 300
})
})
it('settles a send queued behind a running turn when that turn starts, with no doubt in between', async () => {
vi.useFakeTimers()
try {
src/main/claude/claude-structured-dispatch.test.ts
+7 -3
View File
@@ -2,7 +2,7 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it, vi } from 'vitest'
import { dispatchClaudeTurn, resolveClaudeReplayWaiter } from './claude-structured-dispatch'
import { dispatchClaudeTurn, resolveClaudeReplayTurn } from './claude-structured-dispatch'
import { readClaudeImage } from './claude-structured-dispatch-content'
import { claudeUnwrittenUserMessageError } from './claude-agent-sdk-user-message-queue'
import type { ClaudeSession } from './claude-structured-session-state'
@@ -13,6 +13,10 @@ import {
userReplayFrame
} from './claude-structured-dispatch-test-support'
function resolveClaudeReplayWaiter(...args: Parameters<typeof resolveClaudeReplayTurn>): boolean {
return resolveClaudeReplayTurn(...args) !== null
}
describe('Claude structured dispatch image limits', () => {
it.each(['isMeta', 'isSynthetic', 'isCompactSummary'])(
'does not acknowledge a dispatch with %s context even when the client uuid matches',
@@ -52,7 +56,7 @@ describe('Claude structured dispatch image limits', () => {
expect(session.dispatchWaiters).toHaveLength(0)
})
it('recovers the active identity when a replay lands after the child died', async () => {
it('settles a retired identity without reopening a turn after the child died', async () => {
const session = sessionFor()
const dispatched = dispatchClaudeTurn(session, {
clientMessageId: 'client-1',
@@ -65,7 +69,7 @@ describe('Claude structured dispatch image limits', () => {
expect(session.dispatchWaiters).toHaveLength(0)
expect(session.retiredDispatchWaiters).toHaveLength(1)
expect(resolveClaudeReplayWaiter(session, userReplayFrame(sentUuid!, 'one'))).toBe(true)
expect(resolveClaudeReplayWaiter(session, userReplayFrame(sentUuid!, 'one'))).toBe(false)
expect(session.retiredDispatchWaiters).toHaveLength(0)
})
src/main/claude/claude-structured-dispatch.ts
+33 -26
View File
@@ -30,11 +30,13 @@ const MAX_ACTIVE_DISPATCH_WAITERS = 64
/** Settles a provider-proven late outcome; replay rows independently reconcile acceptance. */
export type ClaudeLateDispatchSettlement = (input: ClaudeLateDispatchOutcome) => void
export function resolveClaudeReplayWaiter(
export type ClaudeReplayTurnOrigin = { requestedAt: number | null }
export function resolveClaudeReplayTurn(
session: ClaudeSession,
message: Record<string, unknown>,
onSettledLate?: ClaudeLateDispatchSettlement
): boolean {
): ClaudeReplayTurnOrigin | null {
const envelope = readClaudeMessageEnvelope(message)
const isUserReplay =
envelope?.role === 'user' &&
@@ -45,11 +47,11 @@ export function resolveClaudeReplayWaiter(
(!isUserReplay && !isCompletedCommand) ||
readClaudeFrameString(message, 'session_id') !== session.providerSessionId
) {
return false
return null
}
const uuid = readClaudeFrameString(message, 'uuid')
if (!uuid) {
return false
return null
}
// Newer SDK frames carry the client uuid that caused a turn. A correlation
@@ -62,27 +64,29 @@ export function resolveClaudeReplayWaiter(
)
if (exact) {
settleWaiter(session, exact, uuid, onSettledLate)
return isUserReplay && exact.dispatchSequence === session.dispatchSequence
return isUserReplay ? { requestedAt: exact.requestedAt } : null
}
const retired = session.retiredDispatchWaiters.find(
(candidate) => candidate.sentUuid === userMessageUuid
)
if (retired) {
forgetRetiredWaiter(session, retired)
return recoverLateIdentity(session, retired, uuid, isUserReplay, onSettledLate)
recoverLateIdentity(session, retired, uuid, isUserReplay, onSettledLate)
return null
}
return false
return null
}
const exact = session.dispatchWaiters.find((candidate) => candidate.sentUuid === uuid)
if (exact) {
settleWaiter(session, exact, uuid, onSettledLate)
return isUserReplay && exact.dispatchSequence === session.dispatchSequence
return isUserReplay ? { requestedAt: exact.requestedAt } : null
}
const retired = session.retiredDispatchWaiters.find((candidate) => candidate.sentUuid === uuid)
if (retired) {
forgetRetiredWaiter(session, retired)
return recoverLateIdentity(session, retired, uuid, isUserReplay, onSettledLate)
recoverLateIdentity(session, retired, uuid, isUserReplay, onSettledLate)
return null
}
if (isUserReplay) {
@@ -96,8 +100,9 @@ export function resolveClaudeReplayWaiter(
(candidate) => candidate.replayContentKey === replayContentKey
)
if (compatible.length === 1) {
settleWaiter(session, compatible[0]!, uuid, onSettledLate)
return compatible[0]!.dispatchSequence === session.dispatchSequence
const [candidate] = compatible
settleWaiter(session, candidate!, uuid, onSettledLate)
return { requestedAt: candidate!.requestedAt }
}
} else if (!session.replayContentFallbackBlocked && session.dispatchWaiters.length === 0) {
const lateCompatible = session.retiredDispatchWaiters.filter(
@@ -106,30 +111,31 @@ export function resolveClaudeReplayWaiter(
if (lateCompatible.length === 1) {
const [candidate] = lateCompatible
forgetRetiredWaiter(session, candidate!)
return recoverLateIdentity(session, candidate!, uuid, true, onSettledLate)
recoverLateIdentity(session, candidate!, uuid, true, onSettledLate)
return null
}
}
return false
return null
}
const current = session.dispatchWaiters[0]
if (isCompletedCommand && !current?.acceptsResult) {
return false
return null
}
// A legacy result has no dispatch correlation. Any retired waiter makes queue order ambiguous,
// even when the retired dispatch was an ordinary turn rather than a slash command.
if (isCompletedCommand && session.retiredDispatchWaiters.length > 0) {
return false
return null
}
// Once an eviction occurred, a fresh result uuid cannot be joined to a waiter by queue order.
if (isCompletedCommand && session.replayContentFallbackBlocked) {
return false
return null
}
const waiter = uuid ? session.dispatchWaiters.shift() : undefined
if (waiter && uuid) {
settleWaiter(session, waiter, uuid, onSettledLate)
return isUserReplay
return isUserReplay ? { requestedAt: waiter.requestedAt } : null
}
return false
return null
}
function settleWaiter(
@@ -166,20 +172,18 @@ function recoverLateIdentity(
uuid: string,
isUserReplay: boolean,
onSettledLate?: ClaudeLateDispatchSettlement
): boolean {
): void {
if (!isUserReplay && !waiter.acceptsResult) {
return false
return
}
// The provider acted on this dispatch, so the send it came from is delivered.
// Unfenced on purpose: the dispatch-sequence check below only decides whether
// this replay still opens a turn, while delivery is settled for good either way.
// A retired replay settles delivery only; it cannot reopen a turn.
if (waiter.clientMessageId) {
onSettledLate?.({
clientMessageId: waiter.clientMessageId,
providerIdentity: { provider: 'claude', sessionId: session.providerSessionId, uuid }
})
}
return isUserReplay && waiter.dispatchSequence === session.dispatchSequence
}
/**
@@ -194,7 +198,8 @@ function waitForReplay(
acceptsResult: boolean,
sentUuid: string,
replayContentKey: string,
clientMessageId: string | null
clientMessageId: string | null,
requestedAt: number | null
): { waiter: ClaudeDispatchWaiter; promise: Promise<string | null> } {
let waiter!: ClaudeDispatchWaiter
const promise = new Promise<string | null>((resolve) => {
@@ -203,6 +208,7 @@ function waitForReplay(
clientMessageId,
sentUuid,
dispatchSequence: session.dispatchSequence,
requestedAt,
replayContentKey,
resolve
}
@@ -273,7 +279,7 @@ export function retireClaudeDispatchWaiters(session: ClaudeSession): void {
export async function dispatchClaudeTurn(
session: ClaudeSession,
input: { clientMessageId?: string; body: AgentJournalMessageItem }
input: { clientMessageId?: string; body: AgentJournalMessageItem; requestedAt?: number }
): Promise<AgentSessionDispatchOutcome> {
let content: unknown[]
try {
@@ -294,7 +300,8 @@ export async function dispatchClaudeTurn(
acceptsResult,
sentUuid,
claudeDispatchContentKey(content),
input.clientMessageId ?? null
input.clientMessageId ?? null,
input.requestedAt ?? null
)
const replayed = replay.promise
try {
src/main/claude/claude-structured-journal-translation.ts
+9 -2
View File
@@ -132,7 +132,8 @@ export function createClaudeJournalTranslator(
const handleMessage = (
message: Record<string, unknown>,
startsTurn: boolean,
observedAt: number
observedAt: number,
requestedAt?: number
): boolean => {
const envelope = readClaudeMessageEnvelope(message)
if (!envelope) {
@@ -208,6 +209,7 @@ export function createClaudeJournalTranslator(
frame: message,
startsTurn,
observedAt,
...(requestedAt === undefined ? {} : { requestedAt }),
userItemId: agentJournalItemKey(identity)
})
if (sendEchoTurn) {
@@ -274,7 +276,12 @@ export function createClaudeJournalTranslator(
subagents.observeSystemFrame(event.message)
const kind = claudeProviderFrameKind(event.message)
if (
!handleMessage(event.message, event.startsTurn === true, event.observedAt ?? Date.now())
!handleMessage(
event.message,
event.startsTurn === true,
event.observedAt ?? Date.now(),
event.requestedAt
)
) {
providerFallback.append(kind, event.message)
}
src/main/claude/claude-structured-session-acquisition.ts
+7 -4
View File
@@ -8,7 +8,7 @@ import { CLAUDE_AUTH_SWITCH_IN_PROGRESS_MESSAGE } from '../claude-accounts/envir
import { isClaudeAuthSwitchInProgress } from '../claude-accounts/live-pty-gate'
import { openClaudeStreamJsonConnection } from './claude-stream-json-connection'
import { buildClaudePermissionCallbacks } from './claude-structured-inbound-control'
import { resolveClaudeReplayWaiter } from './claude-structured-dispatch'
import { resolveClaudeReplayTurn } from './claude-structured-dispatch'
import {
claudeAuthDiagnostic,
readClaudeCapabilities,
@@ -117,20 +117,23 @@ export async function acquireClaudeSession({
liveSession.leafUuid = observedLeafUuid
observeClaudeFastModeFacts(liveSession, message)
}
const startsTurn = liveSession
? resolveClaudeReplayWaiter(liveSession, message, (settlement) =>
const turnOrigin = liveSession
? resolveClaudeReplayTurn(liveSession, message, (settlement) =>
deps.onDispatchSettledLate?.({ sessionId, ...settlement })
)
: false
: null
const startsTurn = turnOrigin !== null
// Turn endpoints are stamped on the host clock, never the frame's own timestamp.
const observedAt =
startsTurn || message.type === 'result' ? { observedAt: deps.now?.() ?? Date.now() } : {}
const requestedAt = turnOrigin?.requestedAt
callbacks.deliver(attempt, sessionId, () =>
callbacks.emit(liveSession, input.events, {
type: 'message',
sessionId,
message,
...(startsTurn ? { startsTurn: true } : {}),
...(requestedAt === null || requestedAt === undefined ? {} : { requestedAt }),
...observedAt
})
)
src/main/claude/claude-structured-session-adapter.test.ts
+39
View File
@@ -274,6 +274,45 @@ describe('ClaudeStructuredSessionAdapter.acquire', () => {
).resolves.toEqual({ cancelled: true })
})
it('opens each queued exact replay with its own request origin', async () => {
const claude = fakeClaude({ replayUuid: null })
const events: ClaudeStructuredSessionEvent[] = []
const adapter = await acquired(claude, {}, events)
const connection = claude.connections[0]!
const dispatch = async (clientMessageId: string, requestedAt: number): Promise<void> => {
await expect(
adapter.dispatch({
sessionId: 'session-1',
clientMessageId,
body: USER_MESSAGE,
fence: 7,
requestedAt
})
).resolves.toEqual({ state: 'admitted' })
}
const echo = (index: number): void => {
const sent = connection.sent[index]!
connection.handlers.onMessage?.({
...sent,
uuid: `turn-${index + 1}`,
user_message_uuid: sent.uuid
})
}
await dispatch('client-a', 100)
echo(0)
await dispatch('client-b', 200)
await dispatch('client-c', 300)
echo(1)
echo(2)
expect(
events
.filter((event) => event.type === 'message' && event.startsTurn === true)
.map((event) => (event.type === 'message' ? event.requestedAt : undefined))
).toEqual([100, 200, 300])
})
it('quarantines SDK frames without the acquired session identity', async () => {
const claude = fakeClaude({ replayUuid: null })
const events: ClaudeStructuredSessionEvent[] = []
src/main/claude/claude-structured-session-state.ts
+6 -1
View File
@@ -34,6 +34,9 @@ export type ClaudeStructuredSessionEvent =
message: Record<string, unknown>
/** Present only when this replay acknowledged Orca's in-flight dispatch. */
startsTurn?: true
/** Submission instant of the dispatch this replay acknowledged; the origin
* of the turn it opens. Absent when the host cannot name a send. */
requestedAt?: number
/** Host clock at receipt; stamped on turn boundaries only. */
observedAt?: number
}
@@ -110,8 +113,10 @@ export type ClaudeDispatchWaiter = {
clientMessageId: string | null
/** Client uuid echoed by Claude so a replay is tied to its own dispatch. */
sentUuid: string
/** Sequence used to fence a late identity from a newer dispatch. */
/** Sequence used to identify the latest pending dispatch for control ownership. */
dispatchSequence: number
/** Host submission instant owned by this exact dispatch. */
requestedAt: number | null
/** Set when the provider replay settled this waiter before send returned. */
settledUuid?: string
/** The write failed or the child died, but a replay may still name it. */
src/main/claude/claude-turn-lifecycle-item.ts
+9 -2
View File
@@ -11,6 +11,9 @@ export type ClaudeCurrentTurn = {
sessionId: string
turnId: string
startedAt: number
/** Host clock at the send that opened the turn; absent when the provider
* resumed on its own and no send of Orca's names this turn. */
requestedAt?: number
/** Provider key of the user echo, or the lifecycle row itself when provider
* output opened a turn with no user row to receive its timing. */
userItemId: string
@@ -69,7 +72,10 @@ export function claudeTurnLifecycleItem(
options: StructuredAgentSessionAppendOptions
publishCoalescingKey: string
} {
const { sessionId, turnId, startedAt, userItemId } = turn
const { sessionId, turnId, startedAt, requestedAt, userItemId } = turn
// Write-once: the terminal revision republishes the value the running row
// already carried, because both are built from the same open turn.
const requested = requestedAt === undefined ? {} : { requestedAt }
return {
identity: claudeTurnLifecycleIdentity(sessionId, turnId),
body: agentJournalTurnBody(
@@ -78,11 +84,12 @@ export function claudeTurnLifecycleItem(
turnId,
state: end.state,
startedAt,
...requested,
completedAt: end.completedAt,
userItemId,
...(end.durationMs === undefined ? {} : { durationMs: end.durationMs })
}
: { turnId, state: 'running', startedAt, userItemId }
: { turnId, state: 'running', startedAt, ...requested, userItemId }
),
// The running row's ts is the turn start itself, so clients read no append lag.
options: end ? {} : { observedAt: startedAt },
src/main/claude/claude-turn-opening.ts
+3
View File
@@ -26,6 +26,8 @@ export type ClaudeSendEchoTurnInput = {
/** Orca dispatched this send and the provider is replaying it back. */
startsTurn: boolean
observedAt: number
/** Host clock on the submission row that produced this send, when known. */
requestedAt?: number
/** Provider key of the user row this turn is anchored to. */
userItemId: string
}
@@ -43,6 +45,7 @@ export function claudeTurnOpenedBySendEcho(
sessionId: envelope.sessionId,
turnId: envelope.uuid,
startedAt: input.observedAt,
...(input.requestedAt === undefined ? {} : { requestedAt: input.requestedAt }),
userItemId: input.userItemId
}
: null
src/main/claude/claude-turn-ownership.test.ts
+1
View File
@@ -283,6 +283,7 @@ describe('Claude turn ownership', () => {
clientMessageId: 'client-2',
sentUuid: 'uncertain',
dispatchSequence: 1,
requestedAt: null,
replayContentKey: 'ship-it',
resolve: vi.fn(),
retired: true
src/main/codex/codex-structured-dispatch-admission.test.ts
+171 -2
View File
@@ -1,4 +1,7 @@
import { describe, expect, it } from 'vitest'
import type { AgentJournalItemBody } from '../../shared/agent-session-journal-types'
import { agentJournalSubmissionKey } from '../../shared/agent-session-journal-item-key'
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
import { MAX_CODEX_PENDING_DISPATCH_ECHOES } from './codex-structured-dispatch-echo'
import {
acquiredCodexAdapter,
@@ -12,16 +15,33 @@ import {
function send(
adapter: Awaited<ReturnType<typeof acquiredCodexAdapter>>,
clientMessageId: string
clientMessageId: string,
requestedAt?: number
): Promise<unknown> {
return adapter.dispatch({
sessionId: 'session-1',
clientMessageId,
body: CODEX_TEST_USER_MESSAGE,
fence: 7
fence: 7,
...(requestedAt === undefined ? {} : { requestedAt })
})
}
function lifecycleRecorder(): {
sink: StructuredAgentSessionEventSink
bodies: AgentJournalItemBody[]
} {
const bodies: AgentJournalItemBody[] = []
return {
bodies,
sink: {
appendItem: (_identity, body) => bodies.push(body),
appendTombstone: () => {},
publish: () => {}
}
}
}
describe('codex dispatch admission', () => {
it('admits a send queued behind a running turn and settles it when Codex echoes it', async () => {
// Measured on codex-cli 0.153.4: a `turn/start` issued while a turn runs is
@@ -166,6 +186,155 @@ describe('codex dispatch admission', () => {
])
})
it('does not give a later turn the request time of an abandoned unknown send', async () => {
let attempt = 0
const codex = fakeCodexAppServer({
'turn/start': () => {
attempt += 1
if (attempt === 1) {
throw new Error('request timed out after write')
}
return { turn: { id: 'turn-later' } }
}
})
const settlements: LateSettlement[] = []
const recorded = lifecycleRecorder()
const adapter = await acquiredCodexAdapter({ codex, settlements, sink: recorded.sink })
const connection = codex.connections[0]!
await expect(send(adapter, 'client-unknown', 1_700_000_000_100)).rejects.toThrow(
'request timed out after write'
)
await send(adapter, 'client-later', 1_700_000_000_400)
startTurn(connection, 'turn-later')
echoUserMessage(connection, {
turnId: 'turn-later',
itemId: 'item-later',
clientId: 'client-later'
})
connection.handlers.onNotification?.('turn/completed', {
threadId: CODEX_TEST_THREAD_ID,
turn: { id: 'turn-later' }
})
const turns = recorded.bodies.filter((body) => body.kind === 'turn')
expect(turns).toMatchObject([
{ turnId: 'turn-later', state: 'running', startedAt: 1_700_000_000_500 },
{
turnId: 'turn-later',
state: 'running',
requestedAt: 1_700_000_000_400
},
{
turnId: 'turn-later',
state: 'completed',
requestedAt: 1_700_000_000_400
}
])
expect(
turns.some((turn) => turn.kind === 'turn' && turn.requestedAt === 1_700_000_000_100)
).toBe(false)
})
it('does not attribute a send armed after an autonomous turn started', async () => {
const codex = fakeCodexAppServer({
'turn/start': () => ({ turn: { id: 'turn-resumed', status: 'inProgress' } })
})
const settlements: LateSettlement[] = []
const recorded = lifecycleRecorder()
const adapter = await acquiredCodexAdapter({ codex, settlements, sink: recorded.sink })
const connection = codex.connections[0]!
startTurn(connection, 'turn-resumed')
await send(adapter, 'client-mid-turn', 1_700_000_000_100)
echoUserMessage(connection, {
turnId: 'turn-resumed',
itemId: 'item-mid-turn',
clientId: 'client-mid-turn'
})
const turns = recorded.bodies.filter((body) => body.kind === 'turn')
expect(turns).toHaveLength(1)
expect(turns[0]).not.toHaveProperty('requestedAt')
expect(turns[0]).not.toHaveProperty('userItemId', agentJournalSubmissionKey('client-mid-turn'))
expect(settlements.map(({ clientMessageId }) => clientMessageId)).toEqual(['client-mid-turn'])
})
it('keeps the earliest dispatched origin across out-of-order echoes and a clock step', async () => {
const codex = fakeCodexAppServer({
'turn/start': () => ({ turn: { id: 'turn-1', status: 'inProgress' } })
})
const settlements: LateSettlement[] = []
const recorded = lifecycleRecorder()
const adapter = await acquiredCodexAdapter({ codex, settlements, sink: recorded.sink })
const connection = codex.connections[0]!
await send(adapter, 'client-opening', 1_700_000_000_600)
await send(adapter, 'client-queued', 1_700_000_000_200)
startTurn(connection, 'turn-1')
await send(adapter, 'client-mid-turn', 1_700_000_000_100)
echoUserMessage(connection, {
turnId: 'turn-1',
itemId: 'item-queued',
clientId: 'client-queued'
})
echoUserMessage(connection, {
turnId: 'turn-1',
itemId: 'item-mid-turn',
clientId: 'client-mid-turn'
})
echoUserMessage(connection, {
turnId: 'turn-1',
itemId: 'item-opening',
clientId: 'client-opening'
})
expect(
recorded.bodies
.filter((body) => body.kind === 'turn' && body.state === 'running')
.map((body) => (body.kind === 'turn' ? body.requestedAt : undefined))
).toEqual([undefined, 1_700_000_000_200, 1_700_000_000_600])
expect(settlements.map(({ clientMessageId }) => clientMessageId)).toEqual([
'client-queued',
'client-mid-turn',
'client-opening'
])
expect(recorded.bodies.findLast((body) => body.kind === 'turn')).toMatchObject({
requestedAt: 1_700_000_000_600,
userItemId: agentJournalSubmissionKey('client-opening')
})
})
it('revises a completed turn when its exact echo arrives late', async () => {
const codex = fakeCodexAppServer({
'turn/start': () => ({ turn: { id: 'turn-1', status: 'inProgress' } })
})
const settlements: LateSettlement[] = []
const recorded = lifecycleRecorder()
const adapter = await acquiredCodexAdapter({ codex, settlements, sink: recorded.sink })
const connection = codex.connections[0]!
await send(adapter, 'client-late-echo', 1_700_000_000_100)
startTurn(connection, 'turn-1')
connection.handlers.onNotification?.('turn/completed', {
threadId: CODEX_TEST_THREAD_ID,
turn: { id: 'turn-1' }
})
echoUserMessage(connection, {
turnId: 'turn-1',
itemId: 'item-late',
clientId: 'client-late-echo'
})
expect(recorded.bodies.findLast((body) => body.kind === 'turn')).toMatchObject({
state: 'completed',
requestedAt: 1_700_000_000_100,
userItemId: agentJournalSubmissionKey('client-late-echo')
})
expect(settlements.map(({ clientMessageId }) => clientMessageId)).toEqual(['client-late-echo'])
})
it('refuses overflow without discarding an older accepted send', async () => {
const codex = fakeCodexAppServer({ 'turn/start': () => ({ turn: { id: 'turn-1' } }) })
const settlements: LateSettlement[] = []
src/main/codex/codex-structured-dispatch-echo.test.ts
+20
View File
@@ -26,6 +26,26 @@ describe('codex dispatch echoes', () => {
expect(echoes.size).toBe(0)
})
it('reads each submission instant by client message id', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('stale-unknown', 100)
echoes.arm('later-turn', 200)
expect(echoes.requestOrigin('later-turn')).toEqual({ requestedAt: 200, sequence: 1 })
expect(echoes.requestOrigin('stale-unknown')).toEqual({ requestedAt: 100, sequence: 0 })
expect(echoes.requestOrigin('never-armed')).toBeNull()
expect(echoes.latestSequence()).toBe(1)
})
it('keeps one causal sequence when an unconfirmed send retries', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1', 100)
echoes.arm('client-1', 200)
expect(echoes.requestOrigin('client-1')).toEqual({ requestedAt: 100, sequence: 0 })
expect(echoes.latestSequence()).toBe(0)
})
it('refuses an echo this session never armed', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
src/main/codex/codex-structured-dispatch-echo.ts
+34 -8
View File
@@ -1,9 +1,14 @@
import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types'
/** Sends awaiting their echo, oldest first. A send whose echo never arrives is
/** Sends awaiting their echo. A send whose echo never arrives is
* retired by the journal's pending-submission recovery on exit, not from here. */
export const MAX_CODEX_PENDING_DISPATCH_ECHOES = 256
export type CodexDispatchRequestOrigin = {
requestedAt: number
sequence: number
}
/**
* Which sends this session is still waiting to hear back about, keyed by the
* client message id Codex echoes on the user message.
@@ -14,29 +19,50 @@ export const MAX_CODEX_PENDING_DISPATCH_ECHOES = 256
*/
export type CodexDispatchEchoes = {
/** Arms settlement for a send about to be written; false preserves older waits at capacity. */
arm: (clientMessageId: string) => boolean
arm: (clientMessageId: string, requestedAt?: number) => boolean
/** True once, for a send this session armed and has not yet settled. */
settle: (clientMessageId: string) => boolean
/** Drops an armed send whose write never reached the provider. */
disarm: (clientMessageId: string) => void
/** Submission origin for this exact send, retained until its echo settles it. */
requestOrigin: (clientMessageId: string) => CodexDispatchRequestOrigin | null
/** Highest causal sequence assigned to a dispatch in this session. */
latestSequence: () => number
clear: () => void
readonly size: number
}
export function createCodexDispatchEchoes(): CodexDispatchEchoes {
const armed = new Set<string>()
const armed = new Map<string, { requestedAt: number | null; sequence: number }>()
let nextSequence = 0
return {
arm(clientMessageId) {
if (!armed.has(clientMessageId) && armed.size >= MAX_CODEX_PENDING_DISPATCH_ECHOES) {
arm(clientMessageId, requestedAt) {
const existing = armed.get(clientMessageId)
if (existing) {
if (existing.requestedAt === null && requestedAt !== undefined) {
existing.requestedAt = requestedAt
}
return true
}
if (armed.size >= MAX_CODEX_PENDING_DISPATCH_ECHOES) {
return false
}
armed.delete(clientMessageId)
armed.add(clientMessageId)
armed.set(clientMessageId, { requestedAt: requestedAt ?? null, sequence: nextSequence++ })
return true
},
settle: (clientMessageId) => armed.delete(clientMessageId),
disarm: (clientMessageId) => void armed.delete(clientMessageId),
clear: () => armed.clear(),
requestOrigin: (clientMessageId) => {
const origin = armed.get(clientMessageId)
return origin?.requestedAt === null || origin === undefined
? null
: { requestedAt: origin.requestedAt, sequence: origin.sequence }
},
latestSequence: () => nextSequence - 1,
clear: () => {
armed.clear()
nextSequence = 0
},
get size() {
return armed.size
}
src/main/codex/codex-structured-journal-contracts.ts
+8 -1
View File
@@ -1,4 +1,5 @@
import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types'
import type { CodexDispatchRequestOrigin } from './codex-structured-dispatch-echo'
import 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 { CodexStructuredSessionEvent } from './codex-structured-session-adapter'
@@ -20,6 +21,8 @@ export type CodexJournalTranslatorDeps = {
* identity the journal row carries so a replay computes the same key. */
onUserMessageEcho?: (clientMessageId: string, identity: AgentJournalItemIdentity) => void
primaryThreadId?: () => string | null
/** Submission origin for one exact client message still awaiting its echo. */
dispatchRequestOrigin?: (clientMessageId: string) => CodexDispatchRequestOrigin | null
subagentExecutions?: CodexSubagentExecutions
coalesceMs?: number
maxRetainedBytes?: number
@@ -44,6 +47,10 @@ export type CodexJournalTranslationAdmission =
export type CodexItemTranslation =
| { handled: false }
| { handled: true; admission: CodexJournalTranslationAdmission }
| {
handled: true
admission: CodexJournalTranslationAdmission
dispatchEcho?: { clientMessageId: string; providerIdentity: AgentJournalItemIdentity }
}
export const CODEX_JOURNAL_ADMITTED = { accepted: true } as const
src/main/codex/codex-structured-journal-items.ts
+5 -4
View File
@@ -41,7 +41,7 @@ export class CodexJournalItems {
constructor(
private readonly deps: Pick<
CodexJournalTranslatorDeps,
'sink' | 'coalesceMs' | 'maxRetainedBytes' | 'schedule' | 'onUserMessageEcho'
'sink' | 'coalesceMs' | 'maxRetainedBytes' | 'schedule'
> & { maxMetadataBytes?: number },
private readonly activeTurn: (threadId: string) => string | null,
private readonly suppress: (threadId: string, turnId: string) => void
@@ -80,10 +80,11 @@ export class CodexJournalItems {
// Count echoes for stable resume ordinals, but user bubbles come from submissions.
if (source === 'live' && item.type === 'userMessage') {
const echo = readCodexDispatchEcho(item, identity)
if (echo) {
this.deps.onUserMessageEcho?.(echo.clientMessageId, echo.providerIdentity)
return {
handled: true,
admission: CODEX_JOURNAL_ADMITTED,
...(echo ? { dispatchEcho: echo } : {})
}
return { handled: true, admission: CODEX_JOURNAL_ADMITTED }
}
if (item.type === 'contextCompaction' && event.method === 'item/started') {
return { handled: true, admission: CODEX_JOURNAL_ADMITTED }
src/main/codex/codex-structured-journal-translation-turn-boundaries.ts
+94 -13
View File
@@ -1,4 +1,5 @@
import type { AgentJournalTurnLifecycle } from '../../shared/agent-session-journal-types'
import { agentJournalSubmissionKey } from '../../shared/agent-session-journal-item-key'
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
import {
CODEX_JOURNAL_ADMITTED,
@@ -6,7 +7,11 @@ import {
} from './codex-structured-journal-contracts'
import type { CodexJournalItems } from './codex-structured-journal-items'
import { settleCodexJournalTurn } from './codex-structured-journal-settlement'
import type { CodexJournalActiveTurns } from './codex-structured-journal-translation-turn-state'
import {
CodexJournalRecentTurns,
type CodexJournalActiveTurns
} from './codex-structured-journal-translation-turn-state'
import type { CodexDispatchRequestOrigin } from './codex-structured-dispatch-echo'
import {
codexTurnLifecycleState,
codexTurnUserItemId,
@@ -24,10 +29,13 @@ type TurnBoundaryEvent = {
threadId: string
params: unknown
observedAt?: number
dispatchSequenceAtReceipt?: number
}
/** Opens and settles the durable lifecycle row for each primary-thread turn. */
export class CodexJournalTurnBoundaries {
private readonly recentTurns = new CodexJournalRecentTurns()
constructor(
private readonly deps: {
sink: StructuredAgentSessionEventSink
@@ -61,12 +69,63 @@ export class CodexJournalTurnBoundaries {
startedAt
})
if (admission.accepted) {
this.deps.activeTurns.remember(event.threadId, turnId, startedAt)
this.deps.activeTurns.remember(
event.threadId,
turnId,
startedAt,
event.dispatchSequenceAtReceipt
)
this.deps.resetActivity(event.threadId)
}
return admission
}
/** Revises a turn only after Codex echoes the exact send inside it. */
attributeRequest(input: {
sessionId: string
clientMessageId: string
threadId: string
turnId: string
requestOrigin: CodexDispatchRequestOrigin
}): CodexJournalTranslationAdmission {
if (input.threadId !== this.deps.primaryThreadId()) {
return CODEX_JOURNAL_ADMITTED
}
const requestOrigin = {
...input.requestOrigin,
userItemId: agentJournalSubmissionKey(input.clientMessageId)
}
const activeRevision = this.deps.activeTurns.requestOriginRevision(
input.threadId,
input.turnId,
requestOrigin
)
const settledRevision = activeRevision
? null
: this.recentTurns.requestOriginRevision(input.threadId, input.turnId, requestOrigin)
const revision = activeRevision ?? settledRevision
if (!revision) {
return CODEX_JOURNAL_ADMITTED
}
const admission = publishCodexTurnLifecycle({
sink: this.deps.sink,
primaryThreadId: this.deps.primaryThreadId(),
sessionId: input.sessionId,
threadId: input.threadId,
turnId: input.turnId,
state: settledRevision?.state ?? 'running',
...revision
})
if (admission.accepted) {
if (activeRevision) {
this.deps.activeTurns.rememberRequestOrigin(input.threadId, input.turnId, requestOrigin)
} else if (settledRevision) {
this.recentTurns.remember(input.threadId, settledRevision, requestOrigin)
}
}
return admission
}
complete(event: TurnBoundaryEvent): CodexJournalTranslationAdmission {
const suppressionAdmission = this.deps.flushSuppression()
if (!suppressionAdmission.accepted) {
@@ -80,27 +139,41 @@ export class CodexJournalTurnBoundaries {
// the turn that spawned them and go on reporting into the same group, so a
// turn boundary is no evidence contact was lost. Only `settleSession` may
// write `unverifiable`.
const turnLifecycle =
event.threadId === this.deps.primaryThreadId()
? this.settled(
event.threadId,
turnId,
codexTurnLifecycleState(readCodexTurnStatus(event.params)),
this.receiptTime(event),
readCodexTurnDurationMs(event.params)
)
: null
const requestOrigin = this.deps.activeTurns.requestOrigin(event.threadId, turnId)
const latestDispatchSequence = this.deps.activeTurns.latestDispatchSequence(
event.threadId,
turnId
)
const admission = settleCodexJournalTurn({
sink: this.deps.sink,
sessionId: event.sessionId,
threadId: event.threadId,
turnId,
turnLifecycle:
event.threadId === this.deps.primaryThreadId()
? this.settled(
event.threadId,
turnId,
codexTurnLifecycleState(readCodexTurnStatus(event.params)),
this.receiptTime(event),
readCodexTurnDurationMs(event.params)
)
: null,
turnLifecycle,
streams: this.deps.items.streams,
activeItems: this.deps.items.activeItems,
pendingPrompts: this.deps.pendingPrompts,
...(this.deps.clearPromptTurn ? { clearPromptTurn: this.deps.clearPromptTurn } : {})
})
if (admission.accepted) {
if (turnLifecycle) {
this.recentTurns.remember(
event.threadId,
turnLifecycle,
requestOrigin,
latestDispatchSequence
)
}
this.deps.items.ordinals.forgetTurn(event.threadId, turnId)
this.deps.activeTurns.forget(event.threadId, turnId)
this.deps.resetActivity(event.threadId)
@@ -117,16 +190,24 @@ export class CodexJournalTurnBoundaries {
durationMs: number | null = null
): AgentJournalTurnLifecycle {
const startedAt = this.deps.activeTurns.startedAt(threadId, turnId)
// Carried forward from the exact echoed send that was attributed to this turn.
const requestOrigin = this.deps.activeTurns.requestOrigin(threadId, turnId)
return {
turnId,
state,
userItemId: codexTurnUserItemId(threadId, turnId),
userItemId: requestOrigin?.userItemId ?? codexTurnUserItemId(threadId, turnId),
...(startedAt !== undefined ? { startedAt } : {}),
...(requestOrigin !== undefined ? { requestedAt: requestOrigin.requestedAt } : {}),
completedAt,
...(durationMs !== null ? { durationMs } : {})
}
}
clear(): void {
this.deps.activeTurns.clear()
this.recentTurns.clear()
}
private receiptTime(event: TurnBoundaryEvent): number {
return event.observedAt ?? this.deps.now?.() ?? Date.now()
}
src/main/codex/codex-structured-journal-translation-turn-lifecycle.test.ts
+53 -4
View File
@@ -261,6 +261,55 @@ describe('codex turn lifecycle rows', () => {
deferred.close()
})
it('settles an echoed send only after its request-origin revision is admitted', () => {
const tap = recorder()
let rejectOrigin = true
tap.sink.tryAppendItem = (identity, body, blobs) => {
if (body.kind === 'turn' && body.requestedAt !== undefined && rejectOrigin) {
return { accepted: false, reason: 'backpressure' }
}
tap.sink.appendItem(identity, body, blobs)
return { accepted: true }
}
const onUserMessageEcho = vi.fn()
const translator = createCodexJournalTranslator({
sink: tap.sink,
sessionId: SESSION_ID,
primaryThreadId: () => THREAD_ID,
dispatchRequestOrigin: () => ({ requestedAt: 900, sequence: 0 }),
onUserMessageEcho
})
const echo = notification(
'item/started',
{
turn: { id: TURN_ID },
item: { type: 'userMessage', id: 'user-1', clientId: 'client-1' }
},
1_100
)
translator.handle(notification('turn/started', { turn: { id: TURN_ID } }, 1_000))
expect(translator.handle(echo)).toEqual({ accepted: false, reason: 'backpressure' })
expect(onUserMessageEcho).not.toHaveBeenCalled()
expect(tap.rows.map((row) => row.body)).toEqual([
expect.objectContaining({ kind: 'turn', state: 'running', startedAt: 1_000 })
])
rejectOrigin = false
expect(translator.handle(echo)).toEqual({ accepted: true })
expect(onUserMessageEcho).toHaveBeenCalledOnce()
expect(onUserMessageEcho).toHaveBeenCalledWith(
'client-1',
expect.objectContaining({ provider: 'codex', threadId: THREAD_ID, turnId: TURN_ID })
)
expect(tap.rows.at(-1)?.body).toMatchObject({
kind: 'turn',
state: 'running',
startedAt: 1_000,
requestedAt: 900
})
})
it('carries the provider duration and the same user item onto the terminal row', () => {
const tap = recorder()
const translator = translatorFor(tap)
@@ -363,13 +412,13 @@ describe('codex turn lifecycle rows', () => {
translate
})
expect(retries.handle(SESSION_ID, 'turn/started', { turn: { id: TURN_ID } }, 1_000)).toEqual({
accepted: false,
reason: 'backpressure'
})
expect(
retries.handle(SESSION_ID, 'turn/started', { turn: { id: TURN_ID } }, 1_000, -1)
).toEqual({ accepted: false, reason: 'backpressure' })
await vi.advanceTimersByTimeAsync(50)
expect(translate.mock.calls.map((call) => call[4])).toEqual([1_000, 1_000])
expect(translate.mock.calls.map((call) => call[5])).toEqual([-1, -1])
expect(connection.resumeReading).not.toHaveBeenCalled()
})
src/main/codex/codex-structured-journal-translation-turn-state.test.ts
+64 -1
View File
@@ -1,8 +1,11 @@
import { describe, expect, it } from 'vitest'
import {
CodexJournalActiveTurns,
CodexJournalRecentTurns,
MAX_CODEX_ACTIVE_TURN_BYTES,
MAX_CODEX_ACTIVE_TURNS
MAX_CODEX_ACTIVE_TURNS,
MAX_CODEX_RECENT_TURN_BYTES,
MAX_CODEX_RECENT_TURNS
} from './codex-structured-journal-translation-turn-state'
describe('CodexJournalActiveTurns', () => {
@@ -52,4 +55,64 @@ describe('CodexJournalActiveTurns', () => {
active.forget('thread', 'turn-1')
expect(active.startedAt('thread', 'turn-1')).toBeUndefined()
})
it('uses dispatch order even when the host clock moves backwards', () => {
const active = new CodexJournalActiveTurns()
active.remember('thread', 'turn-1', 1_000, 1)
const laterSend = { requestedAt: 700, sequence: 1, userItemId: 'later-send' }
const openingSend = { requestedAt: 1_100, sequence: 0, userItemId: 'opening-send' }
expect(active.requestOriginRevision('thread', 'turn-1', laterSend)).toMatchObject({
userItemId: 'later-send'
})
active.rememberRequestOrigin('thread', 'turn-1', laterSend)
expect(active.requestOriginRevision('thread', 'turn-1', openingSend)).toMatchObject({
requestedAt: 1_100,
userItemId: 'opening-send'
})
active.rememberRequestOrigin('thread', 'turn-1', openingSend)
expect(active.requestOriginRevision('thread', 'turn-1', laterSend)).toBeNull()
})
it('does not attribute a dispatch armed after the provider turn started', () => {
const active = new CodexJournalActiveTurns()
active.remember('thread', 'turn-1', 1_000, -1)
expect(
active.requestOriginRevision('thread', 'turn-1', {
requestedAt: 900,
sequence: 0,
userItemId: 'mid-turn-send'
})
).toBeNull()
})
})
describe('CodexJournalRecentTurns', () => {
it('evicts the oldest terminal turn at its bounded capacity', () => {
const recent = new CodexJournalRecentTurns()
for (let index = 0; index <= MAX_CODEX_RECENT_TURNS; index += 1) {
recent.remember('thread', {
turnId: `turn-${index}`,
state: 'completed',
userItemId: `user-${index}`,
startedAt: 1_000,
completedAt: 2_000
})
}
expect(recent.size).toBe(MAX_CODEX_RECENT_TURNS)
expect(recent.bytes).toBeLessThanOrEqual(MAX_CODEX_RECENT_TURN_BYTES)
expect(
recent.requestOriginRevision('thread', 'turn-0', {
requestedAt: 900,
sequence: 0,
userItemId: 'opening-send'
})
).toBeNull()
recent.clear()
expect(recent.size).toBe(0)
expect(recent.bytes).toBe(0)
})
})
src/main/codex/codex-structured-journal-translation-turn-state.ts
+170 -1
View File
@@ -1,5 +1,19 @@
import type { AgentJournalTurnLifecycle } from '../../shared/agent-session-journal-types'
import type { CodexDispatchRequestOrigin } from './codex-structured-dispatch-echo'
export const MAX_CODEX_ACTIVE_TURNS = 256
export const MAX_CODEX_ACTIVE_TURN_BYTES = 256 * 1024
export const MAX_CODEX_RECENT_TURNS = 256
export const MAX_CODEX_RECENT_TURN_BYTES = 256 * 1024
export type CodexJournalRequestOrigin = CodexDispatchRequestOrigin & { userItemId: string }
function earlierRequestOrigin(
candidate: CodexJournalRequestOrigin,
current: CodexJournalRequestOrigin | undefined
): boolean {
return current === undefined || candidate.sequence < current.sequence
}
export class CodexJournalActiveTurns {
/** Bounds active turn keys retained across provider threads. */
@@ -7,6 +21,10 @@ export class CodexJournalActiveTurns {
readonly byThread = new Map<string, Set<string>>()
/** Host turn-start receipt per remembered turn; the terminal row carries it forward. */
private readonly startedAtByTurn = new Map<string, number>()
/** Earliest dispatched exact send per remembered turn, carried onto terminal rows. */
private readonly requestOriginByTurn = new Map<string, CodexJournalRequestOrigin>()
/** Last dispatch armed before each provider turn-start event. */
private readonly latestDispatchSequenceByTurn = new Map<string, number>()
private activeCount = 0
private retainedBytes = 0
@@ -43,7 +61,54 @@ export class CodexJournalActiveTurns {
return this.startedAtByTurn.get(this.turnKey(threadId, turnId))
}
remember(threadId: string, turnId: string, startedAt?: number): boolean {
requestOrigin(threadId: string, turnId: string): CodexJournalRequestOrigin | undefined {
return this.requestOriginByTurn.get(this.turnKey(threadId, turnId))
}
latestDispatchSequence(threadId: string, turnId: string): number | undefined {
return this.latestDispatchSequenceByTurn.get(this.turnKey(threadId, turnId))
}
requestOriginRevision(
threadId: string,
turnId: string,
requestOrigin: CodexJournalRequestOrigin
): { startedAt: number; requestedAt: number; userItemId: string } | null {
const startedAt = this.startedAt(threadId, turnId)
const latestDispatchSequence = this.latestDispatchSequence(threadId, turnId)
if (
startedAt === undefined ||
latestDispatchSequence === undefined ||
requestOrigin.sequence > latestDispatchSequence
) {
return null
}
const current = this.requestOrigin(threadId, turnId)
return earlierRequestOrigin(requestOrigin, current)
? {
startedAt,
requestedAt: requestOrigin.requestedAt,
userItemId: requestOrigin.userItemId
}
: null
}
rememberRequestOrigin(
threadId: string,
turnId: string,
requestOrigin: CodexJournalRequestOrigin
): void {
if (this.byThread.get(threadId)?.has(turnId)) {
this.requestOriginByTurn.set(this.turnKey(threadId, turnId), requestOrigin)
}
}
remember(
threadId: string,
turnId: string,
startedAt?: number,
latestDispatchSequence = Number.MAX_SAFE_INTEGER
): boolean {
const active = this.byThread.get(threadId)
if (active?.has(turnId)) {
return true
@@ -54,6 +119,7 @@ export class CodexJournalActiveTurns {
if (startedAt !== undefined) {
this.startedAtByTurn.set(this.turnKey(threadId, turnId), startedAt)
}
this.latestDispatchSequenceByTurn.set(this.turnKey(threadId, turnId), latestDispatchSequence)
if (active) {
active.add(turnId)
} else {
@@ -66,6 +132,8 @@ export class CodexJournalActiveTurns {
forget(threadId: string, turnId: string): void {
this.startedAtByTurn.delete(this.turnKey(threadId, turnId))
this.requestOriginByTurn.delete(this.turnKey(threadId, turnId))
this.latestDispatchSequenceByTurn.delete(this.turnKey(threadId, turnId))
const active = this.byThread.get(threadId)
if (active?.delete(turnId)) {
this.activeCount -= 1
@@ -79,7 +147,108 @@ export class CodexJournalActiveTurns {
clear(): void {
this.byThread.clear()
this.startedAtByTurn.clear()
this.requestOriginByTurn.clear()
this.latestDispatchSequenceByTurn.clear()
this.activeCount = 0
this.retainedBytes = 0
}
}
type RecentTurn = {
lifecycle: AgentJournalTurnLifecycle
requestOrigin?: CodexJournalRequestOrigin
latestDispatchSequence: number
bytes: number
}
/** Bounded terminal lifecycle window for exact echoes that arrive after completion. */
export class CodexJournalRecentTurns {
private readonly turns = new Map<string, RecentTurn>()
private retainedBytes = 0
get size(): number {
return this.turns.size
}
get bytes(): number {
return this.retainedBytes
}
private turnKey(threadId: string, turnId: string): string {
return `${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}`
}
remember(
threadId: string,
lifecycle: AgentJournalTurnLifecycle,
requestOrigin?: CodexJournalRequestOrigin,
latestDispatchSequence?: number
): void {
const key = this.turnKey(threadId, lifecycle.turnId)
const existing = this.turns.get(key)
if (existing) {
this.retainedBytes -= existing.bytes
this.turns.delete(key)
}
const causalSequence =
latestDispatchSequence ?? existing?.latestDispatchSequence ?? Number.MAX_SAFE_INTEGER
const bytes = Buffer.byteLength(
JSON.stringify({
threadId,
lifecycle,
requestOrigin,
latestDispatchSequence: causalSequence
}),
'utf8'
)
if (bytes > MAX_CODEX_RECENT_TURN_BYTES) {
return
}
this.turns.set(key, {
lifecycle,
...(requestOrigin ? { requestOrigin } : {}),
latestDispatchSequence: causalSequence,
bytes
})
this.retainedBytes += bytes
while (
this.turns.size > MAX_CODEX_RECENT_TURNS ||
this.retainedBytes > MAX_CODEX_RECENT_TURN_BYTES
) {
const oldest = this.turns.keys().next().value
if (typeof oldest !== 'string') {
break
}
const removed = this.turns.get(oldest)
this.turns.delete(oldest)
this.retainedBytes = Math.max(0, this.retainedBytes - (removed?.bytes ?? 0))
}
}
requestOriginRevision(
threadId: string,
turnId: string,
requestOrigin: CodexJournalRequestOrigin
): AgentJournalTurnLifecycle | null {
const current = this.turns.get(this.turnKey(threadId, turnId))
const startedAt = current?.lifecycle.startedAt
if (
!current ||
startedAt === undefined ||
requestOrigin.sequence > current.latestDispatchSequence ||
!earlierRequestOrigin(requestOrigin, current.requestOrigin)
) {
return null
}
return {
...current.lifecycle,
requestedAt: requestOrigin.requestedAt,
userItemId: requestOrigin.userItemId
}
}
clear(): void {
this.turns.clear()
this.retainedBytes = 0
}
}
src/main/codex/codex-structured-journal-translation-turns.ts
+4 -1
View File
@@ -56,7 +56,9 @@ export function publishCodexTurnLifecycle(input: {
threadId: string
turnId: string
state: AgentJournalTurnLifecycleState
userItemId?: string
startedAt?: number
requestedAt?: number
completedAt?: number
durationMs?: number
}): StructuredAgentSessionSinkAdmission {
@@ -67,8 +69,9 @@ export function publishCodexTurnLifecycle(input: {
const body = codexTurnLifecycleBody({
turnId: input.turnId,
state: input.state,
userItemId: codexTurnUserItemId(input.threadId, input.turnId),
userItemId: input.userItemId ?? codexTurnUserItemId(input.threadId, input.turnId),
...(input.startedAt !== undefined ? { startedAt: input.startedAt } : {}),
...(input.requestedAt !== undefined ? { requestedAt: input.requestedAt } : {}),
...(input.completedAt !== undefined ? { completedAt: input.completedAt } : {}),
...(input.durationMs !== undefined ? { durationMs: input.durationMs } : {})
})
src/main/codex/codex-structured-journal-translation.ts
+19 -2
View File
@@ -182,7 +182,7 @@ export function createCodexJournalTranslator(
deps.sink.setActivity?.(null)
items.activeItems.clear()
prompts.pending.clear()
activeTurns.clear()
turnBoundaries.clear()
compactions.clear()
goals.clear()
return CODEX_JOURNAL_ADMITTED
@@ -257,6 +257,23 @@ export function createCodexJournalTranslator(
return publishActivity(event, subagentAdmission)
}
const translated = items.handle(event)
if (translated.handled && translated.dispatchEcho) {
const { clientMessageId, providerIdentity } = translated.dispatchEcho
const requestOrigin = deps.dispatchRequestOrigin?.(clientMessageId) ?? null
if (requestOrigin !== null && providerIdentity.provider === 'codex') {
const attribution = turnBoundaries.attributeRequest({
sessionId: event.sessionId,
clientMessageId,
threadId: providerIdentity.threadId,
turnId: providerIdentity.turnId,
requestOrigin
})
if (!attribution.accepted) {
return attribution
}
}
deps.onUserMessageEcho?.(clientMessageId, providerIdentity)
}
return publishActivity(
event,
translated.handled
@@ -284,7 +301,7 @@ export function createCodexJournalTranslator(
prompts.dispose()
genericFrames.dispose()
subagents.dispose()
activeTurns.clear()
turnBoundaries.clear()
compactions.clear()
goals.dispose()
}
src/main/codex/codex-structured-notification-retry.ts
+34 -9
View File
@@ -6,7 +6,13 @@ const MAX_RETRY_EVENTS = 256
const MAX_RETRY_BYTES = 8 * 1024 * 1024
const RETRY_DELAY_MS = 25
type PendingNotification = { method: string; params: unknown; bytes: number; observedAt?: number }
type PendingNotification = {
method: string
params: unknown
bytes: number
observedAt?: number
dispatchSequenceAtReceipt?: number
}
type RetryState = {
connection: CodexAppServerConnection
events: PendingNotification[]
@@ -23,7 +29,8 @@ export function createCodexStructuredNotificationRetry(deps: {
session: CodexSession,
method: string,
params: unknown,
observedAt?: number
observedAt?: number,
dispatchSequenceAtReceipt?: number
) => CodexJournalTranslationAdmission
}) {
const states = new Map<string, RetryState>()
@@ -54,7 +61,8 @@ export function createCodexStructuredNotificationRetry(deps: {
session,
pending.method,
pending.params,
pending.observedAt
pending.observedAt,
pending.dispatchSequenceAtReceipt
)
if (!admission.accepted) {
if (admission.reason === 'backpressure') {
@@ -105,7 +113,8 @@ export function createCodexStructuredNotificationRetry(deps: {
connection: CodexAppServerConnection,
method: string,
params: unknown,
observedAt: number | undefined
observedAt: number | undefined,
dispatchSequenceAtReceipt: number | undefined
): void => {
const bytes = Buffer.byteLength(JSON.stringify({ method, params }), 'utf8')
let state = states.get(sessionId)
@@ -123,7 +132,8 @@ export function createCodexStructuredNotificationRetry(deps: {
method,
params,
bytes,
...(observedAt !== undefined ? { observedAt } : {})
...(observedAt !== undefined ? { observedAt } : {}),
...(dispatchSequenceAtReceipt !== undefined ? { dispatchSequenceAtReceipt } : {})
})
state.bytes += bytes
connection.pauseReading?.()
@@ -134,7 +144,8 @@ export function createCodexStructuredNotificationRetry(deps: {
sessionId: string,
method: string,
params: unknown,
observedAt?: number
observedAt?: number,
dispatchSequenceAtReceipt?: number
): CodexJournalTranslationAdmission => {
const session = deps.sessionFor(sessionId)
if (!session) {
@@ -142,13 +153,27 @@ export function createCodexStructuredNotificationRetry(deps: {
}
const state = states.get(sessionId)
if (state && state.events.length > 0) {
enqueue(sessionId, state.connection, method, params, observedAt)
enqueue(sessionId, state.connection, method, params, observedAt, dispatchSequenceAtReceipt)
retry(sessionId, state.connection)
return { accepted: false, reason: 'backpressure' }
}
const admission = deps.translate(sessionId, session, method, params, observedAt)
const admission = deps.translate(
sessionId,
session,
method,
params,
observedAt,
dispatchSequenceAtReceipt
)
if (!admission.accepted) {
enqueue(sessionId, session.connection, method, params, observedAt)
enqueue(
sessionId,
session.connection,
method,
params,
observedAt,
dispatchSequenceAtReceipt
)
retry(sessionId, session.connection)
}
return admission
src/main/codex/codex-structured-provider-events.ts
+15 -4
View File
@@ -18,15 +18,24 @@ export function translateCodexNotification(input: {
method: string
params: unknown
observedAt?: number
dispatchSequenceAtReceipt?: number
turnCancellation: Pick<CodexStructuredTurnCancellation, 'handleNotification'>
emit: EmitCodexEvent
}): CodexJournalTranslationAdmission {
const { sessionId, session, method, params, observedAt } = input
const { sessionId, session, method, params, observedAt, dispatchSequenceAtReceipt } = input
codexRewind.observeCodexRewindActivity(session, method, params)
if (input.turnCancellation.handleNotification(sessionId, session, method, params, observedAt)) {
return { accepted: true }
}
return deliverCodexNotification(sessionId, session, method, params, input.emit, observedAt)
return deliverCodexNotification(
sessionId,
session,
method,
params,
input.emit,
observedAt,
dispatchSequenceAtReceipt
)
}
export function deliverCodexNotification(
@@ -35,7 +44,8 @@ export function deliverCodexNotification(
method: string,
params: unknown,
emit: EmitCodexEvent,
observedAt?: number
observedAt?: number,
dispatchSequenceAtReceipt?: number
): CodexJournalTranslationAdmission {
if (!session) {
return { accepted: true }
@@ -49,7 +59,8 @@ export function deliverCodexNotification(
threadId,
method,
params,
...(observedAt !== undefined ? { observedAt } : {})
...(observedAt !== undefined ? { observedAt } : {}),
...(dispatchSequenceAtReceipt !== undefined ? { dispatchSequenceAtReceipt } : {})
})
}
src/main/codex/codex-structured-session-acquire.ts
+11 -1
View File
@@ -89,6 +89,7 @@ export async function acquireCodexStructuredSession(input: {
sessionId,
...(deps.now ? { now: deps.now } : {}),
primaryThreadId: () => primaryThreadId,
dispatchRequestOrigin: (clientMessageId) => dispatchEchoes.requestOrigin(clientMessageId),
subagentExecutions,
bindPromptItemId: (journalItemId, threadId, promptKey, turnId) =>
acquisition.prompts.bindJournalItemId(journalItemId, threadId, promptKey, turnId),
@@ -132,10 +133,19 @@ export async function acquireCodexStructuredSession(input: {
onNotification: (method, params) => {
// Stamped at receipt, ahead of any pre-publication buffering or retry.
const observedAt = isCodexTurnBoundary(method) ? (deps.now?.() ?? Date.now()) : undefined
const dispatchSequenceAtReceipt =
method === 'turn/started' ? dispatchEchoes.latestSequence() : undefined
input.deliver(
acquisition,
sessionId,
() => notificationRetries.handle(sessionId, method, params, observedAt),
() =>
notificationRetries.handle(
sessionId,
method,
params,
observedAt,
dispatchSequenceAtReceipt
),
Buffer.byteLength(JSON.stringify(params ?? null), 'utf8')
)
},
src/main/codex/codex-structured-session-adapter.ts
+11 -3
View File
@@ -58,13 +58,14 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
constructor(private readonly deps: CodexStructuredSessionAdapterDeps) {
this.notificationRetries = createCodexStructuredNotificationRetry({
sessionFor: (sessionId) => this.sessions.get(sessionId),
translate: (sessionId, session, method, params, observedAt) =>
translate: (sessionId, session, method, params, observedAt, dispatchSequenceAtReceipt) =>
translateCodexNotification({
sessionId,
session,
method,
params,
observedAt,
dispatchSequenceAtReceipt,
turnCancellation: this.turnCancellation,
emit: (current, event) => this.emit(current, event)
})
@@ -85,8 +86,14 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
emit: (session, event) => {
const admission = this.emit(session, event)
if (!admission.accepted && event.type === 'notification') {
const { sessionId, method, params, observedAt } = event
this.notificationRetries.handle(sessionId, method, params, observedAt)
const { sessionId, method, params, observedAt, dispatchSequenceAtReceipt } = event
this.notificationRetries.handle(
sessionId,
method,
params,
observedAt,
dispatchSequenceAtReceipt
)
}
return admission
}
@@ -203,6 +210,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
clientMessageId: string
body: AgentJournalMessageItem
fence: number
requestedAt?: number
}): Promise<AgentSessionDispatchOutcome> {
const session = this.session(input.sessionId)
session.dispatchPending = true
src/main/codex/codex-structured-session-state.ts
+2
View File
@@ -35,6 +35,8 @@ export type CodexStructuredSessionEvent =
params: unknown
/** Host receipt time of a turn boundary; survives retry and deferral so a replay is not re-stamped. */
observedAt?: number
/** Highest dispatch sequence armed when this turn-start was first received. */
dispatchSequenceAtReceipt?: number
}
| { type: 'server-request'; sessionId: string; threadId: string; method: string; params: unknown }
| { type: 'provider-frame'; sessionId: string; threadId: string; kind: string; payload: unknown }
src/main/codex/codex-structured-turn-start.ts
+10 -4
View File
@@ -90,10 +90,16 @@ function codexTurnOptions(host: CodexTurnHost): Record<string, string> {
*/
export async function startCodexTurn(
host: CodexTurnHost,
input: { clientMessageId: string; body: AgentJournalMessageItem; timeoutMs?: number }
input: {
clientMessageId: string
body: AgentJournalMessageItem
requestedAt?: number
timeoutMs?: number
}
): Promise<boolean> {
// Armed before the write: the echo can land while the response is in flight.
if (!host.dispatchEchoes.arm(input.clientMessageId)) {
// Armed before the write: the echo and `turn/started` can both land while the
// response is in flight, and the start must snapshot this send in its frontier.
if (!host.dispatchEchoes.arm(input.clientMessageId, input.requestedAt)) {
return false
}
await host.connection.request(
@@ -117,7 +123,7 @@ export async function startCodexTurn(
*/
export async function dispatchCodexTurn(
session: CodexTurnHost,
input: { clientMessageId: string; body: AgentJournalMessageItem },
input: { clientMessageId: string; body: AgentJournalMessageItem; requestedAt?: number },
timeoutMs: number | undefined
): Promise<AgentSessionDispatchOutcome> {
try {
src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts
+3
View File
@@ -170,6 +170,9 @@ export type StructuredAgentSessionAdapter = {
clientMessageId: string
body: AgentJournalMessageItem
fence: number
/** Host clock on the submission row this send came from; the origin the turn
* it opens records as `requestedAt`. */
requestedAt?: number
}): Promise<AgentSessionDispatchOutcome>
rewindSupport?(sessionId: string): AgentSessionRewindSupport
recoverRewind?(input: {
src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts
+3
View File
@@ -92,6 +92,9 @@ function settledLifecycle(
if (lifecycle.startedAt !== undefined) {
settled.startedAt = lifecycle.startedAt
}
if (lifecycle.requestedAt !== undefined) {
settled.requestedAt = lifecycle.requestedAt
}
if (verdict.state === 'interrupted') {
settled.completedAt = verdict.completedAt
}
src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts
+10 -3
View File
@@ -54,14 +54,16 @@ function invalid(message: string): { ok: false; refusal: AgentSessionWireRefusal
async function dispatchSafely(
ctx: AgentSessionTurnContext,
clientMessageId: string,
body: AgentJournalMessageItem
body: AgentJournalMessageItem,
requestedAt: number | undefined
): Promise<AgentSessionDispatchOutcome> {
try {
return await ctx.adapter.dispatch({
sessionId: ctx.sessionId,
clientMessageId,
body,
fence: ctx.fence
fence: ctx.fence,
...(requestedAt === undefined ? {} : { requestedAt })
})
} catch (error) {
return { state: 'unknown', reason: error instanceof Error ? error.message : String(error) }
@@ -116,7 +118,12 @@ export async function performSend(
}
ctx.publish()
const outcome = await dispatchSafely(ctx, input.clientMessageId, input.body)
// The row just written is the send's instant on the host clock; the turn this
// dispatch opens records it so the live counter never re-anchors at turn-open.
const requestedAt = ctx.journal
.submissions()
.find((entry) => entry.clientMessageId === input.clientMessageId)?.submittedAt
const outcome = await dispatchSafely(ctx, input.clientMessageId, input.body, requestedAt)
// An admission needs no dispatch row: the submission is already pending.
if (outcome.state === 'admitted') {
ctx.publish()
src/renderer/src/components/native-chat/use-structured-agent-turn-timing.ts
+24 -40
View File
@@ -4,38 +4,19 @@ import type {
AgentJournalSubmission
} from '../../../../shared/agent-session-journal-types'
import type { NativeChatSettledTurns } from '../../../../shared/native-chat-turn-status'
import type { StructuredAgentHostClock } from '../../../../shared/structured-agent-session-reducer'
import {
selectStructuredAgentRunningTurnTiming,
selectStructuredAgentSettledTurns,
structuredAgentTurnLocalStartedAt
selectStructuredAgentSettledTurns
} from '../../../../shared/structured-agent-session-turn-timing'
type TurnAnchor = { turnId: string; startedAt: number | null }
/** The host's clock as last published, paired with the client clock at receipt. */
type HostClock = { hostNow: number; receivedAt: number }
/** The live turn's local-clock anchor. Null when its row carries no host start
* (older hosts), so local observation applies. */
function anchorRunningTurn(
items: readonly AgentJournalRenderItem[],
turnId: string,
hostClock: HostClock | null | undefined
): TurnAnchor {
const timing = selectStructuredAgentRunningTurnTiming(items, turnId)
if (!timing) {
return { turnId, startedAt: null }
}
const now = Date.now()
// Advance the published host clock by the client time since receipt; both
// terms stay single-clock, so a mid-turn attach counts from the real start.
const hostNow = hostClock ? hostClock.hostNow + (now - hostClock.receivedAt) : undefined
return { turnId, startedAt: structuredAgentTurnLocalStartedAt(timing, now, hostNow) }
}
import {
stepStructuredAgentTurnClock,
type StructuredAgentTurnClockLatch
} from '../../../../shared/structured-agent-turn-clock-anchor'
/** Host-recorded turn timing for the structured lane: settled durations straight
* off the journal, and a skew-free start for the live counter stamped once per
* turn so re-renders never move it. */
* off the journal, and a skew-free start for the live counter whose host-to-local
* conversion is latched once per turn. */
export function useStructuredAgentTurnTiming(
{
items,
@@ -44,7 +25,7 @@ export function useStructuredAgentTurnTiming(
}: {
items: readonly AgentJournalRenderItem[]
submissions: readonly AgentJournalSubmission[]
hostClock?: HostClock | null
hostClock?: StructuredAgentHostClock | null
},
turnId: string | null
): { settledTurns: NativeChatSettledTurns; workingStartedAt: number | null } {
@@ -52,19 +33,22 @@ export function useStructuredAgentTurnTiming(
() => selectStructuredAgentSettledTurns(items, submissions),
[items, submissions]
)
const [anchor, setAnchor] = useState<TurnAnchor | null>(null)
const [latch, setLatch] = useState<StructuredAgentTurnClockLatch | null>(null)
const runningTiming = useMemo(
() => (turnId === null ? null : selectStructuredAgentRunningTurnTiming(items, turnId)),
[items, turnId]
)
// Stamp during render (React's derive-from-props pattern) so the first paint of
// a new turn already counts from the right instant.
if (turnId === null) {
if (anchor !== null) {
setAnchor(null)
}
return { settledTurns, workingStartedAt: null }
const step = stepStructuredAgentTurnClock({
timing: runningTiming,
turnId,
now: Date.now,
hostClock,
latch
})
if (step.latch !== latch) {
setLatch(step.latch)
}
if (anchor?.turnId !== turnId) {
const next = anchorRunningTurn(items, turnId, hostClock)
setAnchor(next)
return { settledTurns, workingStartedAt: next.startedAt }
}
return { settledTurns, workingStartedAt: anchor.startedAt }
return { settledTurns, workingStartedAt: step.workingStartedAt }
}
src/shared/agent-session-journal-schemas.ts
+2
View File
@@ -177,6 +177,7 @@ export const AgentJournalItemBodySchema = z.discriminatedUnion('kind', [
state: z.string().min(1),
userItemId: z.string().min(1).optional(),
startedAt: z.number().finite().positive().optional(),
requestedAt: z.number().finite().positive().optional(),
completedAt: z.number().finite().positive().optional(),
durationMs: z.number().finite().nonnegative().optional()
})
@@ -189,6 +190,7 @@ export const AgentJournalItemBodySchema = z.discriminatedUnion('kind', [
state: z.string().min(1),
userItemId: z.string().min(1).optional(),
startedAt: z.number().finite().positive().optional(),
requestedAt: z.number().finite().positive().optional(),
completedAt: z.number().finite().positive().optional(),
durationMs: z.number().finite().nonnegative().optional()
})
src/shared/agent-session-journal-types.ts
+6 -3
View File
@@ -169,11 +169,14 @@ export type AgentJournalTurnLifecycleState = (typeof AGENT_JOURNAL_TURN_LIFECYCL
export type AgentJournalTurnLifecycle = {
turnId: string
state: AgentJournalTurnLifecycleState
/** Provider key of the user item that opened the turn; clients resolve a
* submission alias through it. A lifecycle row may key itself when provider
* output opened a turn with no user item; absent means an older host. */
/** Journal key of the user item that opened the turn. A lifecycle row may key
* itself when provider output opened a turn with no user item; absent means
* an older host. */
userItemId?: string
startedAt?: number
/** Host clock at the send that opened this turn, when one is known. `startedAt`
* remains the provider turn-open instant and is never rewritten. */
requestedAt?: number
completedAt?: number
/** The provider's own measured turn duration, preferred over the host interval. */
durationMs?: number
src/shared/native-chat-turn-status.test.ts
+60
View File
@@ -149,6 +149,30 @@ describe('reduceNativeChatTurnTiming', () => {
expect(next.u1?.startedAt).toBe(500)
})
it('does not move a live turn later before its request origin arrives', () => {
const optimistic = reduceNativeChatTurnTiming(
{},
{ activeTurnKey: 'u1', validTurnKeys, isWorking: true, now: 1_000 }
)
const turnStarted = reduceNativeChatTurnTiming(optimistic, {
activeTurnKey: 'u1',
validTurnKeys,
isWorking: true,
workingStartedAt: 8_000,
now: 8_000
})
const exactOrigin = reduceNativeChatTurnTiming(turnStarted, {
activeTurnKey: 'u1',
validTurnKeys,
isWorking: true,
workingStartedAt: 900,
now: 8_100
})
expect(turnStarted).toBe(optimistic)
expect(exactOrigin.u1).toEqual({ startedAt: 900, workedSeconds: null })
})
it('settles the turn to whole elapsed seconds when work stops', () => {
const working = reduceNativeChatTurnTiming(
{},
@@ -288,6 +312,42 @@ describe('reduceNativeChatTurnTiming', () => {
})
describe('selectNativeChatTurnStatuses', () => {
it('keeps the selected live start monotonic until the exact request origin arrives', () => {
const optimistic = reduceNativeChatTurnTiming(
{},
{ activeTurnKey: 'u1', validTurnKeys: new Set(['u1']), isWorking: true, now: 1_000 }
)
const turnStarted = reduceNativeChatTurnTiming(optimistic, {
activeTurnKey: 'u1',
validTurnKeys: new Set(['u1']),
isWorking: true,
workingStartedAt: 8_000,
now: 8_000
})
const beforeEcho = selectNativeChatTurnStatuses(turnStarted, {
activeTurnKey: 'u1',
isWorking: true,
workingStartedAt: 8_000,
thinking: false
})
const exactOrigin = reduceNativeChatTurnTiming(turnStarted, {
activeTurnKey: 'u1',
validTurnKeys: new Set(['u1']),
isWorking: true,
workingStartedAt: 900,
now: 8_100
})
const afterEcho = selectNativeChatTurnStatuses(exactOrigin, {
activeTurnKey: 'u1',
isWorking: true,
workingStartedAt: 900,
thinking: false
})
expect(beforeEcho.active?.startedAt).toBe(1_000)
expect(afterEcho.active?.startedAt).toBe(900)
})
it('carries the reasoning verdict it is given onto the working turn', () => {
const { active } = selectNativeChatTurnStatuses(
{ u1: { startedAt: 1_000, workedSeconds: null } },
src/shared/native-chat-turn-status.ts
+8 -4
View File
@@ -163,10 +163,14 @@ export function reduceNativeChatTurnTiming(
const timing = retained[activeTurnKey]
if (isWorking) {
// An in-flight turn keeps the start it already had; only a fresh turn (or an
// authoritative host timestamp) restamps it.
// A lifecycle row can arrive before its exact request-origin revision. Keep
// the earlier anchor so publication order can never run the live clock backward.
const startedAt =
workingStartedAt ?? (timing && timing.workedSeconds == null ? timing.startedAt : now)
timing && timing.workedSeconds == null
? workingStartedAt === null || workingStartedAt === undefined
? timing.startedAt
: Math.min(timing.startedAt, workingStartedAt)
: (workingStartedAt ?? now)
if (timing?.startedAt === startedAt && timing.workedSeconds == null) {
return retained
}
@@ -240,7 +244,7 @@ export function selectNativeChatTurnStatuses(
return {
active: isWorking
? {
startedAt: workingStartedAt ?? timingByTurn[activeTurnKey]?.startedAt ?? null,
startedAt: timingByTurn[activeTurnKey]?.startedAt ?? workingStartedAt ?? null,
thinking,
workedSeconds: null
}
src/shared/structured-agent-session-turn-timing.test.ts
+1 -1
View File
@@ -249,7 +249,7 @@ describe('explicit user-item attribution', () => {
})
describe('provider-measured duration', () => {
it('outranks the host interval and floors to seconds', () => {
it('outranks an unattributed host interval and floors to seconds', () => {
expect(
completedStructuredAgentTurnSeconds({
state: 'completed',
src/shared/structured-agent-session-turn-timing.ts
+30 -8
View File
@@ -16,9 +16,12 @@ export type StructuredAgentTurnTiming = {
state: AgentJournalTurnLifecycleState
/** Host clock at provider turn-start receipt. */
startedAt: number
/** Host clock at the send that opened the turn; absent when the host could not
* name one (provider-resumed turns, replayed history, older hosts). */
requestedAt?: number
/** Host clock at the terminal provider event; absent while running or unverifiable. */
completedAt?: number
/** The provider's own measured duration; outranks the host interval. */
/** The provider's own measurement; used when exact host endpoints are unavailable. */
durationMs?: number
/** Host clock when the lifecycle row was appended; with `startedAt` it gives
* the host-side lag a client must subtract to anchor a live counter. */
@@ -30,10 +33,14 @@ function readTiming(item: AgentJournalRenderItem): StructuredAgentTurnTiming | n
if (!turn) {
return null
}
const { state, startedAt, completedAt, durationMs } = turn
const { state, startedAt, requestedAt, completedAt, durationMs } = turn
if (startedAt === undefined || !Number.isFinite(startedAt) || startedAt <= 0) {
return null
}
const requested =
requestedAt !== undefined && Number.isFinite(requestedAt) && requestedAt > 0
? requestedAt
: undefined
const end =
completedAt !== undefined && Number.isFinite(completedAt) && completedAt >= startedAt
? completedAt
@@ -45,15 +52,16 @@ function readTiming(item: AgentJournalRenderItem): StructuredAgentTurnTiming | n
return {
state,
startedAt,
...(requested !== undefined ? { requestedAt: requested } : {}),
...(end !== undefined ? { completedAt: end } : {}),
...(measured !== undefined ? { durationMs: measured } : {}),
observedAt: item.observedAt
}
}
/** Timing keyed by the user message that opened each turn. A row names its
* user item by provider key; a submission the provider later acknowledged is
* reached through its alias. Rows from older hosts carry no key and fall back
/** Timing keyed by the user message that opened each turn. A row can name the
* submission directly or by a provider key that resolves through its alias.
* Rows from older hosts carry no key and fall back
* to the nearest user message before them in journal order — the submission
* row is written ahead of dispatch, so it always precedes the provider's
* turn-start. Untimed rows are skipped unless explicitly unverifiable (null). */
@@ -107,6 +115,13 @@ export function selectStructuredAgentRunningTurnTiming(
return null
}
/** The single instant every reading of a turn's elapsed time counts from: the
* send that opened it when the host named one, the provider turn-open otherwise.
* One origin is what keeps the live counter and the settled duration agreeing. */
export function structuredAgentTurnOrigin(timing: StructuredAgentTurnTiming): number {
return timing.requestedAt ?? timing.startedAt
}
/** Whole seconds a settled turn ran, or null when the host never observed its end. */
export function completedStructuredAgentTurnSeconds(
timing: StructuredAgentTurnTiming | null | undefined
@@ -114,11 +129,15 @@ export function completedStructuredAgentTurnSeconds(
if (!timing || (timing.state !== 'completed' && timing.state !== 'interrupted')) {
return null
}
// Provider durations may begin at turn-open, so exact host endpoints preserve the live origin.
if (timing.requestedAt !== undefined && timing.completedAt !== undefined) {
return Math.max(0, Math.floor((timing.completedAt - timing.requestedAt) / 1000))
}
if (timing.durationMs !== undefined) {
return Math.floor(timing.durationMs / 1000)
}
return timing.completedAt !== undefined
? Math.floor((timing.completedAt - timing.startedAt) / 1000)
? Math.max(0, Math.floor((timing.completedAt - structuredAgentTurnOrigin(timing)) / 1000))
: null
}
@@ -133,10 +152,13 @@ export function structuredAgentTurnLocalStartedAt(
firstSeenAt: number,
hostNow?: number
): number {
const origin = structuredAgentTurnOrigin(timing)
const hostElapsed =
hostNow !== undefined && Number.isFinite(hostNow)
? hostNow - timing.startedAt
: timing.observedAt - timing.startedAt
? hostNow - origin
: timing.observedAt - origin
// Wall-clock, not monotonic: an NTP step can put the origin after the host's
// own reading, and a negative elapsed would run the counter backwards.
return firstSeenAt - Math.max(0, hostElapsed)
}
src/shared/structured-agent-turn-clock-anchor.test.ts
+216
View File
@@ -0,0 +1,216 @@
import { describe, expect, it } from 'vitest'
import type {
AgentJournalRenderItem,
AgentJournalTurnLifecycle
} from './agent-session-journal-types'
import { agentJournalTurnBody } from './agent-session-turn-record'
import {
stepStructuredAgentTurnClock,
type StructuredAgentTurnClockLatch
} from './structured-agent-turn-clock-anchor'
import {
completedStructuredAgentTurnSeconds,
selectStructuredAgentRunningTurnTiming,
structuredAgentTurnOrigin
} from './structured-agent-session-turn-timing'
// The host clock sits an hour ahead of the client's, so any host timestamp that
// leaks into a local anchor shows up as a wild offset instead of hiding on a
// developer machine where the two clocks agree.
const HOST_START = 3_600_000_000
const CLIENT_NOW = 12_345_000
function lifecycle(
turnId: string,
turn: Omit<AgentJournalTurnLifecycle, 'turnId'>,
observedAt: number
): AgentJournalRenderItem {
return {
itemId: `lifecycle-${turnId}`,
revision: 1,
sequence: 1,
observedAt,
body: agentJournalTurnBody({ turnId, ...turn })
}
}
/** A running turn as the host writes it: the row's append time is the provider
* turn-start, so `observedAt - startedAt` is zero and only the origin moves. */
function runningTurn(startedAt: number, requestedAt?: number): AgentJournalRenderItem[] {
return [
lifecycle(
't1',
{ state: 'running', startedAt, ...(requestedAt === undefined ? {} : { requestedAt }) },
startedAt
)
]
}
function anchorFor(
items: AgentJournalRenderItem[],
hostClock: { hostNow: number; receivedAt: number } | null,
latch: StructuredAgentTurnClockLatch | null = null
): { latch: StructuredAgentTurnClockLatch | null; workingStartedAt: number | null } {
return stepStructuredAgentTurnClock({
timing: selectStructuredAgentRunningTurnTiming(items, 't1'),
turnId: 't1',
now: () => CLIENT_NOW,
hostClock,
latch
})
}
describe('structured agent turn clock anchor', () => {
// The defect: the indicator starts at the send but the clock used to anchor at
// the provider turn-open, so it jumped back by exactly the dispatch latency.
// Milliseconds, not the rendered label — second-flooring hides the 898ms case.
it.each([898, 7_051, 25_000])(
'never counts backwards across turn-open with %ims of dispatch latency',
(latencyMs) => {
const requestedAt = HOST_START
const startedAt = HOST_START + latencyMs
const hostClock = { hostNow: startedAt, receivedAt: CLIENT_NOW }
// Before the turn opens the surface counts from its own stamp at the send.
const localStampAtSend = CLIENT_NOW - latencyMs
const elapsedBefore = CLIENT_NOW - localStampAtSend
const { workingStartedAt } = anchorFor(runningTurn(startedAt, requestedAt), hostClock)
const elapsedAfter = CLIENT_NOW - (workingStartedAt ?? CLIENT_NOW)
expect(elapsedAfter).toBeGreaterThanOrEqual(elapsedBefore)
expect(workingStartedAt).toBe(localStampAtSend)
}
)
it('anchors on the client clock, never on the host clock', () => {
const { workingStartedAt } = anchorFor(runningTurn(HOST_START + 1_000, HOST_START), {
hostNow: HOST_START + 1_000,
receivedAt: CLIENT_NOW
})
expect(workingStartedAt).toBe(CLIENT_NOW - 1_000)
// A raw host timestamp assigned straight through would land an hour away.
expect(Math.abs((workingStartedAt ?? 0) - HOST_START)).toBeGreaterThan(1_000_000)
})
// `receivedAt - hostNow` is skew PLUS that sample's one-way delivery latency, and
// the reducer replaces the sample on every frame. Re-deriving would import the
// new latency and could move the anchor later, running the counter backwards.
it('keeps the latched conversion when a later host sample carries more latency', () => {
const items = runningTurn(HOST_START + 1_000, HOST_START)
const first = anchorFor(items, { hostNow: HOST_START + 1_000, receivedAt: CLIENT_NOW })
const jittered = anchorFor(
items,
{ hostNow: HOST_START - 1_000, receivedAt: CLIENT_NOW },
first.latch
)
expect(jittered.latch).toBe(first.latch)
expect(jittered.workingStartedAt).toBe(first.workingStartedAt)
})
it('moves the anchor earlier, never later, when the origin improves', () => {
const startedAt = HOST_START + 5_000
const hostClock = { hostNow: startedAt, receivedAt: CLIENT_NOW }
const withoutOrigin = anchorFor(runningTurn(startedAt), hostClock)
const improved = anchorFor(runningTurn(startedAt, HOST_START), hostClock, withoutOrigin.latch)
expect(improved.workingStartedAt).toBeLessThan(withoutOrigin.workingStartedAt ?? 0)
})
it('falls back to the provider turn-start when the host named no send', () => {
const startedAt = HOST_START + 5_000
const { workingStartedAt } = anchorFor(runningTurn(startedAt), {
hostNow: startedAt,
receivedAt: CLIENT_NOW
})
// Older hosts omit `requestedAt`; the reading is exactly what it is today.
expect(workingStartedAt).toBe(CLIENT_NOW)
})
it('drops the latch when no turn is open', () => {
const open = anchorFor(runningTurn(HOST_START + 1_000, HOST_START), {
hostNow: HOST_START + 1_000,
receivedAt: CLIENT_NOW
})
const closed = stepStructuredAgentTurnClock({
timing: null,
turnId: null,
now: () => CLIENT_NOW,
hostClock: null,
latch: open.latch
})
expect(closed.latch).toBeNull()
expect(closed.workingStartedAt).toBeNull()
})
})
describe('structured agent turn origin', () => {
it('is the send that opened the turn when the host named one', () => {
expect(
structuredAgentTurnOrigin({
state: 'running',
startedAt: HOST_START + 7_051,
requestedAt: HOST_START,
observedAt: HOST_START + 7_051
})
).toBe(HOST_START)
})
// The live counter and the settled row must count from the same instant, or the
// turn ends by contradicting the number it just displayed.
it('settles a turn from the same instant the live counter used', () => {
const settled = completedStructuredAgentTurnSeconds({
state: 'completed',
startedAt: HOST_START + 25_000,
requestedAt: HOST_START,
completedAt: HOST_START + 26_000,
observedAt: HOST_START + 25_000
})
expect(settled).toBe(26)
})
it('does not let a provider start-scoped duration undercut an exact request origin', () => {
const settled = completedStructuredAgentTurnSeconds({
state: 'completed',
startedAt: HOST_START + 25_000,
requestedAt: HOST_START,
completedAt: HOST_START + 26_000,
durationMs: 7_612,
observedAt: HOST_START + 25_000
})
expect(settled).toBe(26)
})
it('falls back to the provider duration when the host did not observe completion', () => {
const settled = completedStructuredAgentTurnSeconds({
state: 'completed',
startedAt: HOST_START + 25_000,
requestedAt: HOST_START,
durationMs: 7_612,
observedAt: HOST_START + 25_000
})
expect(settled).toBe(7)
})
it('clamps a settled host interval when the wall clock moved backward', () => {
const settled = completedStructuredAgentTurnSeconds({
state: 'interrupted',
startedAt: HOST_START - 2_000,
requestedAt: HOST_START,
completedAt: HOST_START - 1_000,
observedAt: HOST_START - 2_000
})
expect(settled).toBe(0)
})
})
src/shared/structured-agent-turn-clock-anchor.ts
+66
View File
@@ -0,0 +1,66 @@
// The client half of a turn's elapsed time: one host-to-local clock conversion,
// latched at first sight of the turn and kept for the rest of it. Desktop and
// mobile both drive this; neither owns a copy.
//
// The offset is latched rather than re-derived because `receivedAt - hostNow` is
// skew PLUS the one-way delivery latency of that one sample, and the reducer
// replaces the sample on every frame carrying one. Re-deriving later would
// import fresh transport jitter and could move the anchor LATER, running the
// counter backwards. With the offset fixed, an origin that improves moves the
// anchor earlier by exactly that much, so displayed elapsed only ever grows.
import type { StructuredAgentHostClock } from './structured-agent-session-reducer'
import {
structuredAgentTurnLocalStartedAt,
type StructuredAgentTurnTiming
} from './structured-agent-session-turn-timing'
/** One turn's conversion basis: the client instant it was first seen, and the
* host's own clock advanced to that instant. Both stamped once, per turn. */
export type StructuredAgentTurnClockLatch = {
turnId: string
firstSeenAt: number
/** Absent until a host sample has arrived; the row's append time applies instead. */
hostNow?: number
}
export function latchStructuredAgentTurnClock(
turnId: string,
now: number,
hostClock: StructuredAgentHostClock | null | undefined
): StructuredAgentTurnClockLatch {
return {
turnId,
firstSeenAt: now,
...(hostClock ? { hostNow: hostClock.hostNow + (now - hostClock.receivedAt) } : {})
}
}
/** Reads the clock. Taken as a thunk because a turn that is already latched must
* not read it at all — the conversion is fixed and a render is not a new sighting. */
export type StructuredAgentTurnClockReader = () => number
/** The live counter's anchor for one render. Returns the latch it was given,
* unchanged, while the turn holds, so a caller can compare by reference. */
export function stepStructuredAgentTurnClock(input: {
timing: StructuredAgentTurnTiming | null
turnId: string | null
now: StructuredAgentTurnClockReader
hostClock: StructuredAgentHostClock | null | undefined
latch: StructuredAgentTurnClockLatch | null
}): { latch: StructuredAgentTurnClockLatch | null; workingStartedAt: number | null } {
const { timing, turnId, latch } = input
if (turnId === null) {
return { latch: null, workingStartedAt: null }
}
const next =
latch?.turnId === turnId
? latch
: latchStructuredAgentTurnClock(turnId, input.now(), input.hostClock)
return {
latch: next,
workingStartedAt: timing
? structuredAgentTurnLocalStartedAt(timing, next.firstSeenAt, next.hostNow)
: null
}
}