mirror of
https://github.com/stablyai/orca.git
synced 2026-09-29 08:03:20 +00:00
fix(runtime): report only live structured sessions in worktree ps
The status feed's `published` map is a broadcast cache, not a roster. It deliberately never retracts — an evicted idle session is still idle, and a reloading renderer must not lose every settled row — so enumerating it lists every session the host has ever opened, and eviction's `forget-session` step deletes the session from the live map while touching nothing else. Reading it as a roster made `worktree ps` report a closed chat forever. The sharp edge is a chat closed while an approval was pending: a deliberate close does not settle a pending prompt, so the retained summary stays `attention`, maps to a `blocked` row, and merges the worktree to `permission` for the whole 30-minute freshness window — on the CLI and on the mobile sidebar it backs. The poller now answers from the sessions the host still holds, intersecting the live map with the retained projections. `subscribe()` and its snapshot are untouched: retention there is the point. Gating on the live session set rather than the visible tabs keeps a headless orchestration worker listed, which is what the agent-facing surface is for. Also folds out two things the enumerator left behind: the working/attention/idle to working/blocked/done mapping now lives once in the shared projection module instead of once per process, which is what actually enforces "the CLI and the GUI cannot disagree"; and the structured row source no longer builds a write-only `payload` behind an `as` cast that compensated for nothing. The structured source construction moves to its own module to keep runtime-worktree-agent-rows.ts clear of the 300-line cap.
This commit is contained in:
@@ -211,8 +211,9 @@ export class StructuredAgentSessionHost {
|
||||
|
||||
listSessionTabs = () => listStructuredAgentSessionTabs(this.sessions)
|
||||
|
||||
/** Last projected status for every structured session, for non-subscribing readers. */
|
||||
readonly publishedStatusSummaries = () => this.statusFeed.publishedSummaries()
|
||||
/** Last projected status for every structured session this host still holds, for non-subscribing
|
||||
* readers. The retained projections of forgotten sessions are deliberately not included. */
|
||||
readonly liveSessionStatusSummaries = () => this.statusFeed.liveSessionSummaries()
|
||||
|
||||
getPersistedVisibleSessionTabIndex(): { present: boolean; sessionIds: string[] } {
|
||||
return this.deps.store.getVisibleSessionTabIndex()
|
||||
|
||||
@@ -523,3 +523,37 @@ describe('StructuredAgentSessionStatusFeed', () => {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
/**
|
||||
* `published` is a broadcast cache, not a roster. It deliberately never retracts — an evicted idle
|
||||
* session is still idle, and a reloading renderer must not lose every settled row — so enumerating
|
||||
* it lists every session this host has ever opened. Eviction's `forget-session` step deletes the
|
||||
* session from the live map and touches nothing else, so a poller has to intersect with that map.
|
||||
*/
|
||||
describe('the polling reader answers from the live sessions, not the retained cache', () => {
|
||||
it('drops an evicted session from the poll while a late subscriber still sees it', async () => {
|
||||
const journal = await openJournal()
|
||||
const sessions = new Map([[SESSION, { journal }]])
|
||||
const { feed } = feedFor(sessions)
|
||||
await journal.appendItem(
|
||||
USER_IDENTITY,
|
||||
{ kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] },
|
||||
{ fence: 1 }
|
||||
)
|
||||
feed.publish(SESSION, journal)
|
||||
expect(feed.liveSessionSummaries().map((summary) => summary.sessionId)).toEqual([SESSION])
|
||||
|
||||
// Exactly what eviction's `forget-session` step does; nothing else touches the feed.
|
||||
sessions.delete(SESSION)
|
||||
|
||||
expect(feed.liveSessionSummaries()).toEqual([])
|
||||
const late: AgentSessionStatusEvent[] = []
|
||||
feed.subscribe({ id: 'list-2', emit: (event) => late.push(event) })
|
||||
expect(late).toEqual([
|
||||
{
|
||||
type: 'snapshot',
|
||||
sessions: [expect.objectContaining({ sessionId: SESSION, status: 'idle' })]
|
||||
}
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -76,13 +76,24 @@ export class StructuredAgentSessionStatusFeed {
|
||||
}
|
||||
|
||||
/**
|
||||
* Every summary this host has projected, for readers that poll instead of subscribing.
|
||||
* Summaries for the sessions this host still holds, for readers that poll instead of subscribing.
|
||||
*
|
||||
* `published` never retracts, so it is a broadcast cache and not a roster: enumerating it lists
|
||||
* every session ever opened here. A caller asking what is running gets the live intersection,
|
||||
* while the retained view a subscriber opens on stays whole.
|
||||
*
|
||||
* Deliberately does NOT re-project: a subscriber's snapshot is the live read, and re-running the
|
||||
* journal reduction per caller would make an enumerating command pay for every session it lists.
|
||||
*/
|
||||
publishedSummaries(): AgentSessionStatusSummary[] {
|
||||
return [...this.published.values()]
|
||||
liveSessionSummaries(): AgentSessionStatusSummary[] {
|
||||
const summaries: AgentSessionStatusSummary[] = []
|
||||
for (const [sessionId] of this.deps.sessions) {
|
||||
const summary = this.published.get(sessionId)
|
||||
if (summary) {
|
||||
summaries.push(summary)
|
||||
}
|
||||
}
|
||||
return summaries
|
||||
}
|
||||
|
||||
unsubscribe(id: string): void {
|
||||
|
||||
@@ -110,8 +110,9 @@ export class OrcaRuntimeWithGetWorktreePs extends OrcaRuntimeWithStructuredAgent
|
||||
hookSnapshots: this.getAgentStatusSnapshotFn?.() ?? [],
|
||||
// Structured sessions have no PTY, so they never reach the hook or retained snapshots. The
|
||||
// host already projects their status for the sidebar; without this the CLI reads a worktree
|
||||
// running a structured chat as idle while the GUI shows it working.
|
||||
structuredSummaries: getStructuredAgentSessionHost()?.publishedStatusSummaries() ?? [],
|
||||
// running a structured chat as idle while the GUI shows it working. Live sessions only: the
|
||||
// status feed retains a forgotten session's last projection for reloading renderers.
|
||||
structuredSummaries: getStructuredAgentSessionHost()?.liveSessionStatusSummaries() ?? [],
|
||||
orchestrationByPaneKey: this.agentOrchestrationProjection.buildByPaneKey(),
|
||||
getSummary: (summaryMap, pathIndex, missingIds, worktreeId) =>
|
||||
this.getSummaryForRuntimeWorktreeId(summaryMap, pathIndex, missingIds, worktreeId)
|
||||
|
||||
@@ -93,13 +93,13 @@ describe('terminal listing is deliberately left alone', () => {
|
||||
)
|
||||
// Guard the guard: an empty read would make every assertion below vacuously true.
|
||||
expect(listing).toContain('async listTerminals(')
|
||||
expect(listing).not.toContain('publishedStatusSummaries')
|
||||
expect(listing).not.toContain('liveSessionStatusSummaries')
|
||||
expect(listing).not.toContain('structuredSummaries')
|
||||
|
||||
const worktreePs = await readFile(
|
||||
new URL('./orca-runtime-get-worktree-ps.ts', import.meta.url),
|
||||
'utf8'
|
||||
)
|
||||
expect(worktreePs).toContain('publishedStatusSummaries')
|
||||
expect(worktreePs).toContain('liveSessionStatusSummaries')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -8,15 +8,12 @@ import {
|
||||
import { terminalStatusPayloadMatchesHook } from '../../shared/agent-terminal-status-equivalence'
|
||||
import type { RuntimeWorktreeAgentRow, RuntimeWorktreePsSummary } from '../../shared/runtime-types'
|
||||
import { parseLegacyNumericPaneKey, parsePaneKey } from '../../shared/stable-pane-id'
|
||||
import {
|
||||
structuredAgentSessionPaneKey,
|
||||
structuredAgentSessionTabId
|
||||
} from '../../shared/structured-agent-session-projection'
|
||||
import type { AgentSessionStatusSummary } from '../../shared/agent-session-wire'
|
||||
import { isWslHookRelayConnectionId } from '../../shared/wsl-hook-relay-contract'
|
||||
import { mergeWorktreeSummaryStatus } from './runtime-worktree-status-projection'
|
||||
import type { RuntimeWorktreeSummaryPathIndex } from './runtime-worktree-summary-paths'
|
||||
import type { RuntimeWorkingTerminalEvidence } from './runtime-worktree-ps-activity'
|
||||
import { structuredRuntimeWorktreeAgentSources } from './runtime-worktree-structured-agent-rows'
|
||||
|
||||
export type RuntimeAgentRowSnapshot = {
|
||||
paneKey: string
|
||||
@@ -41,13 +38,15 @@ type OrchestrationDisplay = {
|
||||
parentPaneKey?: string | null
|
||||
}
|
||||
|
||||
type RuntimeWorktreeAgentSource = {
|
||||
export type RuntimeWorktreeAgentSource = {
|
||||
paneKey: string
|
||||
ptyId?: string
|
||||
tabId?: string
|
||||
worktreeId?: string
|
||||
connectionId: string | null
|
||||
payload: ParsedAgentStatusPayload
|
||||
/** The hook/retained row this source came from. Absent on a structured session: nothing
|
||||
* downstream reads it, and there is no hook report behind one to carry. */
|
||||
payload?: ParsedAgentStatusPayload
|
||||
state: ParsedAgentStatusPayload['state']
|
||||
workingMode?: ParsedAgentStatusPayload['workingMode']
|
||||
agentType: string | null
|
||||
@@ -75,7 +74,8 @@ export function attachRuntimeWorktreeAgentRows(args: {
|
||||
>
|
||||
retainedSnapshots: Iterable<RuntimeAgentRowSnapshot>
|
||||
hookSnapshots: readonly AgentStatusIpcPayload[]
|
||||
/** Structured (non-PTY) sessions, projected from the host's own status feed. */
|
||||
/** The structured (non-PTY) sessions the host still holds, projected from its status feed.
|
||||
* Never the feed's whole retained cache: that keeps a summary after eviction forgets it. */
|
||||
structuredSummaries: readonly AgentSessionStatusSummary[]
|
||||
orchestrationByPaneKey: Record<string, OrchestrationDisplay> | null | undefined
|
||||
getSummary: (
|
||||
@@ -118,6 +118,7 @@ export function attachRuntimeWorktreeAgentRows(args: {
|
||||
if (
|
||||
entry.workingMode === 'monitoring' &&
|
||||
now - entry.receivedAt <= AGENT_STATUS_STALE_AFTER_MS &&
|
||||
existing.payload !== undefined &&
|
||||
terminalStatusPayloadMatchesHook(hookPayload, existing.payload)
|
||||
) {
|
||||
existing.workingMode = 'monitoring'
|
||||
@@ -146,47 +147,11 @@ export function attachRuntimeWorktreeAgentRows(args: {
|
||||
updatedAt: entry.receivedAt
|
||||
})
|
||||
}
|
||||
for (const summary of args.structuredSummaries) {
|
||||
// No turn has been persisted yet, so there is nothing to report - the same read the chat shows.
|
||||
if (!summary.status) {
|
||||
continue
|
||||
for (const source of structuredRuntimeWorktreeAgentSources(args.structuredSummaries)) {
|
||||
// A PTY-backed row already on this pane key wins: it has real process evidence behind it.
|
||||
if (!rowSources.has(source.paneKey)) {
|
||||
rowSources.set(source.paneKey, source)
|
||||
}
|
||||
const tabId = structuredAgentSessionTabId(summary.sessionId)
|
||||
// The DERIVED pane key the renderer already publishes, never the orchestration bearer handle
|
||||
// or the minted worker pane key: both of those are credentials.
|
||||
const paneKey = structuredAgentSessionPaneKey(tabId, summary.sessionId)
|
||||
if (rowSources.has(paneKey)) {
|
||||
continue
|
||||
}
|
||||
// Same projection the sidebar applies, so the CLI and the GUI cannot disagree about one session.
|
||||
const state =
|
||||
summary.status === 'working' ? 'working' : summary.status === 'attention' ? 'blocked' : 'done'
|
||||
rowSources.set(paneKey, {
|
||||
paneKey,
|
||||
tabId,
|
||||
worktreeId: summary.workspaceId,
|
||||
connectionId: null,
|
||||
payload: {
|
||||
state,
|
||||
prompt: summary.latestPrompt,
|
||||
agentType: summary.agent,
|
||||
...(summary.toolName ? { toolName: summary.toolName } : {}),
|
||||
...(summary.toolInput ? { toolInput: summary.toolInput } : {}),
|
||||
...(summary.lastAssistantMessage
|
||||
? { lastAssistantMessage: summary.lastAssistantMessage }
|
||||
: {})
|
||||
} as ParsedAgentStatusPayload,
|
||||
state,
|
||||
agentType: summary.agent,
|
||||
prompt: summary.latestPrompt,
|
||||
lastAssistantMessage: summary.lastAssistantMessage ?? null,
|
||||
toolName: summary.toolName ?? null,
|
||||
toolInput: summary.toolInput ?? null,
|
||||
interrupted: false,
|
||||
stateStartedAt: summary.updatedAt,
|
||||
updatedAt: summary.updatedAt,
|
||||
structured: true
|
||||
})
|
||||
}
|
||||
if (rowSources.size === 0) {
|
||||
return
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
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 { StructuredAgentSessionStatusFeed } from '../native-chat/agent-session-wire/structured-agent-session-status-feed'
|
||||
import { createTrackedJournalOpener } from '../native-chat/agent-session-journal/journal-store-test-open'
|
||||
import type { RuntimeWorktreePsSummary } from '../../shared/runtime-types'
|
||||
import { attachRuntimeWorktreeAgentRows } from './runtime-worktree-agent-rows'
|
||||
|
||||
/**
|
||||
* The whole chain `worktree ps` walks: journal -> status feed -> agent rows -> worktree status.
|
||||
*
|
||||
* The feed's `published` map never retracts, so reading it as a roster reports every session the
|
||||
* app has ever opened. A closed chat that was waiting on an approval is the sharp edge: deliberate
|
||||
* close does not settle a pending prompt, so the retained summary stays `attention`, which maps to
|
||||
* a `blocked` row and merges the worktree to `permission` for the 30-minute freshness window.
|
||||
*/
|
||||
const WORKTREE_ID = 'repo-1::/workspace/app'
|
||||
const SESSION = 'a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d'
|
||||
const IDENTITY = {
|
||||
provider: 'codex',
|
||||
threadId: 'thread-1',
|
||||
turnId: 'turn-1',
|
||||
ordinal: 0
|
||||
} as const
|
||||
|
||||
let root: string
|
||||
const journals = createTrackedJournalOpener()
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-structured-ps-liveness-'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await journals.closeAll()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
/** A session parked on an approval nobody answered — the state a deliberate close leaves behind. */
|
||||
async function awaitingApproval() {
|
||||
const journal = await journals.open({
|
||||
identity: {
|
||||
sessionId: SESSION,
|
||||
workspaceId: WORKTREE_ID,
|
||||
hostId: 'local',
|
||||
agent: 'codex',
|
||||
providerHandle: { kind: 'codex', threadId: 'thread-1' }
|
||||
},
|
||||
journalDir: join(root, SESSION)
|
||||
})
|
||||
await journal.appendItem(
|
||||
{ ...IDENTITY, ordinal: 1 },
|
||||
{ kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'rm the branch' }] },
|
||||
{ fence: 1 }
|
||||
)
|
||||
await journal.appendItem(
|
||||
{ ...IDENTITY, ordinal: 2 },
|
||||
{
|
||||
kind: 'approval',
|
||||
title: 'Run the command?',
|
||||
detail: null,
|
||||
options: [{ id: 'allow', label: 'Allow' }],
|
||||
resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
|
||||
},
|
||||
{ fence: 1 }
|
||||
)
|
||||
const sessions = new Map([
|
||||
[
|
||||
SESSION,
|
||||
{ journal, params: { location: { workspaceId: WORKTREE_ID }, provider: 'codex' as const } }
|
||||
]
|
||||
])
|
||||
const feed = new StructuredAgentSessionStatusFeed({
|
||||
sessions,
|
||||
getRecord: () => null,
|
||||
now: () => Date.now()
|
||||
})
|
||||
feed.publish(SESSION, journal)
|
||||
return { feed, sessions }
|
||||
}
|
||||
|
||||
function worktreeFor(feed: StructuredAgentSessionStatusFeed): RuntimeWorktreePsSummary {
|
||||
const row = {
|
||||
worktreeId: WORKTREE_ID,
|
||||
status: 'inactive',
|
||||
agents: []
|
||||
} as unknown as RuntimeWorktreePsSummary
|
||||
attachRuntimeWorktreeAgentRows({
|
||||
summaries: new Map([[WORKTREE_ID, row]]),
|
||||
pathIndex: { byPath: new Map(), byRealPath: new Map() } as never,
|
||||
missingWorktreeIds: new Set(),
|
||||
mirroredWorktreeIdByTabId: new Map(),
|
||||
connectedPtyEvidence: { tabIds: new Set(), paneKeys: new Set(), ptyIds: new Set() },
|
||||
workingTerminalEvidenceByWorktreeId: new Map(),
|
||||
retainedSnapshots: [],
|
||||
hookSnapshots: [],
|
||||
structuredSummaries: feed.liveSessionSummaries(),
|
||||
orchestrationByPaneKey: null,
|
||||
getSummary: (map, _paths, _missing, id) => map.get(id) ?? null
|
||||
})
|
||||
return row
|
||||
}
|
||||
|
||||
describe('worktree ps and a closed structured chat', () => {
|
||||
it('reports the blocked row while the session is still held', async () => {
|
||||
const { feed } = await awaitingApproval()
|
||||
const row = worktreeFor(feed)
|
||||
expect(row.agents).toHaveLength(1)
|
||||
expect(row.agents[0]?.state).toBe('blocked')
|
||||
expect(row.status).toBe('permission')
|
||||
})
|
||||
|
||||
it('stops reporting it once eviction forgets the session', async () => {
|
||||
const { feed, sessions } = await awaitingApproval()
|
||||
// `forget-session`, the last eviction step, does exactly this and nothing to the feed.
|
||||
sessions.delete(SESSION)
|
||||
|
||||
const row = worktreeFor(feed)
|
||||
expect(row.agents).toHaveLength(0)
|
||||
expect(row.status).toBe('inactive')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,48 @@
|
||||
import type { AgentSessionStatusSummary } from '../../shared/agent-session-wire'
|
||||
import {
|
||||
structuredAgentSessionPaneKey,
|
||||
structuredAgentSessionStatusState,
|
||||
structuredAgentSessionTabId
|
||||
} from '../../shared/structured-agent-session-projection'
|
||||
import type { RuntimeWorktreeAgentSource } from './runtime-worktree-agent-rows'
|
||||
|
||||
/**
|
||||
* Row sources for the structured (non-PTY) sessions a host still holds.
|
||||
*
|
||||
* A structured session reaches none of the hook or retained snapshots every other row comes from,
|
||||
* so `worktree ps` projects it from the host's status feed instead. The feed's retained
|
||||
* projections are not a roster — the caller passes only sessions the host still holds.
|
||||
*/
|
||||
export function structuredRuntimeWorktreeAgentSources(
|
||||
summaries: readonly AgentSessionStatusSummary[]
|
||||
): RuntimeWorktreeAgentSource[] {
|
||||
const sources: RuntimeWorktreeAgentSource[] = []
|
||||
for (const summary of summaries) {
|
||||
// No turn has been persisted yet, so there is nothing to report - the same read the chat shows.
|
||||
if (!summary.status) {
|
||||
continue
|
||||
}
|
||||
const tabId = structuredAgentSessionTabId(summary.sessionId)
|
||||
// The DERIVED pane key the renderer already publishes, never the orchestration bearer handle
|
||||
// or the minted worker pane key: both of those are credentials.
|
||||
sources.push({
|
||||
paneKey: structuredAgentSessionPaneKey(tabId, summary.sessionId),
|
||||
tabId,
|
||||
worktreeId: summary.workspaceId,
|
||||
connectionId: null,
|
||||
// The shared mapping the sidebar applies, so the CLI and the GUI cannot disagree about one
|
||||
// session. No hook payload: nothing reads one off a structured row.
|
||||
state: structuredAgentSessionStatusState(summary.status),
|
||||
agentType: summary.agent,
|
||||
prompt: summary.latestPrompt,
|
||||
lastAssistantMessage: summary.lastAssistantMessage ?? null,
|
||||
toolName: summary.toolName ?? null,
|
||||
toolInput: summary.toolInput ?? null,
|
||||
interrupted: false,
|
||||
stateStartedAt: summary.updatedAt,
|
||||
updatedAt: summary.updatedAt,
|
||||
structured: true
|
||||
})
|
||||
}
|
||||
return sources
|
||||
}
|
||||
@@ -2,7 +2,10 @@ import { useEffect, useMemo, useSyncExternalStore } from 'react'
|
||||
import { useShallow } from 'zustand/react/shallow'
|
||||
import { agentProviderSessionsEqual } from '../../../../shared/agent-session-resume'
|
||||
import type { AgentSessionStatusSummary } from '../../../../shared/agent-session-wire'
|
||||
import { structuredAgentSessionPaneKey } from '../../../../shared/structured-agent-session-projection'
|
||||
import {
|
||||
structuredAgentSessionPaneKey,
|
||||
structuredAgentSessionStatusState
|
||||
} from '../../../../shared/structured-agent-session-projection'
|
||||
import type { Tab } from '../../../../shared/tab-types'
|
||||
import { isAgentSessionHandleProvider } from '../../../../shared/agent-session-provider-handle'
|
||||
import { getRuntimeEnvironmentIdForWorktree } from '@/lib/worktree-runtime-owner'
|
||||
@@ -67,12 +70,8 @@ function projectStatus(tab: StructuredTab, summary: AgentSessionStatusSummary |
|
||||
return
|
||||
}
|
||||
const desired = {
|
||||
state:
|
||||
summary.status === 'working'
|
||||
? 'working'
|
||||
: summary.status === 'attention'
|
||||
? 'blocked'
|
||||
: 'done',
|
||||
// Shared with `worktree ps`, so the CLI and this row cannot disagree about one session.
|
||||
state: structuredAgentSessionStatusState(summary.status),
|
||||
prompt: summary.latestPrompt,
|
||||
agentType: tab.agentSessionAgent,
|
||||
// The host projects these from the journal so the row reads like a hook-reported one:
|
||||
|
||||
@@ -278,6 +278,14 @@ export function projectStructuredAgentSessionStatusSummary(
|
||||
}
|
||||
}
|
||||
|
||||
/** The agent-status state one projected session status stands for. Shared across the process
|
||||
* boundary so `worktree ps` and the sidebar cannot disagree about the same session. */
|
||||
export function structuredAgentSessionStatusState(
|
||||
status: StructuredAgentSessionProjectedStatus
|
||||
): 'working' | 'blocked' | 'done' {
|
||||
return status === 'working' ? 'working' : status === 'attention' ? 'blocked' : 'done'
|
||||
}
|
||||
|
||||
export function structuredAgentSessionPaneKey(tabId: string, sessionId: string): string {
|
||||
const bytes = sha256(new TextEncoder().encode(sessionId))
|
||||
const hex = Array.from(bytes.slice(0, 16), (byte) => byte.toString(16).padStart(2, '0')).join('')
|
||||
|
||||
Reference in New Issue
Block a user