fix(cli): keep agent show/read work bounded by the agent they name

This commit is contained in:
Brennan Benson
2026-11-13 18:01:12 -08:00
parent 8e9fef2407
commit 8a3db9db51
27 changed files with 561 additions and 108 deletions
+1 -1
View File
@@ -249,7 +249,7 @@ orca agent read orca_session_id:claude_1234 --limit 10
orca agent read orca_session_id:claude_1234 --cursor <cursor> --json
```
Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`), what Copy Orca Session ID copies on a chat and Copy Agent ID on a terminal. `agent list` lists the agents open in this host's workspaces with their IDs; a terminal agent shows none until it first reports to Orca. A chat keeps its ID across `/clear`; a terminal agent gets a new one. IDs belong to the host that runs the agent: agents on other hosts are not listed, and the output names those hosts with the `--environment` that lists them.
Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`), what Copy Orca Session ID copies. `agent list` lists the agents open in this host's workspaces with their IDs; a terminal agent shows none until it first reports to Orca. A chat keeps its ID across `/clear`; a terminal agent gets a new one. IDs belong to the host that runs the agent: agents on other hosts are not listed, and the output names those hosts with the `--environment` that lists them.
`agent read` reads the conversation, not a screen: a chat's history, or a terminal agent's own transcript. The agent need not be running and reading never starts it. A terminal agent can be read while Orca knows where its transcript is (it is open, or asleep in a workspace). Pass the printed `cursor` back with `--cursor` to read only what came after. To message an agent, use `orca terminal send` or `orca orchestration send`.
+1 -1
View File
@@ -70,7 +70,7 @@ If a launch reply is lost, use the returned `operationId` to re-issue the exact
## Agents by ID
Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`; what Copy Orca Session ID copies on a chat and Copy Agent ID on a terminal). Use it to find and read agents:
Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`; what Copy Orca Session ID copies). Use it to find and read agents:
- `ORCA agent list --json` lists the agents open in this host's workspaces, with `orcaSessionId`, `kind` (chat or terminal), state, workspace and last reply. A terminal agent has no ID until it first reports to Orca. Agents on other hosts are not listed; the output names those hosts and the `--environment` to use.
- `ORCA agent show <id> --json` shows where one agent is open (none if closed, two if resumed in two panes).
+10 -2
View File
@@ -18,6 +18,7 @@ import {
type AgentConversationRow,
type AgentListResult,
type AgentReadResult,
type AgentReadTool,
type AgentShowResult
} from '../../shared/agent-conversations/agent-conversation-contracts'
import { AGENT_STATUS_STATES } from '../../shared/agent-status/contracts/agent-status-types'
@@ -143,16 +144,23 @@ function formatAgentShow(result: AgentShowResult): string {
return result.places.map(formatAgentRow).join('\n\n')
}
function formatTool(tool: AgentReadTool): string {
return `[tool ${tool.name}${tool.state ? `: ${tool.state}` : ''}]`
}
function formatAgentRead(result: AgentReadResult): string {
const lines = result.entries.map((entry) => {
if (entry.kind === 'message') {
return `${entry.role}: ${entry.text}`
return [
...(entry.text ? [`${entry.role}: ${entry.text}`] : []),
...(entry.tools ?? []).map((tool) => formatTool(tool))
].join('\n')
}
if (entry.kind === 'error') {
return `[error] ${entry.text}`
}
if (entry.kind === 'tool') {
return `[tool ${entry.name}${entry.state ? `: ${entry.state}` : ''}]`
return formatTool(entry)
}
return `[${entry.kind}${entry.pending ? ', waiting for an answer' : ''}] ${entry.text}`
})
+1 -1
View File
@@ -298,7 +298,7 @@ const ORCA_CLI_MARKDOWN = [
"",
"## Agents by ID",
"",
"Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`; what Copy Orca Session ID copies on a chat and Copy Agent ID on a terminal). Use it to find and read agents:",
"Every agent, chat or terminal, has an Orca session ID (`orca_session_id:<agent>_<uuid>`; what Copy Orca Session ID copies). Use it to find and read agents:",
"",
"- `ORCA agent list --json` lists the agents open in this host's workspaces, with `orcaSessionId`, `kind` (chat or terminal), state, workspace and last reply. A terminal agent has no ID until it first reports to Orca. Agents on other hosts are not listed; the output names those hosts and the `--environment` to use.",
"- `ORCA agent show <id> --json` shows where one agent is open (none if closed, two if resumed in two panes).",
+1 -1
View File
@@ -22,7 +22,7 @@ export const AGENT_CONVERSATION_COMMAND_SPECS: CommandSpec[] = [
allowedFlags: [...GLOBAL_FLAGS, 'id'],
positionalArgs: ['id'],
notes: [
'<id> is the `orca_session_id:` ID from `orca agent list`, Copy Orca Session ID on a chat or Copy Agent ID on a terminal.',
'<id> is the `orca_session_id:` ID from `orca agent list` or Copy Orca Session ID.',
'An agent resumed in two panes shows both; one that is not open shows none.'
],
examples: ['orca agent show orca_session_id:claude_1234 --json']
@@ -14,6 +14,7 @@ import { isOrcaSessionId, type OrcaSessionId } from '../../../shared/orca-sessio
export class InProcessAgentIdentityClient implements AgentIdentityClient {
identifyRequests = 0
hasRequests = 0
providerKeyRequests = 0
/** When set, `identify` waits for `release()` before answering. */
hold = false
/** Answers the next N identify requests as unavailable after running their SQL. */
@@ -47,6 +48,7 @@ export class InProcessAgentIdentityClient implements AgentIdentityClient {
path: string,
orcaSessionId: string
): Promise<{ key: AgentProviderKey | null } | undefined> {
this.providerKeyRequests++
const response = handleAgentIdentityRequest({
id: 0,
kind: 'providerKeyOf',
@@ -117,11 +117,12 @@ export class AgentIdentityDemand {
}
/** Membership shares the failure gate: during the backoff it answers unavailable at once. */
has = (id: string) => this.lookup((client, path) => client.has(path, id))
has = (id: string) => this.lookup(`has\0${id}`, (client, path) => client.has(path, id))
/** An explicit read-only question under the store's one-at-a-time queue and failure gate;
* undefined when the store cannot answer now. A sent question leaves `lookups`. */
/** An explicit read-only question under the store's one-at-a-time queue and failure gate,
* shared by simultaneous callers of one `key`; undefined when the store cannot answer now. */
async lookup<T>(
key: string,
run: (client: AgentIdentityClient, path: string) => Promise<T | undefined>
): Promise<T | undefined> {
this.syncAuthority()
@@ -129,9 +130,11 @@ export class AgentIdentityDemand {
return undefined
}
const generation = this.generationValue
const answer = this.lookups.enqueue(run)
const pending = this.lookups.enqueue(key, run)
this.schedulePump()
return generation === this.generationValue ? await answer : undefined
const answer = await pending
// An answer accepted before the owner closed is not this generation's to give.
return generation === this.generationValue ? answer : undefined
}
/** A different store path is a different authority: nothing carries across. */
@@ -1,49 +1,73 @@
import type { AgentIdentityClient } from './agent-identity-worker-client'
/** One explicit read-only question: `send` resolves true when the store could not answer. */
/** One explicit read-only question, shared by every caller asking it: `send` resolves true when
* the store could not answer. */
export type AgentIdentityLookup = {
send: (client: AgentIdentityClient, path: string) => Promise<boolean>
cancel: () => void
}
/** Questions for the identity store's one-at-a-time queue; each caller waits at most `waitMs`,
* and a question already sent may finish after its caller stopped waiting. */
type PendingLookup = {
lookup: AgentIdentityLookup
waiters: Set<(value: unknown) => void>
sent: boolean
}
/** Questions for the identity store's one-at-a-time queue, keyed so simultaneous callers of one
* question share one worker operation. Each caller waits at most `waitMs`; a question already
* sent may finish after its callers stopped waiting. Nothing is kept once answered. */
export class AgentIdentityLookupQueue {
/** Callers still waiting, in arrival order; a sent question stays until it is answered. */
private readonly waiting = new Set<AgentIdentityLookup>()
private readonly sent = new WeakSet<AgentIdentityLookup>()
/** Questions with callers still waiting, in arrival order; a sent one stays until answered. */
private readonly pending = new Map<string, PendingLookup>()
constructor(private readonly waitMs: number) {}
/** `key` names the question: one key always runs the same `run`, so it answers with `T`. */
enqueue<T>(
key: string,
run: (client: AgentIdentityClient, path: string) => Promise<T | undefined>
): Promise<T | undefined> {
let entry = this.pending.get(key)
if (!entry) {
const created: PendingLookup = {
waiters: new Set(),
sent: false,
lookup: {
send: async (client, path) => {
const value = await run(client, path)
this.answer(key, created, value)
return value === undefined
}
}
}
entry = created
this.pending.set(key, entry)
}
const shared = entry
return new Promise((resolve) => {
const done = (value: T | undefined): void => {
const waiter = (value: unknown): void => {
clearTimeout(timer)
this.waiting.delete(lookup)
resolve(value)
shared.waiters.delete(waiter)
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a key always names the same question, so its answer is this caller's `T`.
resolve(value as T | undefined)
}
const lookup: AgentIdentityLookup = {
send: async (client, path) => {
const value = await run(client, path)
done(value)
return value === undefined
},
cancel: () => done(undefined)
}
const timer = setTimeout(lookup.cancel, this.waitMs)
const timer = setTimeout(() => {
waiter(undefined)
// An unsent question nobody waits for is never sent.
if (!shared.sent && shared.waiters.size === 0 && this.pending.get(key) === shared) {
this.pending.delete(key)
}
}, this.waitMs)
timer.unref()
this.waiting.add(lookup)
shared.waiters.add(waiter)
})
}
/** The oldest question not yet sent, marked sent. */
next(): AgentIdentityLookup | undefined {
for (const lookup of this.waiting) {
if (!this.sent.has(lookup)) {
this.sent.add(lookup)
return lookup
for (const entry of this.pending.values()) {
if (!entry.sent) {
entry.sent = true
return entry.lookup
}
}
return undefined
@@ -51,8 +75,17 @@ export class AgentIdentityLookupQueue {
/** An ended store authority: every waiting caller gets no answer now. */
cancelAll(): void {
for (const lookup of this.waiting) {
lookup.cancel()
for (const [key, entry] of this.pending) {
this.answer(key, entry, undefined)
}
}
private answer(key: string, entry: PendingLookup, value: unknown): void {
if (this.pending.get(key) === entry) {
this.pending.delete(key)
}
for (const waiter of entry.waiters) {
waiter(value)
}
}
}
@@ -30,11 +30,15 @@ CREATE TABLE IF NOT EXISTS provider_links(
const SELECT_LINK = `SELECT orca_session_id FROM provider_links
WHERE execution_host = ? AND wsl_distro = ? AND agent = ? AND provider_session_id = ? AND locator = ?`
// Reverse lookups for `orca agent` reads; indexes only, so older builds open the file unchanged.
// Reverse lookups for `orca agent` reads: indexes only, so older builds open the file unchanged.
const LOOKUP_INDEX_SQL = `CREATE INDEX IF NOT EXISTS provider_links_by_orca_session
ON provider_links(orca_session_id);
CREATE INDEX IF NOT EXISTS provider_links_by_provider_session
ON provider_links(provider_session_id);`
const LOOKUP_INDEX_NAMES = [
'provider_links_by_orca_session',
'provider_links_by_provider_session'
] as const
const PROVIDER_SESSION_ID_MATCH_LIMIT = 8
/** Orca ids for provider conversations, which nothing can rebuild: the file is never deleted, and
@@ -45,6 +49,9 @@ export class AgentIdentityStore {
/** A newer build's file: existing ids resolve, nothing is minted. */
readonly readOnly: boolean
/** Reverse lookups run only on their indexes, never as a scan of every link. */
private lookupIndexed = false
constructor(path: string, busyTimeoutMs = 25) {
const opened = openIdentityDatabase(path, busyTimeoutMs)
this.db = opened.db
@@ -52,6 +59,26 @@ export class AgentIdentityStore {
hardenSqliteDatabaseFiles(path)
}
/** Adds the lookup indexes on first use; a busy or read-only file throws, so the caller's
* failure backoff retries it rather than scanning. */
private requireLookupIndexes(): void {
if (this.lookupIndexed) {
return
}
if (!this.readOnly) {
this.db.exec(LOOKUP_INDEX_SQL)
}
const present = this.db
.prepare(
`SELECT COUNT(*) AS count FROM sqlite_master WHERE type = 'index' AND name IN (${LOOKUP_INDEX_NAMES.map(() => '?').join(', ')})`
)
.get(...LOOKUP_INDEX_NAMES)
if (Number(present?.count) !== LOOKUP_INDEX_NAMES.length) {
throw new Error('agent identity lookup indexes unavailable')
}
this.lookupIndexed = true
}
lookup(key: AgentProviderKey): OrcaSessionId | null {
const row = this.db.prepare(SELECT_LINK).get(...keyBindings(key))
const id = row?.orca_session_id
@@ -105,6 +132,7 @@ export class AgentIdentityStore {
/** The provider conversation `id` names: each mint links exactly one key. */
providerKeyOf(id: string): AgentProviderKey | null {
this.requireLookupIndexes()
const row = this.db
.prepare(
`SELECT execution_host, wsl_distro, agent, provider_session_id, locator
@@ -124,6 +152,7 @@ export class AgentIdentityStore {
/** Orca ids linked to a provider's own session id, so a caller can be told which to use. */
idsOfProviderSession(providerSessionId: string): OrcaSessionId[] {
this.requireLookupIndexes()
return this.db
.prepare(
`SELECT DISTINCT orca_session_id FROM provider_links WHERE provider_session_id = ? LIMIT ${PROVIDER_SESSION_ID_MATCH_LIMIT}`
@@ -176,12 +205,6 @@ function openIdentityDatabase(
throw error
}
}
try {
db.exec(LOOKUP_INDEX_SQL)
} catch (error) {
// A busy file still answers lookups, by scan; the next open adds the indexes.
console.warn('[agent-hooks] agent identity lookup indexes deferred', error)
}
return { db, readOnly: false }
} catch (error) {
db.close()
@@ -9,6 +9,7 @@ import { isOrcaSessionId } from '../../../shared/orca-session-address'
import SyncDatabase from '../../sqlite/sync-database'
import { spawnWorkerThread } from '../../worker-threads/worker-thread-spawn'
import { AgentIdentityWorkerClient } from './agent-identity-worker-client'
import { AgentIdentityStore } from './agent-identity-worker-entry'
import {
InProcessAgentIdentityClient,
ScriptedIdentityWorkers
@@ -414,3 +415,56 @@ it('answers unavailable when the worker cannot load, or is stopped while loading
// No worker outlives its owner.
await vi.waitFor(() => expect(workers.every(exited)).toBe(true))
})
describe('PaneAgentIdentities explicit lookups', () => {
it.each([12, 120])(
'asks the store once for %i simultaneous callers of one id',
async (callers) => {
const client = new InProcessAgentIdentityClient()
const { identities } = inProcessService(databasePath(), client)
const id = await identities.identify(sighting({ id: 'dormant' }))
client.hasRequests = 0
const answers = await Promise.all(
Array.from({ length: callers }, () => identities.isTerminalIdentity(id!))
)
expect(answers.every((known) => known === true)).toBe(true)
expect(client.hasRequests).toBe(1)
}
)
it('answers reverse lookups only through their indexes, once a busy writer lets them be built', () => {
const path = databasePath()
const store = new AgentIdentityStore(path)
opened.push(store)
const id = store.identify({
executionHost: 'local',
wslDistro: '',
agent: 'claude',
providerSessionId: 'session-a',
locator: ''
})
const writer = new SyncDatabase(path)
opened.push(writer)
writer.exec('BEGIN IMMEDIATE')
// Never a scan of every link: a store that cannot build the indexes cannot answer yet.
expect(() => store.providerKeyOf(id!)).toThrow()
writer.exec('COMMIT')
expect(store.providerKeyOf(id!)).toMatchObject({ providerSessionId: 'session-a' })
expect(store.idsOfProviderSession('session-a')).toEqual([id])
const plans = [
'SELECT locator FROM provider_links WHERE orca_session_id = ?',
'SELECT DISTINCT orca_session_id FROM provider_links WHERE provider_session_id = ?'
].map((sql) =>
writer
.prepare(`EXPLAIN QUERY PLAN ${sql}`)
.all('x')
.map((row) => String(row.detail))
.join('; ')
)
expect(plans[0]).toContain('USING INDEX provider_links_by_orca_session')
expect(plans[1]).toContain('USING INDEX provider_links_by_provider_session')
})
})
@@ -168,11 +168,13 @@ export class PaneAgentIdentities {
/** The provider conversation this host linked `id` to: `{ key: null }` for none, undefined
* when the store cannot answer now. An explicit lookup, never per hook. */
async providerKeyOf(id: string): Promise<{ key: AgentProviderKey | null } | undefined> {
return this.demand.lookup((client, path) => client.providerKeyOf(path, id))
return this.demand.lookup(`providerKeyOf\0${id}`, (client, path) =>
client.providerKeyOf(path, id)
)
}
async idsOfProviderSession(providerSessionId: string): Promise<OrcaSessionId[] | undefined> {
return this.demand.lookup((client, path) =>
return this.demand.lookup(`idsOfProviderSession\0${providerSessionId}`, (client, path) =>
client.idsOfProviderSession(path, providerSessionId)
)
}
@@ -196,17 +196,20 @@ export abstract class AgentHookServerAgentIdentities extends AgentHookServerStat
/** Where terminal agent `id`'s transcript is: a current row carrying the id, in any state (a read
* needs no running agent), else a sleeping record naming its provider key. The session id and
* path come from one payload, never from a pane, so a row attributed to the wrong pane still
* names its own conversation. null: nothing names it now; undefined: the store cannot answer. */
* names its own conversation. null: nothing names it now; undefined: the store cannot answer.
* `knownKey` is the key an earlier answer gave: an id is never relinked, so a recheck needs no
* second store question. */
async terminalAgentTranscriptSource(
id: string,
listSleeping: () => readonly SleepingAgentSighting[]
readSleeping: () => Readonly<Record<string, SleepingAgentSighting>>,
knownKey?: AgentProviderKey
): Promise<TerminalAgentTranscriptSource | null | undefined> {
const authority = this.agentIdentities.authority()
const live = this.liveTranscriptSource(id)
if (live) {
return live
}
const linked = await this.agentIdentities.providerKeyOf(id)
const linked = knownKey ? { key: knownKey } : await this.agentIdentities.providerKeyOf(id)
if (!linked || authority !== this.agentIdentities.generation) {
return undefined
}
@@ -215,7 +218,10 @@ export abstract class AgentHookServerAgentIdentities extends AgentHookServerStat
if (!key || process.platform === 'win32') {
return null
}
for (const record of listSleeping()) {
for (const record of Object.values(readSleeping())) {
if (record.providerSession.id !== key.providerSessionId) {
continue
}
const recordKey = agentProviderKeyOf({
agent: record.agent,
providerSession: record.providerSession,
@@ -20,8 +20,10 @@ afterEach(() => {
})
class IdentityTestServer extends AgentHookServer {
readonly identityClient = new InProcessAgentIdentityClient()
protected override createAgentIdentityClient(): AgentIdentityClient {
return new InProcessAgentIdentityClient()
return this.identityClient
}
}
@@ -70,12 +72,12 @@ it('reads a terminal agent by its own conversation, never by the pane it was att
expect(second).toMatch(/^claude_/)
expect(second).not.toBe(first)
expect(await server.terminalAgentTranscriptSource(second!, () => [])).toMatchObject({
expect(await server.terminalAgentTranscriptSource(second!, () => ({}))).toMatchObject({
key: { wslDistro: 'Ubuntu', agent: 'claude', providerSessionId: 'session-b' },
providerSession: { id: 'session-b', transcriptPath: '/home/u/.claude/projects/p/b.jsonl' }
})
// The first conversation's pane now shows the second: it has no transcript location, not b's.
expect(await server.terminalAgentTranscriptSource(first!, () => [])).toBeNull()
expect(await server.terminalAgentTranscriptSource(first!, () => ({}))).toBeNull()
expect(await server.terminalAgentIdsOfProviderSession('session-a')).toEqual([first])
})
@@ -97,15 +99,23 @@ it.skipIf(process.platform === 'win32')(
connectionId: WSL
})
expect(await server.terminalAgentTranscriptSource(id!, () => [sleeping('other')])).toBeNull()
expect(
await server.terminalAgentTranscriptSource(id!, () => [
sleeping('other'),
sleeping('session-a')
])
).toMatchObject({
await server.terminalAgentTranscriptSource(id!, () => ({ other: sleeping('other') }))
).toBeNull()
let partitionReads = 0
const records = () => {
partitionReads++
return { other: sleeping('other'), a: sleeping('session-a') }
}
server.identityClient.providerKeyRequests = 0
const found = await server.terminalAgentTranscriptSource(id!, records)
expect(found).toMatchObject({
key: { wslDistro: 'Ubuntu', providerSessionId: 'session-a' },
providerSession: { transcriptPath: '/home/u/.claude/projects/p/session-a.jsonl' }
})
// The post-read recheck reuses the key it was given: one partition read, no store question.
expect(await server.terminalAgentTranscriptSource(id!, records, found!.key)).toEqual(found)
expect(server.identityClient.providerKeyRequests).toBe(1)
expect(partitionReads).toBe(2)
}
)
@@ -57,6 +57,8 @@ export function createStructuredAgentSessionConversationReads(context: () => Rea
return {
journalCurrentRead: async (sessionId: string, inputIds?: readonly string[]) =>
(await context().conversation(sessionId)).journal.currentRead(inputIds),
/** The journal's head, opened at rest: a reader skips an unchanged history. */
journalCursor: async (sessionId: string) => (await journalOf(sessionId)).cursor(),
readModelCatalog,
attentionReads: {
@@ -145,7 +145,7 @@ export class StructuredAgentSessionHost {
hasSession = (sessionId: string): boolean => this.sessions.has(sessionId)
loadedSessionIds = this.sessions.loadedSessionIds
hasLoadedProviders = (): boolean => this.loadedSessionIds().length > 0
private readonly reads = createStructuredAgentSessionConversationReads(() => ({
readonly reads = createStructuredAgentSessionConversationReads(() => ({
deps: this.deps,
sessions: this.sessions,
serialize: this.serialize,
@@ -61,7 +61,9 @@ export async function readNativeChatTranscriptTailFile(
decodeLifecycle?: NativeChatTurnLifecycleDecoder | null,
signal?: AbortSignal,
onConsumedBoundary?: (boundary: Buffer) => void,
sourceOrder = false
sourceOrder = false,
/** Stop looking back after this many bytes, so a tail with few messages costs a bounded read. */
maxScanBytes = Number.POSITIVE_INFINITY
): Promise<{
messages: NativeChatMessage[]
lifecycle?: NativeChatTurnLifecycle
@@ -109,9 +111,10 @@ export async function readNativeChatTranscriptTailFile(
if (cursor === 0) {
onConsumedBoundary?.(finalBoundaryByte)
}
while (cursor > 0 && newestFirst.length <= limit) {
const scanFloor = Math.max(0, cursor - maxScanBytes)
while (cursor > scanFloor && newestFirst.length <= limit) {
signal?.throwIfAborted()
const start = Math.max(0, cursor - TAIL_CHUNK_BYTES)
const start = Math.max(scanFloor, cursor - TAIL_CHUNK_BYTES)
const buffer = Buffer.allocUnsafe(cursor - start)
const { bytesRead } = await readTranscriptFile(
handle,
@@ -168,7 +171,8 @@ export async function readNativeChatTranscriptTailFile(
messages: selected.map((entry) => entry.message),
...(lifecycle ? { lifecycle } : {}),
consumedTo,
hasMore: limit > 0 && chronological.length > limit,
// A scan stopped at its byte bound may have left older messages unread.
hasMore: limit > 0 && (chronological.length > limit || (cursor > 0 && cursor === scanFloor)),
beforeOffset: selected[0]?.offset ?? end,
...(Number.isFinite(latest) ? { latestTimestamp: latest } : {}),
...(malformedRecordCount > 0 ? { malformedRecordCount } : {}),
@@ -0,0 +1,48 @@
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, expect, it } from 'vitest'
import { readNativeChatTranscriptTailFile } from './transcript-tail-reader'
const directories: string[] = []
afterEach(() => {
for (const directory of directories.splice(0)) {
rmSync(directory, { recursive: true, force: true })
}
})
// 128-byte records an agent writes between messages, none of them displayable.
const PROGRESS = `${JSON.stringify({ type: 'progress', data: 'x'.repeat(98) })}\n`
it.each([2, 20])(
'looks back a bounded number of bytes over a %i× larger tail with no messages',
async (multiple) => {
const directory = mkdtempSync(join(tmpdir(), 'transcript-tail-scan-'))
directories.push(directory)
const filePath = join(directory, 'session.jsonl')
const scanBytes = 64 * 1024
writeFileSync(filePath, PROGRESS.repeat((multiple * scanBytes) / PROGRESS.length))
let decoded = 0
const page = await readNativeChatTranscriptTailFile(
filePath,
20,
() => {
decoded++
return null
},
false,
undefined,
undefined,
undefined,
undefined,
false,
scanBytes
)
expect(PROGRESS.length).toBe(128)
// The last 64 KiB of records, however long the file.
expect(decoded).toBe(512)
expect(page).toMatchObject({ messages: [], hasMore: true })
}
)
@@ -84,7 +84,7 @@ export async function resolveAgentConversation(
)
: new AgentConversationRefusal(
CODES.invalidId,
`${text || 'That'} is not an Orca session ID. Find agents' IDs with \`orca agent list\`, or copy one with Copy Orca Session ID (Copy Agent ID on a terminal).`
`${text || 'That'} is not an Orca session ID. Find agents' IDs with \`orca agent list\`, or copy one with Copy Orca Session ID.`
)
}
@@ -0,0 +1,57 @@
import { expect, it, vi } from 'vitest'
import { agentConversationPlaces } from './agent-conversation-list'
import { testOrcaSessionId } from '../../../shared/orca-session-address-test-fixture'
const TERMINAL = testOrcaSessionId('claude_5688af50_12a0_4b0b_9a4a_098d87fe1e68')
const hookServer = vi.hoisted(() => ({
terminalAgentPlaces: vi.fn(),
getPublicStatusSnapshotForPane: vi.fn()
}))
vi.mock('../../agent-hooks/server/agent-hook-server', () => ({ agentHookServer: hookServer }))
function status(paneKey: string, orcaSessionId: string) {
return {
paneKey,
worktreeId: 'repo::/workspace',
orcaSessionId,
agentType: 'claude',
state: 'working',
receivedAt: 5
}
}
it.each([750, 7_500])(
'shows one agent from its own panes, whatever the other %i panes run',
async (unrelatedPanes) => {
const panes = ['tab-a:1', 'tab-b:1']
hookServer.terminalAgentPlaces.mockResolvedValue(panes.map((paneKey) => ({ paneKey })))
hookServer.getPublicStatusSnapshotForPane.mockReset()
const rows = new Map(
Array.from({ length: unrelatedPanes }, (_, index) => [
`tab-${index}:1`,
[status(`tab-${index}:1`, `claude_other_${index}`)]
])
)
for (const paneKey of panes) {
rows.set(paneKey, [status(paneKey, TERMINAL)])
}
hookServer.getPublicStatusSnapshotForPane.mockImplementation(
(paneKey: string) => rows.get(paneKey) ?? []
)
const runtime = {
getWorkspaceDisplayName: vi.fn(() => 'Workspace'),
getWorktreePs: vi.fn()
}
const places = await agentConversationPlaces(runtime, {
kind: 'terminal',
orcaSessionId: TERMINAL
})
expect(places.map((place) => place.paneKey)).toEqual(panes)
expect(places[0]).toMatchObject({ workspace: 'Workspace', state: 'working', kind: 'terminal' })
expect(hookServer.getPublicStatusSnapshotForPane).toHaveBeenCalledTimes(2)
expect(runtime.getWorkspaceDisplayName).toHaveBeenCalledTimes(2)
expect(runtime.getWorktreePs).not.toHaveBeenCalled()
}
)
@@ -5,6 +5,13 @@ import type {
import type { AgentStatusState } from '../../../shared/agent-status/contracts/agent-status-types'
import type { RuntimeWorktreePsResult } from '../../../shared/runtime-worktree-contracts'
import { structuredAgentSessionIdFromTabId } from '../../../shared/agent-session/view/structured-agent-session-tab-id'
import {
structuredAgentSessionPaneKey,
structuredAgentSessionTabId
} from '../../../shared/agent-session/view/structured-agent-session-projection'
import { resolveLineageRunningSession } from '../orchestration/addressing/session-lineage/structured-session-lineage'
import { agentHookServer } from '../../agent-hooks/server/agent-hook-server'
import type { ResolvedAgentConversation } from './agent-conversation-address'
/** `worktree.ps` agent rows as one flat list of chats and terminal agents, oldest state first. */
export function projectAgentConversationList(
@@ -42,3 +49,55 @@ export function projectAgentConversationList(
...(ps.hostScope ? { hostScope: ps.hostScope } : {})
}
}
/** Where one resolved conversation is open now, from its own keyed status rows: never a listing
* of every workspace. A closed chat or a terminal agent with no live pane has none. */
export async function agentConversationPlaces(
runtime: { getWorkspaceDisplayName(worktreeId: string): string | null },
resolved: ResolvedAgentConversation
): Promise<AgentConversationRow[]> {
const { orcaSessionId } = resolved
let paneKeys: string[]
if (resolved.kind === 'chat') {
// A chat's status row is keyed by the session that runs its conversation now.
const running = resolveLineageRunningSession(resolved.store, resolved.record.sessionId)
paneKeys =
running.kind === 'here'
? [
structuredAgentSessionPaneKey(
structuredAgentSessionTabId(running.sessionId),
running.sessionId
)
]
: []
} else {
paneKeys = ((await agentHookServer.terminalAgentPlaces(orcaSessionId)) ?? []).map(
(place) => place.paneKey
)
}
return paneKeys.flatMap((paneKey) =>
agentHookServer.getPublicStatusSnapshotForPane(paneKey).flatMap((status) => {
const { worktreeId } = status
const ours =
resolved.kind === 'chat'
? status.structuredHost !== undefined
: status.orcaSessionId === orcaSessionId
return worktreeId && ours
? [
{
orcaSessionId,
kind: resolved.kind,
agent: status.agentType ?? null,
state: status.state,
worktreeId,
workspace: runtime.getWorkspaceDisplayName(worktreeId) ?? worktreeId,
paneKey: status.paneKey,
title: null,
lastAssistantMessage: status.lastAssistantMessage ?? null,
updatedAt: status.evidenceObservedAt ?? status.receivedAt
}
]
: []
})
)
}
@@ -0,0 +1,93 @@
import { beforeEach, expect, it, vi } from 'vitest'
import type { RpcContext } from '../rpc/core'
import { readAgentConversation } from './agent-conversation-read'
import type { ResolvedAgentConversation } from './agent-conversation-address'
const SESSION = 'codex_0b8f2a4e_6c1d_4f7a_9e3b_5a2c8d1f0e47'
const host = vi.hoisted(() => ({
reads: { journalCursor: vi.fn() },
history: vi.fn()
}))
vi.mock('../rpc/methods/structured-sessions/creation/structured-agent-session-gate', () => ({
requireStructuredAgentAudience: () => undefined,
requireInstalledStructuredHost: async () => host
}))
vi.mock('../orchestration/addressing/session-lineage/structured-session-lineage', () => ({
resolveLineageRunningSession: () => ({ kind: 'here', sessionId: SESSION })
}))
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the chat read reads only the record's id and agent.
const chat = {
kind: 'chat',
orcaSessionId: SESSION,
record: { sessionId: SESSION, provider: 'codex' },
store: {}
} as unknown as ResolvedAgentConversation
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the chat branch passes the context only to the mocked gates.
const ctx = {} as RpcContext
function message(sequence: number) {
return {
itemId: `item-${sequence}`,
sequence,
body: {
kind: 'message',
role: 'assistant',
blocks: [{ type: 'text', text: `reply ${sequence}` }]
}
}
}
function page(items: ReturnType<typeof message>[], newest: number, more: boolean) {
return {
ok: true,
page: {
items,
hasOlder: true,
hasNewer: more,
window: { oldest: null, newest: null, nextCursor: { epoch: 'e', sequence: newest } },
liveCursor: { epoch: 'e', sequence: newest }
}
}
}
beforeEach(() => {
host.history.mockReset()
host.reads.journalCursor.mockReset()
})
it.each([20_000, 200_000])(
'follows an unchanged chat of %i rows with no history read',
async (rows) => {
host.reads.journalCursor.mockResolvedValue({ epoch: 'e', sequence: rows })
host.history.mockResolvedValue(page([message(rows)], rows, false))
const first = await readAgentConversation(ctx, chat, { limit: 1 })
host.history.mockClear()
for (let read = 0; read < 10; read++) {
const again = await readAgentConversation(ctx, chat, { cursor: first.cursor! })
expect(again).toMatchObject({ entries: [], cursor: first.cursor })
}
expect(host.history).not.toHaveBeenCalled()
}
)
it('gives a cursor read at most --limit entries and continues where it stopped', async () => {
host.reads.journalCursor.mockResolvedValue({ epoch: 'e', sequence: 300 })
host.history.mockResolvedValueOnce(page([], 100, false))
const start = await readAgentConversation(ctx, chat, { limit: 1 })
host.history.mockResolvedValueOnce(page([message(101)], 101, true))
const next = await readAgentConversation(ctx, chat, { cursor: start.cursor!, limit: 1 })
expect(host.history).toHaveBeenLastCalledWith({
sessionId: SESSION,
direction: 'after',
cursor: { epoch: 'e', sequence: 100 },
limit: 1
})
expect(next.entries).toEqual([{ kind: 'message', role: 'assistant', text: 'reply 101' }])
expect(next.limited).toBe(true)
})
@@ -5,6 +5,7 @@ import {
type AgentReadResult
} from '../../../shared/agent-conversations/agent-conversation-contracts'
import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session/agent-session-wire'
import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity'
import type { AgentJournalRenderItem } from '../../../shared/agent-session/journal/agent-session-journal-types'
import { isRootAgentJournalItem } from '../../../shared/agent-session/journal/agent-session-journal-producer'
import { agentSessionPromptQuestions } from '../../../shared/agent-session/agent-session-question-answer'
@@ -23,7 +24,10 @@ import {
import { createWorkerOutputSourceIdentity } from '../orchestration/workers/transcripts/worker-output-cursor'
import { agentHookServer } from '../../agent-hooks/server/agent-hook-server'
import type { TerminalAgentTranscriptSource } from '../../agent-hooks/server/server-agent-identities'
import { sameProviderKey } from '../../agent-hooks/identity/agent-provider-key'
import {
sameProviderKey,
type AgentProviderKey
} from '../../agent-hooks/identity/agent-provider-key'
import type { ResolvedAgentConversation } from './agent-conversation-address'
import { AgentConversationRefusal } from './agent-conversation-refusal'
import {
@@ -78,18 +82,39 @@ async function readChat(
throw sourceChanged(orcaSessionId)
}
const host = await requireInstalledStructuredHost(ctx, sessionId)
const encode = (journal: AgentJournalCursor): string =>
encodeReadCursor({ v: 1, id: orcaSessionId, s: 'chat', session: sessionId, journal })
// A follower re-reading with its cursor costs no history work until the journal moves.
const head = cursor ? await host.reads.journalCursor(sessionId) : null
if (
cursor &&
head &&
head.epoch === cursor.journal.epoch &&
head.sequence <= cursor.journal.sequence
) {
return {
orcaSessionId,
kind: 'chat',
agent: record.provider,
source: 'chat',
entries: [],
cursor: encode(cursor.journal),
limited: false
}
}
// A cursor read takes at most `limit` journal rows, each at most one entry, and continues
// from the last row it took, so none are skipped or repeated.
const history = await host.history({
sessionId,
direction: cursor ? 'after' : 'tail',
...(cursor ? { cursor: cursor.journal } : {}),
limit: AGENT_SESSION_HISTORY_MAX_LIMIT
limit: cursor ? limit : AGENT_SESSION_HISTORY_MAX_LIMIT
})
if (!history.ok && cursor) {
throw sourceChanged(orcaSessionId)
}
const { page } = history
const all = chatEntries(page.items, page.actionablePromptIds)
// A first read is the latest entries; a cursor read is everything after it, so none are skipped.
const entries = cursor ? all : all.slice(-limit)
const next = cursor ? page.window.nextCursor : (page.liveCursor ?? page.window.newest)
return {
@@ -98,9 +123,7 @@ async function readChat(
agent: record.provider,
source: 'chat',
entries,
cursor: next
? encodeReadCursor({ v: 1, id: orcaSessionId, s: 'chat', session: sessionId, journal: next })
: null,
cursor: next ? encode(next) : null,
limited: cursor ? page.hasNewer : page.hasOlder || all.length > limit
}
}
@@ -114,9 +137,13 @@ async function readTerminalTranscript(
if (cursor && cursor.s !== 'transcript') {
throw invalidCursor()
}
const locate = (): Promise<TerminalAgentTranscriptSource | null | undefined> =>
agentHookServer.terminalAgentTranscriptSource(orcaSessionId, () =>
ctx.runtime.listSleepingAgentSessionRecords()
const locate = (
knownKey?: AgentProviderKey
): Promise<TerminalAgentTranscriptSource | null | undefined> =>
agentHookServer.terminalAgentTranscriptSource(
orcaSessionId,
() => ctx.runtime.readLocalSleepingAgentSessions(),
knownKey
)
const source = await locate()
if (source === undefined) {
@@ -165,7 +192,7 @@ async function readTerminalTranscript(
transcript.sourceFingerprint
])
// The conversation must still name the file just read: its row may have moved on meanwhile.
const after = await locate()
const after = await locate(key)
if (
(cursor?.s === 'transcript' && cursor.i !== sourceIdentity) ||
!after ||
@@ -245,6 +272,8 @@ function chatEntries(
return entries
}
/** One entry per transcript message, its tool calls folded in, so `limit` messages are at most
* `limit` entries and the cursor after them skips nothing. */
function transcriptEntries(messages: readonly NativeChatMessage[]): AgentReadEntry[] {
const entries: AgentReadEntry[] = []
for (const message of messages) {
@@ -256,17 +285,18 @@ function transcriptEntries(messages: readonly NativeChatMessage[]): AgentReadEnt
.filter((part) => part.length > 0)
.join('\n')
.trim()
if (text.length > 0) {
entries.push({ kind: 'message', role: message.role, text })
}
for (const block of message.blocks) {
if (block.type === 'tool-call') {
entries.push({
kind: 'tool',
name: block.name,
...(block.state ? { state: block.state } : {})
})
}
const tools = message.blocks.flatMap((block) =>
block.type === 'tool-call'
? [{ name: block.name, ...(block.state ? { state: block.state } : {}) }]
: []
)
if (text.length > 0 || tools.length > 0) {
entries.push({
kind: 'message',
role: message.role,
text,
...(tools.length > 0 ? { tools } : {})
})
}
}
return entries
@@ -51,7 +51,7 @@ export type StructuredAttentionMobileDelivery = {
/** Read display labels from persisted host metadata without disk/Git lookups. */
export function readStructuredAttentionWorkspaceLabels(
store: Pick<RuntimeStore, 'getRepo' | 'getWorktreeMeta' | 'getFolderWorkspaces'>,
scope: AgentSessionExecutionLocation
scope: Pick<AgentSessionExecutionLocation, 'workspaceId'>
): StructuredAttentionWorkspaceLabels {
const folderScope = parseWorkspaceKey(scope.workspaceId)
if (folderScope?.type === 'folder') {
@@ -48,7 +48,18 @@ export async function readInitialLocalWorkerTranscriptPage(
if (!before) {
return { ok: false, reason: 'transcript_unreadable', warnings: [] }
}
const page = await readNativeChatTranscriptTailFile(filePath, limit, decode, false)
const page = await readNativeChatTranscriptTailFile(
filePath,
limit,
decode,
false,
undefined,
undefined,
undefined,
undefined,
false,
MAX_TRANSCRIPT_SCAN_BYTES
)
const after = await readLocalTranscriptSourceIdentity(filePath)
if (workerTranscriptSourceChanged(before, after, page.consumedTo)) {
return sourceChanged()
@@ -9,7 +9,10 @@ import type {
AgentReadResult,
AgentShowResult
} from '../../../../../shared/agent-conversations/agent-conversation-contracts'
import { projectAgentConversationList } from '../../../agent-conversations/agent-conversation-list'
import {
agentConversationPlaces,
projectAgentConversationList
} from '../../../agent-conversations/agent-conversation-list'
import { resolveAgentConversation } from '../../../agent-conversations/agent-conversation-address'
import { readAgentConversation } from '../../../agent-conversations/agent-conversation-read'
@@ -32,13 +35,10 @@ export const AGENT_CONVERSATION_METHODS = [
params: AgentShowParams,
handler: async (params, { runtime }): Promise<AgentShowResult> => {
const resolved = await resolveAgentConversation(runtime, params.id)
const { agents } = projectAgentConversationList(
await runtime.getWorktreePs(AGENT_LIST_WORKTREE_LIMIT)
)
return {
orcaSessionId: resolved.orcaSessionId,
kind: resolved.kind,
places: agents.filter((row) => row.orcaSessionId === resolved.orcaSessionId)
places: await agentConversationPlaces(runtime, resolved)
}
}
}),
@@ -1,8 +1,9 @@
import { crewWorkspaceNames } from '../../orchestration/crew/crew-workspace-names'
import { isDeepStrictEqual } from 'node:util'
import type { ExecutionHostId } from '../../../../shared/execution-host'
import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../../../shared/execution-host'
import type { WorkspaceSessionState } from '../../../../shared/workspace-session/workspace-session-state-types'
import type { SleepingAgentSessionRecord } from '../../../../shared/agent-session/lifecycle/agent-session-resume'
import { readStructuredAttentionWorkspaceLabels } from '../../agent-session/structured-agent-session-mobile-attention'
import { TerminalListHostScope } from '../../terminal/query/terminal-list-host-scope'
export abstract class OrcaRuntimeWithGetRuntimeId extends TerminalListHostScope {
@@ -50,20 +51,23 @@ export abstract class OrcaRuntimeWithGetRuntimeId extends TerminalListHostScope
}
}
/** Every saved sleeping-agent record, for an explicit lookup by conversation (never per hook). */
listSleepingAgentSessionRecords(): SleepingAgentSessionRecord[] {
const sessions = new Set<WorkspaceSessionState>()
for (const worktreeId of this.getKnownWorkspaceSessionWorktreeIds()) {
const session = this.workspaceSessions.get(worktreeId)
if (session) {
sessions.add(session)
}
}
return [...sessions].flatMap((session) =>
Object.values(session.sleepingAgentSessionsByPaneKey ?? {})
/** This host's own saved sleeping-agent records, folder workspaces included: terminal agent
* ids name conversations on this host only, so no other partition or workspace is visited. */
readLocalSleepingAgentSessions(): Readonly<Record<string, SleepingAgentSessionRecord>> {
return (
this.store?.getWorkspaceSession?.(LOCAL_EXECUTION_HOST_ID)?.sleepingAgentSessionsByPaneKey ??
{}
)
}
/** One workspace's display name by id, without listing every workspace. */
getWorkspaceDisplayName(worktreeId: string): string | null {
return this.store
? (readStructuredAttentionWorkspaceLabels(this.store, { workspaceId: worktreeId })
.worktreeLabel ?? null)
: null
}
protected getKnownWorkspaceSessionWorktreeIds(): Set<string> {
return this.workspaceSessions.getKnownWorktreeIds()
}
@@ -30,13 +30,17 @@ export type AgentListResult = {
export type AgentShowResult = {
orcaSessionId: string
kind: AgentConversationKind
/** Where it is open now: none when it isn't open, two for one conversation resumed in two panes. */
/** Where it is open now: none when it isn't open, two for one conversation resumed in two panes.
* Read from the conversation's own status rows, so they carry no title (`list` does). */
places: AgentConversationRow[]
}
export type AgentReadTool = { name: string; state?: string }
export type AgentReadEntry =
| { kind: 'message'; role: string; text: string }
| { kind: 'tool'; name: string; state?: string }
/** `tools`: the tool calls a transcript message made, folded into its one entry. */
| { kind: 'message'; role: string; text: string; tools?: AgentReadTool[] }
| ({ kind: 'tool' } & AgentReadTool)
| { kind: 'approval' | 'question'; text: string; pending: boolean }
| { kind: 'error'; text: string }