Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -84,11 +84,14 @@ export type JournalTombstoneRow = JournalRowBase & {
* `journal-store-open.ts`) but ignores an unknown key; a row kind of its own once released hosts
* skip unknown kinds instead. */
export type JournalStopEvent = {
/** Identity, minted by the writer; a rewind's restatement keeps it. Absent only on rows dev
* builds wrote before it, which no answer names. */
id?: string
/** Persisted: never rename an arm. Only `user-stop` pauses the queue. */
reason: StructuredAgentSessionStopCause
/** The turn the Stop named, else the one running when it took effect. */
turnId?: string
/** When it took effect; a rewind's restatement keeps it. */
/** When it took effect: timing only, never identity. A rewind's restatement keeps it. */
at: number
/** Who asked (`StructuredAgentSessionCaller.callerKey`). */
caller?: string
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ describe("a Stop's event and a Resume", () => {
{ fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
)
const cursor = journal.cursor()
await journal.appendStopEvent({ reason: 'user-stop' }, 1)
await journal.appendStopEvent({ id: 'stop-1', reason: 'user-stop' }, 1)
await journal.appendQueueResume(1)
const since = journal.readSince(cursor)
if (!since.ok) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,14 @@ export function buildJournalQueueResumeRow(input: RowPlace): JournalTombstoneRow
}
}

/** A Stop event as its writer hands it over: with the id the writer minted, so the writer can name
* the event without reading it back, and untimed. */
export type JournalNewStopEvent = Omit<JournalStopEvent, 'at' | 'id'> & { id: string }

/** A Stop taking effect now: its event's time is its row's. */
export function journalStopEventRowBuilder(
state: () => JournalReducerState,
event: Omit<JournalStopEvent, 'at'>,
event: JournalNewStopEvent,
fence: number
): (seq: number, ts: number) => JournalTombstoneRow {
return (seq, ts) =>
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,200 @@
// Whether a Stop still counts, read from its notes' answers: until it has one, and then while any
// says it took. The turn-end rule reads it, so a turn a Stop that took nothing named ends as the
// news it is.

import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import {
AGENT_JOURNAL_THREAD_SCOPE,
type AgentJournalItemIdentity,
type AgentJournalStopNoteAnswer,
type AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record'
import { createTrackedJournalOpener } from './journal-host-database-test-support'
import type { AgentSessionJournal } from './journal-store'

const IDENTITY: AgentSessionJournalIdentity = {
sessionId: 'session-answers',
workspaceId: 'ws-1',
hostId: 'host-1',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: 'thread-1' }
}
const SCOPE = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE }
const TURN: AgentJournalItemIdentity = {
provider: 'codex',
threadId: 'thread-1',
turnId: 'turn-1',
ordinal: 0
}

let root: string
let journal: AgentSessionJournal
let minted = 0
const journals = createTrackedJournalOpener()

beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-stop-answers-'))
minted = 0
// One millisecond for every row: an answer must find its Stop by id, never by time.
journal = await journals.open({ identity: IDENTITY, stateDirectory: root, now: () => 1_000 })
})

afterEach(async () => {
await journals.closeAll()
await rm(root, { recursive: true, force: true })
})

/** A Stop of `turn-1`, a person's unless `reason` says otherwise, as its writer writes it; its
* event's id. */
async function stopped(reason: 'user-stop' | 'user-close' = 'user-stop'): Promise<string> {
const id = `stop-${++minted}`
await journal.appendStopEvent({ id, reason, turnId: 'turn-1' }, 1)
return id
}

/** A Stop's note at `key`, as `performCancel` writes it. */
async function answered(
key: string,
stop: AgentJournalStopNoteAnswer | undefined,
identity: AgentJournalItemIdentity = { provider: 'orca', clientMessageId: `stop:${key}` }
): Promise<void> {
await journal.appendItem(
identity,
{ kind: 'status', text: 'Cancellation requested.', ...(stop ? { stop } : {}) },
SCOPE
)
}

function stillCounts(): boolean {
return journal.stopMarks.stillCounts(journal.stopMarks.latest()!.event)
}

describe('journalStopStillCounts', () => {
it('counts a Stop with no answer yet', async () => {
await stopped()

expect(stillCounts()).toBe(true)
})

it.each(['declined', 'interrupt-unconfirmed', 'no-effect'] as const)(
'stops counting a Stop whose only answer is %s',
async (answer) => {
const eventId = await stopped()
await answered('turn-1', { answer, eventId })

expect(stillCounts()).toBe(false)
}
)

it('keeps counting while any answer says it took, a later one that it did not included', async () => {
const eventId = await stopped()
await answered('turn-1', { answer: 'took', eventId })
await answered('op-2', { answer: 'no-effect', eventId })

expect(stillCounts()).toBe(true)
})

it('counts an owed end as taking effect', async () => {
const eventId = await stopped()
await answered('turn-1', { answer: 'end-owed', eventId })

expect(stillCounts()).toBe(true)
})

it("reads only answers to this Stop: an earlier Stop's in the same millisecond, one naming no event, and an older host's note are none", async () => {
const earlier = await stopped()
await answered('turn-1', { answer: 'no-effect', eventId: earlier })
await stopped()
expect(journal.stopMarks.latest()?.event.at).toBe(1_000)
await answered('op-2', { answer: 'declined' })
await answered('op-3', undefined)

expect(stillCounts()).toBe(true)
})

it("keeps counting a person's close written in the same millisecond as their declined Stop", async () => {
await answered('turn-1', { answer: 'declined', eventId: await stopped() })
await stopped('user-close')

expect(stillCounts()).toBe(true)
expect(journal.stopMarks.personStopDecides('turn-1')).toBe(true)
})

it("reads an answer only from a Stop's note", async () => {
const eventId = await stopped()
await answered('row', { answer: 'declined', eventId }, TURN)

expect(stillCounts()).toBe(true)
})

it('counts an event an early build wrote with no id: no answer names it', async () => {
await answered('op-1', { answer: 'declined' })

expect(journal.stopMarks.stillCounts({})).toBe(true)
})

it('links an answer a rewind keeps to the Stop it restates, by the id the restatement keeps', async () => {
const eventId = await stopped()
const note = { provider: 'orca' as const, clientMessageId: 'stop:turn-1' }

await journal.replaceEpochItems('legacy_import', 1, [
{
identity: note,
body: { kind: 'status', text: 'Codex didn’t stop.', stop: { answer: 'declined', eventId } }
}
])

expect(journal.stopMarks.latest()?.event).toMatchObject({ id: eventId, reason: 'user-stop' })
expect(stillCounts()).toBe(false)
})
})

