mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
refactor: drop the unused sync spool drain and thread-goal wrapper; reuse the shared absence check (#29820)
The synchronous drainAgentHookSpool and currentAgentSessionThreadGoalBySequence lost their last production callers earlier today; tests now exercise the functions production calls. git-object-quarantine reuses isDefinitiveAbsence instead of a private ENOENT/ENOTDIR copy. Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -14,7 +14,7 @@ import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import {
|
||||
AGENT_HOOK_SPOOL_MAX_FILES,
|
||||
drainAgentHookSpool,
|
||||
drainAgentHookSpoolAsync,
|
||||
launchTokenHash,
|
||||
type SpoolRecord
|
||||
} from '../../shared/agent-hook-spool'
|
||||
@@ -36,7 +36,7 @@ describe('agent hook spool', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('waits for a newline before replaying a complete-looking final record', () => {
|
||||
it('waits for a newline before replaying a complete-looking final record', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-spool-unterminated-'))
|
||||
const spool = join(dir, 'spool')
|
||||
mkdirSync(spool)
|
||||
@@ -52,18 +52,20 @@ describe('agent hook spool', () => {
|
||||
const options = {
|
||||
endpointDir: dir,
|
||||
getPersistedLaunchTokenHash: () => undefined,
|
||||
ingest: (value: SpoolRecord) => ingested.push(value)
|
||||
ingest: (value: SpoolRecord) => {
|
||||
ingested.push(value)
|
||||
}
|
||||
}
|
||||
expect(drainAgentHookSpool(options)).toBe(0)
|
||||
expect(await drainAgentHookSpoolAsync(options)).toBe(0)
|
||||
expect(readFileSync(file, 'utf8')).toBe(record)
|
||||
|
||||
appendFileSync(file, '\n')
|
||||
expect(drainAgentHookSpool(options)).toBe(1)
|
||||
expect(await drainAgentHookSpoolAsync(options)).toBe(1)
|
||||
expect(ingested).toHaveLength(1)
|
||||
expect(readFileSync(file)).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('does not let historical empty pane files starve newer records', () => {
|
||||
it('does not let historical empty pane files starve newer records', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-spool-empty-files-'))
|
||||
const spool = join(dir, 'spool')
|
||||
mkdirSync(spool)
|
||||
@@ -76,16 +78,18 @@ describe('agent hook spool', () => {
|
||||
`\n${JSON.stringify({ paneKey: 'tab:live', source: 'codex', receivedAt: Date.now(), payload: { state: 'done' } })}\n`
|
||||
)
|
||||
const ingested: SpoolRecord[] = []
|
||||
drainAgentHookSpool({
|
||||
await drainAgentHookSpoolAsync({
|
||||
endpointDir: dir,
|
||||
getPersistedLaunchTokenHash: () => undefined,
|
||||
ingest: (record) => ingested.push(record)
|
||||
ingest: (record) => {
|
||||
ingested.push(record)
|
||||
}
|
||||
})
|
||||
expect(ingested).toHaveLength(1)
|
||||
expect(ingested[0]?.paneKey).toBe('tab:live')
|
||||
})
|
||||
|
||||
it('rejects stale launch tokens before ingest and truncates in place', () => {
|
||||
it('rejects stale launch tokens before ingest and truncates in place', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-spool-'))
|
||||
const spool = join(dir, 'spool')
|
||||
mkdirSync(spool)
|
||||
@@ -96,17 +100,19 @@ describe('agent hook spool', () => {
|
||||
)
|
||||
const inode = statSync(file).ino
|
||||
const ingested: unknown[] = []
|
||||
drainAgentHookSpool({
|
||||
await drainAgentHookSpoolAsync({
|
||||
endpointDir: dir,
|
||||
getPersistedLaunchTokenHash: () => launchTokenHash('new')!,
|
||||
ingest: (record) => ingested.push(record)
|
||||
ingest: (record) => {
|
||||
ingested.push(record)
|
||||
}
|
||||
})
|
||||
expect(ingested).toHaveLength(0)
|
||||
expect(readFileSync(file)).toHaveLength(0)
|
||||
expect(statSync(file).ino).toBe(inode)
|
||||
})
|
||||
|
||||
it('ingests a record with the matching launch token', () => {
|
||||
it('ingests a record with the matching launch token', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'orca-spool-'))
|
||||
const spool = join(dir, 'spool')
|
||||
mkdirSync(spool)
|
||||
@@ -116,10 +122,12 @@ describe('agent hook spool', () => {
|
||||
`\n${JSON.stringify({ paneKey: 'tab:1', source: 'codex', launchToken: 'same', receivedAt: Date.now(), payload: { state: 'done' } })}\n`
|
||||
)
|
||||
const ingested: unknown[] = []
|
||||
drainAgentHookSpool({
|
||||
await drainAgentHookSpoolAsync({
|
||||
endpointDir: dir,
|
||||
getPersistedLaunchTokenHash: () => launchTokenHash('same')!,
|
||||
ingest: (record) => ingested.push(record)
|
||||
ingest: (record) => {
|
||||
ingested.push(record)
|
||||
}
|
||||
})
|
||||
expect(ingested).toHaveLength(1)
|
||||
expect(ingested[0]).toMatchObject({ paneKey: 'tab:1', isReplay: true })
|
||||
|
||||
@@ -9,7 +9,7 @@ import {
|
||||
} from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { drainAgentHookSpool, drainAgentHookSpoolAsync } from './agent-hook-spool'
|
||||
import { drainAgentHookSpoolAsync } from './agent-hook-spool'
|
||||
|
||||
const directories: string[] = []
|
||||
afterEach(() => {
|
||||
@@ -30,29 +30,22 @@ function fixture() {
|
||||
return { endpointDir, path, contents, line, getPersistedLaunchTokenHash: () => undefined }
|
||||
}
|
||||
|
||||
it.each(['sync', 'async'] as const)(
|
||||
'retains the complete file after a %s admission fails',
|
||||
async (mode) => {
|
||||
const data = fixture()
|
||||
const ingest = vi.fn((record) => {
|
||||
if (record.payload === 'tool') {
|
||||
throw new Error('admission failed')
|
||||
}
|
||||
})
|
||||
if (mode === 'sync') {
|
||||
expect(() => drainAgentHookSpool({ ...data, ingest })).toThrow('admission failed')
|
||||
} else {
|
||||
await expect(
|
||||
drainAgentHookSpoolAsync({ ...data, ingest: async (record) => ingest(record) })
|
||||
).rejects.toThrow('admission failed')
|
||||
it('retains the complete file after an admission fails', async () => {
|
||||
const data = fixture()
|
||||
const ingest = vi.fn((record) => {
|
||||
if (record.payload === 'tool') {
|
||||
throw new Error('admission failed')
|
||||
}
|
||||
expect(readFileSync(data.path, 'utf8')).toBe(data.contents)
|
||||
const retried = vi.fn()
|
||||
expect(await drainAgentHookSpoolAsync({ ...data, ingest: retried })).toBe(2)
|
||||
expect(retried.mock.calls.map(([record]) => record.payload)).toEqual(['start', 'tool'])
|
||||
expect(readFileSync(data.path, 'utf8')).toBe('')
|
||||
}
|
||||
)
|
||||
})
|
||||
await expect(
|
||||
drainAgentHookSpoolAsync({ ...data, ingest: async (record) => ingest(record) })
|
||||
).rejects.toThrow('admission failed')
|
||||
expect(readFileSync(data.path, 'utf8')).toBe(data.contents)
|
||||
const retried = vi.fn()
|
||||
expect(await drainAgentHookSpoolAsync({ ...data, ingest: retried })).toBe(2)
|
||||
expect(retried.mock.calls.map(([record]) => record.payload)).toEqual(['start', 'tool'])
|
||||
expect(readFileSync(data.path, 'utf8')).toBe('')
|
||||
})
|
||||
|
||||
it('waits for the owner probe before later records and preserves appended records', async () => {
|
||||
const data = fixture()
|
||||
|
||||
@@ -103,7 +103,7 @@ export function readSpoolFile(
|
||||
export type SpoolDrainOptions = {
|
||||
endpointDir: string
|
||||
getPersistedLaunchTokenHash: (paneKey: string) => string | undefined
|
||||
ingest: (record: SpoolRecord) => void
|
||||
ingest: (record: SpoolRecord) => Promise<void> | void
|
||||
now?: number
|
||||
}
|
||||
|
||||
@@ -165,28 +165,8 @@ function matchesLaunchToken(
|
||||
return !expected || launchTokenHash(record.launchToken) === expected
|
||||
}
|
||||
|
||||
export function drainAgentHookSpool(options: SpoolDrainOptions): number {
|
||||
const now = options.now ?? Date.now()
|
||||
let drained = 0
|
||||
for (const candidate of spoolCandidates(options.endpointDir)) {
|
||||
const { records, consumed } = readSpoolFile(candidate.path, now)
|
||||
for (const record of records) {
|
||||
if (matchesLaunchToken(record, options)) {
|
||||
options.ingest({ ...record, isReplay: true })
|
||||
drained += 1
|
||||
}
|
||||
}
|
||||
consumeSpoolPrefix(candidate.path, consumed)
|
||||
}
|
||||
return drained
|
||||
}
|
||||
|
||||
/** Await each admission before consuming a file so owner probes preserve replay order and retry. */
|
||||
export async function drainAgentHookSpoolAsync(
|
||||
options: Omit<SpoolDrainOptions, 'ingest'> & {
|
||||
ingest: (record: SpoolRecord) => Promise<void> | void
|
||||
}
|
||||
): Promise<number> {
|
||||
export async function drainAgentHookSpoolAsync(options: SpoolDrainOptions): Promise<number> {
|
||||
const now = options.now ?? Date.now()
|
||||
let drained = 0
|
||||
for (const candidate of spoolCandidates(options.endpointDir)) {
|
||||
|
||||
@@ -13,7 +13,7 @@ import {
|
||||
} from './agent-session-context-clear'
|
||||
import {
|
||||
currentAgentSessionThreadGoal,
|
||||
currentAgentSessionThreadGoalBySequence
|
||||
currentAgentSessionThreadGoalSourceBySequence
|
||||
} from './agent-session-thread-goal'
|
||||
import {
|
||||
activeStructuredAgentSessionTurnId,
|
||||
@@ -140,15 +140,15 @@ describe('completed context clear projections', () => {
|
||||
|
||||
it('resets the current goal in sorted snapshots and unordered host maps', () => {
|
||||
expect(currentAgentSessionThreadGoal([goalRow, clear])).toBeNull()
|
||||
expect(currentAgentSessionThreadGoalBySequence([clear, goalRow])).toBeNull()
|
||||
expect(currentAgentSessionThreadGoalSourceBySequence([clear, goalRow])?.goal).toBeNull()
|
||||
const newGoal = row(6, {
|
||||
kind: 'status',
|
||||
text: 'Goal set',
|
||||
threadGoal: { state: 'set', goal: { ...goal, objective: 'New goal' } }
|
||||
})
|
||||
expect(currentAgentSessionThreadGoalBySequence([newGoal, goalRow, clear])?.objective).toBe(
|
||||
'New goal'
|
||||
)
|
||||
expect(
|
||||
currentAgentSessionThreadGoalSourceBySequence([newGoal, goalRow, clear])?.goal?.objective
|
||||
).toBe('New goal')
|
||||
expect(currentAgentSessionThreadGoal([goalRow], clear.sequence)).toBeUndefined()
|
||||
})
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ import {
|
||||
agentSessionThreadGoalElapsedSeconds,
|
||||
agentSessionThreadGoalStatusChange,
|
||||
currentAgentSessionThreadGoal,
|
||||
currentAgentSessionThreadGoalBySequence,
|
||||
currentAgentSessionThreadGoalSourceBySequence,
|
||||
isAgentSessionThreadGoalOpen
|
||||
} from './agent-session-thread-goal'
|
||||
|
||||
@@ -75,8 +75,8 @@ describe('current thread goal', () => {
|
||||
['b', goalRow(5, 'set', paused)],
|
||||
['a', goalRow(2, 'set')]
|
||||
])
|
||||
expect(currentAgentSessionThreadGoalBySequence(unordered.values())).toEqual(paused)
|
||||
expect(currentAgentSessionThreadGoalBySequence([])).toBeUndefined()
|
||||
expect(currentAgentSessionThreadGoalSourceBySequence(unordered.values())?.goal).toEqual(paused)
|
||||
expect(currentAgentSessionThreadGoalSourceBySequence([])?.goal).toBeUndefined()
|
||||
})
|
||||
|
||||
it('answers null once the latest transition cleared the goal', () => {
|
||||
|
||||
@@ -88,12 +88,6 @@ export function currentAgentSessionThreadGoalSourceBySequence<T extends GoalCand
|
||||
return latest === null ? undefined : { item: latest, goal: goalFromRow(latest.body) }
|
||||
}
|
||||
|
||||
export function currentAgentSessionThreadGoalBySequence(
|
||||
items: Iterable<GoalCandidate>
|
||||
): AgentJournalThreadGoal | null | undefined {
|
||||
return currentAgentSessionThreadGoalSourceBySequence(items)?.goal
|
||||
}
|
||||
|
||||
/** Goal statuses that still describe work in progress, so readers keep them in view. */
|
||||
export function isAgentSessionThreadGoalOpen(goal: AgentJournalThreadGoal | null): boolean {
|
||||
return goal !== null && goal.status !== 'complete'
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { lstat, mkdtemp, readdir, rename, stat } from 'node:fs/promises'
|
||||
import { posix, win32 } from 'node:path'
|
||||
import { isWindowsAbsolutePathLike } from './cross-platform-path'
|
||||
import { isDefinitiveAbsence } from './definitive-filesystem-absence'
|
||||
import { removeTree } from './windows-transient-lock-removal'
|
||||
|
||||
/**
|
||||
@@ -61,14 +62,6 @@ export function inheritedObjectStore(
|
||||
return alternates ? { inheritedAlternates: alternates } : {}
|
||||
}
|
||||
|
||||
function isMissingFileError(error: unknown): boolean {
|
||||
return (
|
||||
error instanceof Error &&
|
||||
'code' in error &&
|
||||
(error.code === 'ENOENT' || error.code === 'ENOTDIR')
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* The alternates for a quarantined run, or undefined to run unquarantined. Inherited value first,
|
||||
* as Git's own temporary object dirs append. A real store with its own alternates is not
|
||||
@@ -82,7 +75,7 @@ async function quarantineAlternates(objects: GitObjectsDirectory): Promise<strin
|
||||
)
|
||||
const hasAlternatesFile = await stat(alternatesFile).then(
|
||||
() => true,
|
||||
(error: unknown) => !isMissingFileError(error)
|
||||
(error: unknown) => !isDefinitiveAbsence(error)
|
||||
)
|
||||
if (hasAlternatesFile) {
|
||||
return undefined
|
||||
|
||||
@@ -5,7 +5,7 @@ import { join } from 'node:path'
|
||||
import { RelayAgentHookServer } from './agent-hook-server'
|
||||
import type { AgentHookRelayEnvelope } from '../shared/agent-hook-relay'
|
||||
import { makePaneKey } from '../shared/stable-pane-id'
|
||||
import { drainAgentHookSpool } from '../shared/agent-hook-spool'
|
||||
import { drainAgentHookSpoolAsync } from '../shared/agent-hook-spool'
|
||||
import { createHookListenerState } from '../shared/agent-hook-listener/listener-state'
|
||||
import {
|
||||
bindOpenCodeSession,
|
||||
@@ -188,7 +188,7 @@ describe('legacy TUI identity admitted by the execution-host relay', () => {
|
||||
expect(forward.mock.calls[0][0]).toMatchObject({ paneKey: PANE_C, launchToken: 'unknown-old' })
|
||||
})
|
||||
|
||||
it('fences real spool replay before borrowing owner identity and retains legacy metadata', () => {
|
||||
it('fences real spool replay before borrowing owner identity and retains legacy metadata', async () => {
|
||||
const state = createHookListenerState()
|
||||
bindOpenCodeSession(state, 'ses_b', { paneKey: PANE_B, boundAt: 1, basis: 'tui' })
|
||||
trackOpenCodePaneLaunchToken(state, PANE_B, 'creator-live')
|
||||
@@ -212,19 +212,19 @@ describe('legacy TUI identity admitted by the execution-host relay', () => {
|
||||
getAgentLaunchToken: (paneKey: string) => authority.get(paneKey)
|
||||
}
|
||||
const drain = () =>
|
||||
drainAgentHookSpool({
|
||||
drainAgentHookSpoolAsync({
|
||||
endpointDir: dir,
|
||||
getPersistedLaunchTokenHash: () => undefined,
|
||||
ingest: (item) => ingestRelayHookSpoolRecord(item, state, 'remote', host)
|
||||
})
|
||||
for (const launchToken of ['viewer-old', undefined, '']) {
|
||||
writeRecord({ ...record, launchToken })
|
||||
expect(drain()).toBe(1)
|
||||
expect(await drain()).toBe(1)
|
||||
expect(readFileSync(file).length).toBe(0)
|
||||
}
|
||||
expect(forward).not.toHaveBeenCalled()
|
||||
writeRecord({ ...record, launchToken: 'viewer-new' })
|
||||
drain()
|
||||
await drain()
|
||||
expect(forward.mock.calls[0][0]).toMatchObject({ paneKey: PANE_B, launchToken: 'creator-live' })
|
||||
forward.mockClear()
|
||||
writeRecord({
|
||||
@@ -233,10 +233,10 @@ describe('legacy TUI identity admitted by the execution-host relay', () => {
|
||||
opencodeSharedServer: 1,
|
||||
payload: { hook_event_name: 'SessionIdle', sessionID: 'ses_unknown' }
|
||||
})
|
||||
drain()
|
||||
await drain()
|
||||
expect(forward).not.toHaveBeenCalled()
|
||||
writeRecord({ ...record, opencodeMajor: 2 })
|
||||
drain()
|
||||
await drain()
|
||||
expect(forward.mock.calls[0][0]).toMatchObject({ paneKey: PANE_C, launchToken: 'viewer-old' })
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user