describe('the turn-end rule', () => {
/** `turn-1` running, and a person's Stop of it; the event's id. */
async function stoppedWhileRunning(): Promise<string> {
await journal.appendItem(
TURN,
{ kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: Date.now() - 1_000 },
SCOPE
)
return stopped()
}

function endTurn(): Promise<unknown> {
return journal.appendItem(
TURN,
{ kind: 'turn', turnId: 'turn-1', state: 'interrupted', completedAt: Date.now() + 1_000 },
SCOPE
)
}

function outcome(): string | undefined {
return journal
.snapshot()
.items.map((item) => readAgentJournalTurn(item.body))
.find((turn) => turn?.turnId === 'turn-1')?.outcome
}

it("makes the end of a turn a Stop that took named the person's cancellation", async () => {
await answered('turn-1', { answer: 'took', eventId: await stoppedWhileRunning() })

await endTurn()

expect(outcome()).toBe('cancellation')
})

it('leaves the end of a turn a declined Stop named as news, so a crash reads as one', async () => {
await answered('turn-1', { answer: 'declined', eventId: await stoppedWhileRunning() })

// Still the person's Stop of the turn, which a host stop defers to; not their end.
expect(journal.stopMarks.personStopCovers('turn-1')).toBe(true)
expect(journal.stopMarks.personStopDecides('turn-1')).toBe(false)
await endTurn()

expect(outcome()).toBeUndefined()
})
})
41 changes: 41 additions & 0 deletions src/main/native-chat/agent-session-journal/journal-stop-answers.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
// Whether a Stop still counts, from the answers its notes record: the one rule every reader of a
// Stop's effect on a turn's end applies (the turn-end write, and through it the readers that must
// choose before the end is written, and the "Stopping…" state).

import {
agentJournalStopAnswerTook,
isStructuredAgentSessionStopNote,
readAgentJournalStopAnswer
} from '../../../shared/agent-session-stop-answer'
import type { JournalReducerState } from './journal-reducer'
import type { JournalStopEvent } from './journal-row-schema'

/**
* Whether the Stop that wrote `event` still counts as taking effect: it has no answer yet, or any
* answer it has says it took. A later answer that it did not take never cancels one that says it
* did. Answers name the event by its id; an event with none (an early dev build's), an older
* host's note, and an answer this build does not know have no answer.
*/
export function journalStopStillCounts(
state: Pick<JournalReducerState, 'items'>,
event: Pick<JournalStopEvent, 'id'>
): boolean {
if (event.id === undefined) {
return true
}
let answered = false
for (const [itemId, item] of state.items) {
if (!isStructuredAgentSessionStopNote(itemId)) {
continue
}
Comment on lines +27 to +30

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Answer lookup scans entire chat. When a Stop’s only answer is declined or no-effect, this lookup examines every journal item and parses each key. It runs when a covered turn ends, adding work that grows with the conversation on the serialized write path. In long chats, that can make turn endings progressively slower.

const stop = readAgentJournalStopAnswer(item.body)
if (stop?.eventId !== event.id) {
continue
}
if (agentJournalStopAnswerTook(stop.answer)) {
return true
}
answered = true
}
return !answered
}
16 changes: 15 additions & 1 deletion src/main/native-chat/agent-session-journal/journal-stop-marks.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,13 @@
// What the journal answers about its Stops beyond the queue's pause: the latest Stop event, which
// the turn-end rule reads (`journal-stop-turn-end.ts`), and whether a person's still decides.
// the turn-end rule reads (`journal-stop-turn-end.ts`), whether a person's covers or still decides
// a turn, and whether a Stop still counts by its answers (`journal-stop-answers.ts`).

import type { JournalReducerState } from './journal-reducer'
import type { JournalStopEvent } from './journal-row-schema'
import { journalStopStillCounts } from './journal-stop-answers'
import {
latestAcceptedSendUnopened,
personStopCoversTurn,
personStopDecidesTurn,
type JournalLatestStop
} from './journal-stop-turn-end'
Expand All @@ -20,8 +24,18 @@ export class JournalStopMarks {
return latestAcceptedSendUnopened(this.deps.state())
}

/** `personStopCoversTurn`: a person's Stop is in force for turn `turnId`, whatever it answered. */
personStopCovers(turnId: string | null): boolean {
return personStopCoversTurn(this.deps.state(), turnId)
}

/** `personStopDecidesTurn`: a person's Stop decides how turn `turnId` ends. */
personStopDecides(turnId: string | null, endedAt?: number, openedBy?: string): boolean {
return personStopDecidesTurn(this.deps.state(), turnId, endedAt, openedBy)
}

/** `journalStopStillCounts`: the Stop that wrote `event` has no answer yet, or one says it took. */
stillCounts(event: Pick<JournalStopEvent, 'id'>): boolean {
return journalStopStillCounts(this.deps.state(), event)
}
}
Loading
Loading