diff --git a/src/main/agent-hooks/local-tmux-managed-pty.test.ts b/src/main/agent-hooks/local-tmux-managed-pty.test.ts new file mode 100644 index 000000000000..bb2798ec7e8f --- /dev/null +++ b/src/main/agent-hooks/local-tmux-managed-pty.test.ts @@ -0,0 +1,63 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { createLocalTmuxManagedPtyResolver } from './local-tmux-managed-pty' +import type { PtyProcessInfo } from '../providers/types' + +const row: PtyProcessInfo = { + id: 'pty', + incarnationId: 'generation', + rootProcessId: 123, + cwd: '/workspace', + title: 'shell', + worktreeId: 'folder:workspace' +} +afterEach(() => { + vi.useRealTimers() +}) + +describe('authenticated local tmux root resolution', () => { + it('shares inventory captures and accepts only the current daemon binding', async () => { + const listProcesses = vi.fn(async () => [row]) + const resolve = createLocalTmuxManagedPtyResolver({ getPtyId: () => 'pty', listProcesses }) + const [first, second] = await Promise.all([resolve('pane'), resolve('pane')]) + expect(first).toEqual({ + pid: 123, + incarnation: 'generation', + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder:workspace', + workspaceKind: 'folder' + } + }) + expect(second).toEqual(first) + expect(listProcesses).toHaveBeenCalledTimes(1) + }) + it('refuses a binding changed during capture and excludes WSL roots from a native probe', async () => { + let binding = 'pty' + const resolve = createLocalTmuxManagedPtyResolver({ + getPtyId: () => binding, + listProcesses: async () => { + binding = 'replacement' + return [row] + } + }) + expect(await resolve('pane')).toBeNull() + const wsl = createLocalTmuxManagedPtyResolver({ + getPtyId: () => 'pty', + listProcesses: async () => [{ ...row, wslDistro: 'Ubuntu-24.04' }] + }) + expect(await wsl('pane')).toBeNull() + }) + it('refreshes expired inventory instead of assigning a reused PID to the old incarnation', async () => { + vi.useFakeTimers() + vi.setSystemTime(1000) + const listProcesses = vi + .fn() + .mockResolvedValueOnce([row]) + .mockResolvedValueOnce([{ ...row, incarnationId: 'replacement' }]) + const resolve = createLocalTmuxManagedPtyResolver({ getPtyId: () => 'pty', listProcesses }) + expect((await resolve('pane'))?.incarnation).toBe('generation') + vi.setSystemTime(2001) + expect((await resolve('pane'))?.incarnation).toBe('replacement') + }) +}) diff --git a/src/main/agent-hooks/local-tmux-managed-pty.ts b/src/main/agent-hooks/local-tmux-managed-pty.ts new file mode 100644 index 000000000000..f4d58c8a98f7 --- /dev/null +++ b/src/main/agent-hooks/local-tmux-managed-pty.ts @@ -0,0 +1,56 @@ +import type { PtyProcessInfo } from '../providers/types' +import type { TmuxManagedPty } from '../../shared/tmux-agent-hook-owner' + +/** Reuse the daemon's authenticated inventory; never resolve an SSH pane from the client. */ +export function createLocalTmuxManagedPtyResolver(options: { + getPtyId: (paneKey: string) => string | undefined | null + listProcesses: () => Promise +}): (paneKey: string) => Promise { + let pending: Promise | undefined + let inventory: PtyProcessInfo[] = [] + let capturedAt = -Infinity + const read = async (): Promise => { + if (Date.now() - capturedAt < 1000) { + return inventory + } + if (!pending) { + pending = options + .listProcesses() + .then((rows) => { + inventory = rows + capturedAt = Date.now() + return rows + }) + .finally(() => { + pending = undefined + }) + } + return pending + } + return async (paneKey) => { + if (process.platform === 'win32') { + return null + } + const id = options.getPtyId(paneKey) + if (!id) { + return null + } + const row = (await read()).find((entry) => entry.id === id) + if (!row?.rootProcessId || !row.incarnationId || !row.worktreeId || row.wslDistro) { + return null + } + if (options.getPtyId(paneKey) !== id) { + return null + } + return { + pid: row.rootProcessId, + incarnation: row.incarnationId, + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: row.worktreeId, + workspaceKind: row.worktreeId.startsWith('folder:') ? 'folder' : 'git-worktree' + } + } + } +} diff --git a/src/main/agent-hooks/server-tmux-canonical.test.ts b/src/main/agent-hooks/server-tmux-canonical.test.ts new file mode 100644 index 000000000000..01d0baad4b95 --- /dev/null +++ b/src/main/agent-hooks/server-tmux-canonical.test.ts @@ -0,0 +1,70 @@ +import { createHash } from 'node:crypto' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AgentHookServer, _internals } from './server' +import { postHookEvent } from './server.test-fixtures' +import { + TMUX_TEST_PANE, + TMUX_TEST_ROOT, + TMUX_TEST_ROWS, + tmuxTestBody +} from '../../shared/tmux-status.test-fixture' + +vi.mock('../telemetry/client', () => ({ track: vi.fn() })) +vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: () => ({}) })) +const probe = vi.hoisted(() => vi.fn()) +vi.mock('../../shared/tmux-host-attachment-probe', () => ({ probeTmuxHostAttachments: probe })) +class CanonicalTestServer extends AgentHookServer { + hasLegacyRow(): boolean { + return this.state.lastStatusByPaneKey.has(TMUX_TEST_PANE) + } +} +let server: CanonicalTestServer +beforeEach(async () => { + _internals.resetCachesForTests() + probe.mockResolvedValue({ clients: [{ pid: 101, pane: '%0' }], rows: TMUX_TEST_ROWS }) + server = new CanonicalTestServer() + server.setTmuxManagedPtyResolver(async () => TMUX_TEST_ROOT) + await server.start({ env: 'production' }) +}) +afterEach(() => server.stop()) +describe('main canonical tmux projection', () => { + it('publishes one canonical outer row, blocks legacy overwrites, and cleans the exact outer and inner subjects', async () => { + expect((await postHookEvent(server, tmuxTestBody(), '/hook/opencode')).status).toBe(204) + expect(server.hasLegacyRow()).toBe(false) + expect( + server.attestCompatibilityAuthority({ + paneKey: TMUX_TEST_PANE, + launchTokenHash: createHash('sha256').update('generation').digest('hex'), + connectionId: null, + terminalProvenance: 'current_runtime' + }) + ).toEqual({ paneKey: TMUX_TEST_PANE, source: 'current_hook' }) + expect(server.getCanonicalStatusSnapshot().parents).toHaveLength(2) + expect(server.getStatusSnapshot()).toMatchObject([{ paneKey: TMUX_TEST_PANE, state: 'done' }]) + server.ingestTerminalStatus({ + paneKey: TMUX_TEST_PANE, + worktreeId: 'workspace', + payload: { state: 'working', prompt: 'untrusted repaint', agentType: 'opencode' } + }) + expect(server.hasLegacyRow()).toBe(false) + expect(server.getStatusSnapshot()[0]?.state).toBe('done') + server.clearPaneState(TMUX_TEST_PANE) + expect(server.getCanonicalStatusSnapshot().parents).toHaveLength(0) + expect(server.getStatusSnapshot()).toHaveLength(0) + }) + it('removes canonical tmux status on certified retirement', async () => { + await postHookEvent(server, tmuxTestBody(), '/hook/opencode') + server.retirePaneAuthority(TMUX_TEST_PANE) + expect(server.getStatusSnapshot()).toHaveLength(0) + expect(server.getCanonicalStatusSnapshot().parents).toHaveLength(0) + }) + it('dismisses a canonical projection without creating a legacy status copy', async () => { + await postHookEvent(server, tmuxTestBody(), '/hook/opencode') + server.dropStatusEntry(TMUX_TEST_PANE, { preserveResumeIdentity: false }) + expect(server.hasLegacyRow()).toBe(false) + expect(server.getStatusSnapshot()).toHaveLength(0) + expect(server.getCanonicalStatusSnapshot().parents).toHaveLength(2) + server.dropStatusEntriesByTabPrefix('tab-tmux') + expect(server.getCanonicalStatusSnapshot().parents).toHaveLength(0) + }) +}) diff --git a/src/main/agent-hooks/server-tmux-wire.test.ts b/src/main/agent-hooks/server-tmux-wire.test.ts new file mode 100644 index 000000000000..5037d173adc0 --- /dev/null +++ b/src/main/agent-hooks/server-tmux-wire.test.ts @@ -0,0 +1,79 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AgentHookServer, _internals } from './server' +import { PANE } from './server.test-fixtures' +import { buildRelayHookEnvelope } from '../../relay/agent-hook-envelope-build' + +vi.mock('../telemetry/client', () => ({ track: vi.fn() })) +vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: () => ({}) })) +const time = 1_800_000_000_000 +const envelope = { + source: 'opencode', + paneKey: PANE, + worktreeId: 'workspace', + connectionId: null, + launchToken: 'generation', + payload: { agentType: 'opencode', state: 'waiting', prompt: 'approval' } +} as const +const unavailable = { ...envelope, statusUnavailable: true, payload: null } as const + +describe('optional tmux projection wire metadata', () => { + let server: AgentHookServer + beforeEach(() => { + _internals.resetCachesForTests() + vi.useFakeTimers() + vi.setSystemTime(time) + server = new AgentHookServer() + }) + afterEach(() => { + server.stop() + vi.useRealTimers() + }) + it('preserves host evidence age while keeping local delivery order current', () => { + server.ingestRemote({ ...envelope, evidenceAgeMs: 60_000 }, 'connection') + const row = server.getStatusSnapshot().find((entry) => entry.paneKey === PANE) + expect(row?.evidenceObservedAt).toBe(time - 60_000) + expect(row?.receivedAt).toBe(time) + vi.setSystemTime(time + 10_000) + server.ingestRemote({ ...envelope, evidenceAgeMs: 70_000, isReplay: true }, 'connection') + expect( + server.getStatusSnapshot().find((entry) => entry.paneKey === PANE)?.evidenceObservedAt + ).toBe(time - 60_000) + }) + it('clears a selected projection only for the matching host, workspace and generation', () => { + server.ingestRemote(envelope, 'connection') + server.ingestRemote(unavailable, 'old-connection') + server.ingestRemote({ ...unavailable, launchToken: 'old-generation' }, 'connection') + server.ingestRemote({ ...unavailable, worktreeId: 'foreign' }, 'connection') + expect(server.getStatusSnapshot()).toHaveLength(1) + server.ingestRemote(unavailable, 'connection') + expect(server.getStatusSnapshot()).toHaveLength(0) + }) + it('accepts old relay envelopes and ignores unknown null payloads without fabricating Done', () => { + server.ingestRemote(envelope, 'connection') + expect(server.getStatusSnapshot()[0]?.evidenceObservedAt).toBe(time) + server.ingestRemote({ ...unavailable, statusUnavailable: undefined }, 'connection') + expect(server.getStatusSnapshot()[0]?.state).toBe('waiting') + }) + it('rejects invalid ages and never forwards absolute host clocks', () => { + for (const age of [-1, Number.NaN, Infinity, Number.MAX_SAFE_INTEGER + 1, 1.5]) { + server.ingestRemote({ ...envelope, evidenceAgeMs: age }, 'connection') + } + expect(server.getStatusSnapshot()).toHaveLength(0) + const built = buildRelayHookEnvelope( + { ...envelope, hostEvidenceObservedAt: time - 1000 }, + 'opencode' + ) + expect(built.evidenceAgeMs).toBe(1000) + expect(built).not.toHaveProperty('hostEvidenceObservedAt') + vi.setSystemTime(time + 1000) + expect( + buildRelayHookEnvelope( + { ...envelope, hostEvidenceObservedAt: time - 1000 }, + 'opencode', + undefined, + undefined, + { isReplay: true } + ).evidenceAgeMs + ).toBe(2000) + }) +}) diff --git a/src/main/agent-hooks/server/server-authority-fences.ts b/src/main/agent-hooks/server/server-authority-fences.ts index 3bd289d9c551..4ffa68033b95 100644 --- a/src/main/agent-hooks/server/server-authority-fences.ts +++ b/src/main/agent-hooks/server/server-authority-fences.ts @@ -49,6 +49,7 @@ export abstract class AgentHookServerAuthorityFences extends AgentHookServerAuth }) const hadStatus = retiredRows.length > 0 for (const key of paneKeys) { + this.clearTmuxInnerSubjects(key) this.markPaneClosedForAgentStatus(key) this.restartedStatusLaunchTokenHashByPaneKey.delete(key) this.clearAssistantMessageRetry(key) diff --git a/src/main/agent-hooks/server/server-cleanup.ts b/src/main/agent-hooks/server/server-cleanup.ts index 71be07e61703..15a773a9b22a 100644 --- a/src/main/agent-hooks/server/server-cleanup.ts +++ b/src/main/agent-hooks/server/server-cleanup.ts @@ -68,13 +68,14 @@ export abstract class AgentHookServerCleanup extends AgentHookServerAuthorityFen const existing = this.state.lastStatusByPaneKey.get(resolvedPaneKey) as | EnrichedAgentHookEventPayload | undefined + const canonical = existing ?? this.getTmuxSelectedStatus(resolvedPaneKey) // Why: stateStartedAt pins the turn; the renderer's updatedAt is stamped at or after this // receivedAt (runtime-sync and recovery paths use Date.now()/capturedAt), so a strictly // newer cached event is the only replacement worth protecting. if ( - !existing || - existing.stateStartedAt !== identity.stateStartedAt || - existing.receivedAt > identity.receivedAt + !canonical || + canonical.stateStartedAt !== identity.stateStartedAt || + canonical.receivedAt > identity.receivedAt ) { continue } @@ -167,7 +168,7 @@ export abstract class AgentHookServerCleanup extends AgentHookServerAuthorityFen * itself lives beside `clearPaneCacheState`, so adding a latch cannot leave this behind in a * different file. */ protected hasLiveClaimsForPaneKey(paneKey: string): boolean { - return paneHasStateClaims(this.state, paneKey) + return Boolean(this.getTmuxSelectedStatus(paneKey)) || paneHasStateClaims(this.state, paneKey) } /** Clear statuses proven to belong to one lost SSH transport. */ @@ -232,7 +233,7 @@ export abstract class AgentHookServerCleanup extends AgentHookServerAuthorityFen | EnrichedAgentHookEventPayload | undefined if (!existing) { - return null + return this.deleteTmuxSelectedStatus(resolvedPaneKey) ?? null } deleteLegacyAgentStatus(this.state, resolvedPaneKey) this.activeHookTurnCompletedAtByPaneKey.delete(resolvedPaneKey) diff --git a/src/main/agent-hooks/server/server-ingest-normalization.ts b/src/main/agent-hooks/server/server-ingest-normalization.ts index 5e71e6ef033f..accd71884c70 100644 --- a/src/main/agent-hooks/server/server-ingest-normalization.ts +++ b/src/main/agent-hooks/server/server-ingest-normalization.ts @@ -3,9 +3,9 @@ import { normalizeHookPayload } from '../../../shared/agent-hook-listener' import { isAgentHookSource, type AgentHookSource } from '../../../shared/agent-hook-relay' import { isOpenCodeSharedServerPost } from '../../../shared/agent-hook-listener/opencode-session-registry' import type { NormalizedLocalHook } from './server-types' -import { AgentHookServerOpenCodeBinder } from './server-opencode-binder' +import { AgentHookServerTmuxStatus } from './server-tmux-status' -export abstract class AgentHookServerIngestNormalization extends AgentHookServerOpenCodeBinder { +export abstract class AgentHookServerIngestNormalization extends AgentHookServerTmuxStatus { protected setClaudeBackgroundEvidence( paneKey: string, hasRunningTask: boolean, diff --git a/src/main/agent-hooks/server/server-ingest-remote.ts b/src/main/agent-hooks/server/server-ingest-remote.ts index 28eac6e56dc1..5fc4456f30f3 100644 --- a/src/main/agent-hooks/server/server-ingest-remote.ts +++ b/src/main/agent-hooks/server/server-ingest-remote.ts @@ -54,6 +54,8 @@ export abstract class AgentHookServerIngestRemote extends AgentHookServerIngestS claudeRunningNonAgentTask?: unknown /** The producing peer's advertised run-capability set — a property of the peer/connection that built this envelope, not an orthogonal call parameter. Absent (older relay/HTTP paths) defaults to the unadvertised-legacy-peer set. */ advertisedAgentStatusCapabilities?: readonly string[] + statusUnavailable?: unknown + evidenceAgeMs?: unknown payload: unknown }, connectionId: string | null @@ -138,6 +140,25 @@ export abstract class AgentHookServerIngestRemote extends AgentHookServerIngestS toolAgentType, providerSession } = normalizeRemoteEnvelopeFields(envelope) + if (envelope.statusUnavailable === true && envelope.payload === null) { + const previous = this.state.lastStatusByPaneKey.get(paneKey) + if ( + (source === 'opencode' || source === 'opencode2') && + previous?.connectionId === trimmedConnectionId && + trimmedConnectionId !== null && + previous.worktreeId === worktreeId && + previous.source === source && + previous.launchToken === envelope.launchToken && + this.getAgentStatusDisposition(paneKey) !== 'suppress' + ) { + this.clearPaneState(paneKey, { statusUnavailable: true }) + } + return + } + const age = envelope.evidenceAgeMs + if (age !== undefined && (typeof age !== 'number' || !Number.isSafeInteger(age) || age < 0)) { + return + } // Why: relay crosses a trust boundary — re-run the canonical normalizer to enforce caps/invariants (returns null on malformed). const validatedPayload = normalizeAgentStatusPayload(envelope.payload) if (!validatedPayload) { @@ -200,10 +221,7 @@ export abstract class AgentHookServerIngestRemote extends AgentHookServerIngestS if (hookEventName === 'PreCompact' || source !== 'claude') { return } - // Why: a relay predating this change strips `compactTrigger` from its cached PostCompact - // before replaying it, so the replay has no manual/auto discriminator. That relay's mapping is - // fixed and known — manual produced `done`, auto produced `working` — so the payload state - // stands in for the missing trigger. Trigger substitution only; ownership is still checked. + // Older relays omit the trigger; their state supplies it without bypassing ownership. const effectiveTrigger = resolveLegacyCompactTrigger(compactTrigger, normalizedPayload.state) // Why: an auto compact happens inside a turn that resumes and emits its own Stop. An older // relay maps it to `working`, and this ingest applies the relay's payload verbatim — so @@ -298,7 +316,9 @@ export abstract class AgentHookServerIngestRemote extends AgentHookServerIngestS this.state.claudeRunningNonAgentTaskPaneKeys.delete(paneKey) } } - : undefined + : undefined, + 'hook', + typeof age === 'number' ? Math.max(0, Date.now() - age) : undefined ) } } diff --git a/src/main/agent-hooks/server/server-lifecycle.ts b/src/main/agent-hooks/server/server-lifecycle.ts index e6e1bf80061d..80c81f091427 100644 --- a/src/main/agent-hooks/server/server-lifecycle.ts +++ b/src/main/agent-hooks/server/server-lifecycle.ts @@ -103,6 +103,11 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerRuntimeEnv const hookBody = mergeAgentHookRequestHeaders(body, req.headers) trackEmptyPaneKeyHook(hookBody) const aliasedBody = this.normalizeHookBodyPaneKeyAlias(hookBody) + if (await this.ingestTmuxHook(source, aliasedBody)) { + res.writeHead(204) + res.end() + return + } const normalized = this.normalizeLocalHookPayload(source, aliasedBody) const statusDisposition = normalized.event ? this.getAgentStatusDisposition(normalized.event.paneKey, { @@ -202,6 +207,7 @@ export abstract class AgentHookServerLifecycle extends AgentHookServerRuntimeEnv // Why: flush the pending debounced write before clearing the map, else a hook <250ms before quit is lost on relaunch. this.flushStatusPersistSync() this.stopOpenCodeBinderLoop() + this.stopTmuxStatus() this.rollbackTransportStart() this.env = 'production' this.onAgentStatus = null diff --git a/src/main/agent-hooks/server/server-listeners.ts b/src/main/agent-hooks/server/server-listeners.ts index 961d3a3e9c87..1de09c3ebdf7 100644 --- a/src/main/agent-hooks/server/server-listeners.ts +++ b/src/main/agent-hooks/server/server-listeners.ts @@ -1,3 +1,4 @@ +import { isTmuxInnerSubject } from '../../../shared/tmux-agent-hook-owner' import type { AgentStatusClearIpcPayload, AgentStatusIpcPayload @@ -54,7 +55,7 @@ export abstract class AgentHookServerListeners extends AgentHookServerState { }) } for (const parent of this.canonicalStatusStore.getParents()) { - if (!parent.status) { + if (!parent.status || isTmuxInnerSubject(parent.subject)) { continue } rows.push({ diff --git a/src/main/agent-hooks/server/server-state.ts b/src/main/agent-hooks/server/server-state.ts index 49bb8ade3ac9..f429c0e63501 100644 --- a/src/main/agent-hooks/server/server-state.ts +++ b/src/main/agent-hooks/server/server-state.ts @@ -28,7 +28,7 @@ import type { LegacyPaneKeyAliasEntry } from '../../../shared/persisted-state-ty import type { SpoolRecord } from '../../../shared/agent-hook-spool' import { createAgentStatusStore, type AgentStatusStore } from '../../../shared/agent-status-store' import { AGENT_STATUS_2A_CURRENT_PRODUCER_MODE } from '../../../shared/agent-status-legacy-adapter' -import type { AgentStatusStructuredSessionSubject } from '../../../shared/agent-status-subject' +import type { AgentStatusSubject } from '../../../shared/agent-status-subject' import type { AgentHookAuthorityEvidence, AgentHookProviderSessionIdentity, @@ -73,10 +73,7 @@ export abstract class AgentHookServerState { return this.canonicalStatusStoreInstance } protected readonly canonicalListingOrder = new Map() - protected readonly canonicalSubjectsByPane = new Map< - string, - Map - >() + protected readonly canonicalSubjectsByPane = new Map>() private statusListingOrder = 0 protected nextStatusListingOrder = (): number => ++this.statusListingOrder @@ -163,6 +160,14 @@ export abstract class AgentHookServerState { ) protected abstract withdrawReplayObservation(paneKey: string): void + protected abstract getTmuxSelectedStatus( + paneKey: string + ): EnrichedAgentHookEventPayload | undefined + protected abstract deleteTmuxSelectedStatus( + paneKey: string + ): EnrichedAgentHookEventPayload | undefined + protected abstract clearTmuxInnerSubjects(paneKey: string): void + protected abstract clearTmuxTabSubjects(tabId: string): void protected abstract ingestSpoolRecord(record: SpoolRecord): void protected abstract emitPaneStatusCleared(clear: AgentStatusClearIpcPayload): void protected abstract buildStatusChangeNotification(): { @@ -272,7 +277,11 @@ export abstract class AgentHookServerState { protected abstract clearPaneState( paneKey: string, - options?: { emitStatusRowMutation?: boolean } + options?: { + emitStatusRowMutation?: boolean + preserveTmuxInnerSubjects?: boolean + statusUnavailable?: true + } ): void protected abstract deleteStatusEntry( paneKey: string, diff --git a/src/main/agent-hooks/server/server-structured-status-row.ts b/src/main/agent-hooks/server/server-structured-status-row.ts index 93fda643409b..d54b0b040bca 100644 --- a/src/main/agent-hooks/server/server-structured-status-row.ts +++ b/src/main/agent-hooks/server/server-structured-status-row.ts @@ -17,6 +17,9 @@ export function structuredStatusLegacyEvent( stateStartedAt: row.stateStartedAt, evidenceObservedAt: row.evidenceObservedAt, structuredHost: row.structuredHost, + launchToken: row.launchToken, + terminalHandle: row.terminalHandle, + promptInteractionKey: row.promptInteractionKey, ...(row.providerSession ? { providerSession: row.providerSession } : {}), ...(row.observation ? { observation: row.observation } : {}), payload: pickParsedAgentStatusPayload(row) diff --git a/src/main/agent-hooks/server/server-tab-cleanup.ts b/src/main/agent-hooks/server/server-tab-cleanup.ts index 599665cc98b5..bc1d481498bf 100644 --- a/src/main/agent-hooks/server/server-tab-cleanup.ts +++ b/src/main/agent-hooks/server/server-tab-cleanup.ts @@ -7,6 +7,7 @@ export abstract class AgentHookServerTabCleanup extends AgentHookServerCleanup { /** Drop every status/cache claim attributable to a closed tab prefix. */ dropStatusEntriesByTabPrefix(tabId: string): void { this.markTabClosedForAgentStatus(tabId) + this.clearTmuxTabSubjects(tabId) const paneKeysToClear = new Set() const statusPaneKeysToClear = new Set() const statusRowsToClear: EnrichedAgentHookEventPayload[] = [] @@ -100,8 +101,18 @@ export abstract class AgentHookServerTabCleanup extends AgentHookServerCleanup { } } - clearPaneState(paneKey: string, options?: { emitStatusRowMutation?: boolean }): void { + clearPaneState( + paneKey: string, + options?: { + emitStatusRowMutation?: boolean + preserveTmuxInnerSubjects?: boolean + statusUnavailable?: true + } + ): void { const resolvedPaneKey = this.resolvePaneKeyAlias(paneKey) + if (!options?.preserveTmuxInnerSubjects) { + this.clearTmuxInnerSubjects(resolvedPaneKey) + } const paneKeys = new Set([paneKey, resolvedPaneKey]) // Why: only persist when a status entry was actually evicted; dropping prompt/tool caches doesn't change the file. const previousStatus = this.state.lastStatusByPaneKey.get(resolvedPaneKey) as @@ -143,7 +154,10 @@ export abstract class AgentHookServerTabCleanup extends AgentHookServerCleanup { this.runtimeObservedStatusPaneKeys.delete(resolvedPaneKey) this.scheduleStatusPersist() this.notifyStatusChangeListeners() - this.emitPaneStatusCleared({ paneKey: resolvedPaneKey }) + this.emitPaneStatusCleared({ + paneKey: resolvedPaneKey, + statusUnavailable: options?.statusUnavailable + }) } } } diff --git a/src/main/agent-hooks/server/server-tmux-status.ts b/src/main/agent-hooks/server/server-tmux-status.ts new file mode 100644 index 000000000000..59df76c7ed64 --- /dev/null +++ b/src/main/agent-hooks/server/server-tmux-status.ts @@ -0,0 +1,163 @@ +import { TmuxAgentHookOwner, type TmuxManagedPty } from '../../../shared/tmux-agent-hook-owner' +import { commitTmuxSelectedStatus } from '../../../shared/tmux-selected-status' +import { + serializeAgentStatusSubject, + type AgentStatusSubject +} from '../../../shared/agent-status-subject' +import { structuredStatusLegacyEvent } from './server-structured-status-row' +import type { AgentHookSource } from '../../../shared/agent-hook-relay' +import { AgentHookServerOpenCodeBinder } from './server-opencode-binder' + +export abstract class AgentHookServerTmuxStatus extends AgentHookServerOpenCodeBinder { + private tmuxOwner: TmuxAgentHookOwner | undefined + private tmuxRootResolver: (paneKey: string) => Promise = async () => null + + setTmuxManagedPtyResolver(resolver: (paneKey: string) => Promise): void { + this.tmuxRootResolver = resolver + } + + private get owner(): TmuxAgentHookOwner { + this.tmuxOwner ??= new TmuxAgentHookOwner({ + store: () => this.canonicalStatusStore, + getRoot: (paneKey) => this.tmuxRootResolver(paneKey), + isRetired: (paneKey) => this.getAgentStatusDisposition(paneKey) === 'suppress', + publish: (event, observedAt, subject, stateStartedAt) => { + if (this.getAgentStatusDisposition(event.paneKey, event) === 'suppress') { + return + } + this.clearPaneState(event.paneKey, { preserveTmuxInnerSubjects: true }) + const previous = this.canonicalStatusStore.getParent(subject)?.status + const status = commitTmuxSelectedStatus( + this.canonicalStatusStore, + subject, + event, + observedAt, + stateStartedAt + ) + if (!status) { + return + } + const key = serializeAgentStatusSubject(subject) + const subjects = + this.canonicalSubjectsByPane.get(event.paneKey) ?? new Map() + subjects.set(key, subject) + this.canonicalSubjectsByPane.set(event.paneKey, subjects) + if (!this.canonicalListingOrder.has(key)) { + this.canonicalListingOrder.set(key, this.nextStatusListingOrder()) + } + const after = { + ...structuredStatusLegacyEvent(status), + source: event.source, + hookEventName: event.hookEventName, + hasExplicitPrompt: event.hasExplicitPrompt, + launchToken: event.launchToken, + promptInteractionKey: event.promptInteractionKey + } + this.commitStatusRowMutation(previous && structuredStatusLegacyEvent(previous), after) + this.recordCurrentAuthorityObservation(after) + this.notifyStatusChangeListeners() + this.emitEnrichedStatus(after) + }, + unavailable: (paneKey, subject) => { + if (this.getAgentStatusDisposition(paneKey) === 'suppress') { + return + } + if (subject && !this.canonicalSubjectsByPane.has(paneKey)) { + this.canonicalSubjectsByPane.set( + paneKey, + new Map([[serializeAgentStatusSubject(subject), subject]]) + ) + this.canonicalStatusStore.applyMutation({ parent: { subject } }) + } + this.clearTmuxSelectedStatus(paneKey, true) + this.clearPaneState(paneKey, { preserveTmuxInnerSubjects: true, statusUnavailable: true }) + } + }) + return this.tmuxOwner + } + + protected ingestTmuxHook(source: AgentHookSource, body: unknown): Promise { + return this.owner.ingest(source, body, this.env) + } + + protected clearTmuxInnerSubjects(paneKey: string): void { + this.tmuxOwner?.clearPane(paneKey) + this.clearTmuxSelectedStatus(paneKey) + } + + protected clearTmuxTabSubjects(tabId: string): void { + this.tmuxOwner?.clearTab(tabId) + for (const paneKey of this.canonicalSubjectsByPane.keys()) { + if (paneKey.startsWith(`${tabId}:`)) { + this.clearTmuxSelectedStatus(paneKey) + } + } + } + + protected getTmuxSelectedStatus(paneKey: string) { + for (const subject of this.canonicalSubjectsByPane.get(paneKey)?.values() ?? []) { + if (subject.kind !== 'pty') { + continue + } + const status = this.canonicalStatusStore.getParent(subject)?.status + if (status) { + return structuredStatusLegacyEvent(status) + } + } + return undefined + } + + protected deleteTmuxSelectedStatus(paneKey: string) { + const previous = this.getTmuxSelectedStatus(paneKey) + if (!previous) { + return undefined + } + for (const subject of this.canonicalSubjectsByPane.get(paneKey)?.values() ?? []) { + if (subject.kind === 'pty') { + this.canonicalStatusStore.applyMutation({ parent: { subject } }) + } + } + return previous + } + + private clearTmuxSelectedStatus(paneKey: string, unavailable = false): void { + const subjects = this.canonicalSubjectsByPane.get(paneKey) + let changed = false + for (const [key, subject] of subjects ?? []) { + if (subject.kind !== 'pty') { + continue + } + const previous = this.canonicalStatusStore.getParent(subject)?.status + if ( + !this.canonicalStatusStore.applyMutation( + unavailable ? { parent: { subject } } : { removeParent: subject } + ) + ) { + continue + } + if (!unavailable) { + subjects?.delete(key) + this.canonicalListingOrder.delete(key) + } + if (previous) { + this.commitStatusRowMutation(structuredStatusLegacyEvent(previous), undefined) + } + changed = true + } + if (subjects?.size === 0) { + this.canonicalSubjectsByPane.delete(paneKey) + } + if (changed) { + this.notifyStatusChangeListeners() + this.emitPaneStatusCleared({ + paneKey, + ...(unavailable ? { statusUnavailable: true as const } : {}) + }) + } + } + + protected stopTmuxStatus(): void { + this.tmuxOwner?.stop() + this.tmuxOwner = undefined + } +} diff --git a/src/main/opencode/status-plugin-post-source.ts b/src/main/opencode/status-plugin-post-source.ts index 7c219857bca4..ee8091a054f2 100644 --- a/src/main/opencode/status-plugin-post-source.ts +++ b/src/main/opencode/status-plugin-post-source.ts @@ -8,9 +8,12 @@ export function getStatusPluginPostSource(hookPathname: string): string[] { ' const coords = resolveHookCoords();', ' const paneKey = process.env.ORCA_PANE_KEY;', ' if (!coords.port || !coords.token || !paneKey) return false;', + ' const tmuxMatch = /^(.*),[0-9]+,[0-9]+$/.exec(process.env.TMUX || "");', + ' const tmuxPane = process.env.TMUX_PANE;', ` const url = \`http://127.0.0.1:\${coords.port}${hookPathname}\`;`, ' const body = JSON.stringify({', ' paneKey,', + ' ...(tmuxMatch && /^%[0-9]+$/.test(tmuxPane || "") ? { tmux: { socket: tmuxMatch[1], pane: tmuxPane } } : {}),', ' launchToken: process.env.ORCA_AGENT_LAUNCH_TOKEN || "",', ' tabId: process.env.ORCA_TAB_ID || "",', ' worktreeId: process.env.ORCA_WORKTREE_ID || "",', diff --git a/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts b/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts index 0ac77cfd32c3..19dac11ffd1b 100644 --- a/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts +++ b/src/main/ssh/ssh-relay-session-agent-hooks.integration.test.ts @@ -306,6 +306,39 @@ describe('SshRelaySession agent hooks over a fake relay transport', () => { }) }) + it('preserves tmux evidence age and unavailable across the real notification adapter', async () => { + relay = createFakeRelay() + vi.mocked(deployAndLaunchRelay).mockResolvedValue({ + transport: relay.transport, + serverBuildId: 'test-relay-build', + platform: 'linux-x64' + }) + const events: CapturedStatus[] = [] + captureAgentStatuses(events) + session = createSession('conn-tmux') + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Mocked deployment never reads the connection. + await session.establish({} as SshConnection) + const envelope = makeEnvelope({ + source: 'opencode', + evidenceAgeMs: 60_000, + payload: { state: 'done', prompt: 'older inner turn', agentType: 'opencode' } + }) + const before = Date.now() + relay.notifyAgentHook(envelope) + await waitForStatusCount(events, 1) + const row = agentHookServer + .getStatusSnapshot() + .find((entry) => entry.paneKey === envelope.paneKey) + expect(row?.evidenceObservedAt).toBeGreaterThanOrEqual(before - 60_000) + expect(row?.evidenceObservedAt).toBeLessThanOrEqual(Date.now() - 60_000) + relay.notifyAgentHook({ ...envelope, statusUnavailable: true, payload: null }) + await vi.waitFor(() => + expect( + agentHookServer.getStatusSnapshot().find((entry) => entry.paneKey === envelope.paneKey) + ).toBeUndefined() + ) + }) + it('preserves Claude monitoring mode across the SSH relay boundary', async () => { relay = createFakeRelay() vi.mocked(deployAndLaunchRelay).mockResolvedValue({ diff --git a/src/main/ssh/ssh-relay-session.ts b/src/main/ssh/ssh-relay-session.ts index c29ec93c11e1..673fc25fed58 100644 --- a/src/main/ssh/ssh-relay-session.ts +++ b/src/main/ssh/ssh-relay-session.ts @@ -1718,6 +1718,8 @@ export class SshRelaySession { : undefined, // Why: the SSH relay protocol advertises no run-serving capability. advertisedAgentStatusCapabilities: AGENT_STATUS_LEGACY_UNADVERTISED_PEER_CAPABILITIES, + evidenceAgeMs: envelope.evidenceAgeMs, + statusUnavailable: envelope.statusUnavailable, payload: envelope.payload }, this.targetId diff --git a/src/main/startup/main-process-pty-startup.ts b/src/main/startup/main-process-pty-startup.ts index 648db87a3b61..03cef4732991 100644 --- a/src/main/startup/main-process-pty-startup.ts +++ b/src/main/startup/main-process-pty-startup.ts @@ -22,6 +22,7 @@ import { isAgentStatusHooksEnabledForAgent } from '../agent-hooks/managed-agent-hook-controls' import { agentHookServer } from '../agent-hooks/server' +import { createLocalTmuxManagedPtyResolver } from '../agent-hooks/local-tmux-managed-pty' import { indexPersistedPaneKeyPtyIds, isLocalExecutionHost, @@ -123,6 +124,12 @@ export async function reapRestoredSubagentsWithoutLiveAgent(): Promise { } export function startTerminalRuntimeStartupServices(): WindowsDesktopStartupServices { + agentHookServer.setTmuxManagedPtyResolver( + createLocalTmuxManagedPtyResolver({ + getPtyId: getPtyIdForPaneKey, + listProcesses: async () => (await getDaemonProvider()?.listProcesses()) ?? [] + }) + ) logStartupMilestone('first-window-startup-services-start') const startupServices = startFirstWindowStartupServices({ // Why: both desktop and headless serve must adopt the same persistent provider before creating terminals or a renderer. diff --git a/src/relay/agent-hook-canonical-status.test.ts b/src/relay/agent-hook-canonical-status.test.ts new file mode 100644 index 000000000000..24c2b8e01045 --- /dev/null +++ b/src/relay/agent-hook-canonical-status.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { RelayAgentHookServer } from './agent-hook-server' +import { + TMUX_TEST_PANE, + TMUX_TEST_ROOT, + TMUX_TEST_ROWS, + tmuxTestBody +} from '../shared/tmux-status.test-fixture' +const probe = vi.hoisted(() => vi.fn()) +vi.mock('../shared/tmux-host-attachment-probe', () => ({ probeTmuxHostAttachments: probe })) +class CanonicalTestServer extends RelayAgentHookServer { + snapshot() { + return this.canonicalStatusStore.getSnapshot() + } +} +let server: CanonicalTestServer +let directory: string +const forward = vi.fn() +const unavailable = vi.fn() +beforeEach(async () => { + directory = mkdtempSync(join(tmpdir(), 'orca-tmux-canonical-')) + forward.mockClear() + unavailable.mockClear() + probe.mockResolvedValue({ clients: [{ pid: 101, pane: '%0' }], rows: TMUX_TEST_ROWS }) + server = new CanonicalTestServer({ + endpointDir: directory, + forward, + forwardUnavailable: unavailable, + getTmuxManagedPty: async () => TMUX_TEST_ROOT + }) + await server.start({ publishEndpoint: false }) +}) +afterEach(() => { + server.stop() + rmSync(directory, { recursive: true, force: true }) +}) +async function post(body: unknown = tmuxTestBody(), source = 'opencode') { + const coordinates = server.getCoordinates() + return fetch(`http://127.0.0.1:${coordinates.port}/hook/${source}`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Orca-Agent-Hook-Token': coordinates.token }, + body: JSON.stringify(body) + }) +} +describe('relay canonical tmux ownership and replay', () => { + it('keeps inner and selected observations in its owner store and replays only the selected outer row', async () => { + expect((await post()).status).toBe(204) + expect(server.snapshot().parents).toHaveLength(2) + forward.mockClear() + expect(server.replayCachedPayloadsForPanes()).toBe(1) + expect(forward).toHaveBeenCalledTimes(1) + expect(forward.mock.lastCall?.[0]).toMatchObject({ + paneKey: TMUX_TEST_PANE, + payload: { state: 'done' } + }) + server.clearPaneState(TMUX_TEST_PANE) + expect(server.snapshot().parents).toHaveLength(0) + expect(server.replayCachedPayloadsForPanes()).toBe(0) + }) + it('replaces a legacy OpenCode 2 row with unavailable and preserves its launch identity', async () => { + const body = tmuxTestBody() + const { tmux: _tmux, ...legacy } = body + expect((await post(legacy, 'opencode2')).status).toBe(204) + probe.mockResolvedValue({ clients: [], rows: TMUX_TEST_ROWS }) + await post(body, 'opencode2') + await vi.waitFor(() => expect(unavailable).toHaveBeenCalled(), { timeout: 2500 }) + forward.mockClear() + unavailable.mockClear() + expect(server.replayCachedPayloadsForPanes()).toBe(1) + expect(forward).not.toHaveBeenCalled() + expect(unavailable.mock.lastCall?.[0]).toMatchObject({ + source: 'opencode2', + tabId: 'tab-tmux', + launchToken: 'generation', + statusUnavailable: true + }) + }) + it('derives first unavailable identity from the same outer inner observation', async () => { + probe.mockResolvedValue({ clients: [{ pid: 101, pane: '%9' }], rows: TMUX_TEST_ROWS }) + await post(tmuxTestBody(), 'opencode2') + await vi.waitFor(() => expect(unavailable).toHaveBeenCalled(), { timeout: 2500 }) + expect(unavailable.mock.lastCall?.[0]).toMatchObject({ + source: 'opencode2', + tabId: 'tab-tmux', + launchToken: 'generation' + }) + }) + it('retains and replays an unavailable projection when it was emitted without a connected reader', async () => { + await post() + probe.mockResolvedValue({ clients: [], rows: TMUX_TEST_ROWS }) + await vi.waitFor(() => expect(unavailable).toHaveBeenCalledTimes(1), { timeout: 2500 }) + expect(server.snapshot().parents).toHaveLength(2) + expect(server.snapshot().parents.filter((parent) => parent.status)).toHaveLength(1) + unavailable.mockClear() + forward.mockClear() + expect(server.replayCachedPayloadsForPanes()).toBe(1) + expect(forward).not.toHaveBeenCalled() + expect(unavailable.mock.lastCall?.[0]).toMatchObject({ + paneKey: TMUX_TEST_PANE, + worktreeId: 'workspace', + launchToken: 'generation', + statusUnavailable: true, + payload: null + }) + probe.mockResolvedValue(null) + await new Promise((resolve) => setTimeout(resolve, 1100)) + unavailable.mockClear() + server.replayCachedPayloadsForPanes() + expect(unavailable).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/relay/agent-hook-canonical-status.ts b/src/relay/agent-hook-canonical-status.ts new file mode 100644 index 000000000000..3ffa906ed376 --- /dev/null +++ b/src/relay/agent-hook-canonical-status.ts @@ -0,0 +1,115 @@ +import { randomUUID } from 'node:crypto' +import { createAgentStatusStore } from '../shared/agent-status-store' +import { isTmuxInnerSubject, type TmuxAgentHookOwner } from '../shared/tmux-agent-hook-owner' +import { readTmuxUnavailable, tmuxCanonicalStatusEvent } from '../shared/tmux-selected-status' +import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' +import type { AgentHookSource } from '../shared/agent-hook-relay' +import { createRelayTmuxHookOwner } from './relay-tmux-hook-owner' +import { buildRelayHookEnvelope } from './agent-hook-envelope-build' +import type { RelayHookServerOptions } from './agent-hook-server-contract' + +/** The hook server owns the authority; legacy ingress cannot write canonical pane subjects. */ +export class RelayAgentHookCanonicalStatus { + protected canonicalStatusStore = createAgentStatusStore({ + epoch: randomUUID(), + mode: 'authority' + }) + private tmuxOwner: TmuxAgentHookOwner | undefined + private options: RelayHookServerOptions | undefined + private getLegacyIdentity: (paneKey: string) => AgentHookEventPayload | undefined = () => + undefined + private clearLegacyProjection: (paneKey: string) => void = () => {} + + protected configureCanonicalHooks( + options: RelayHookServerOptions, + clearLegacy: (paneKey: string) => void, + getLegacyIdentity: (paneKey: string) => AgentHookEventPayload | undefined + ): void { + this.getLegacyIdentity = getLegacyIdentity + this.options = options + this.clearLegacyProjection = clearLegacy + } + + protected startCanonicalHooks(): void { + const options = this.options + if (!options) { + return + } + this.tmuxOwner ??= createRelayTmuxHookOwner({ + store: () => this.canonicalStatusStore, + getRoot: options.getTmuxManagedPty, + isRetired: options.isPaneSurfaceRetired ?? (() => false), + publish: (event) => { + this.clearLegacyProjection(event.paneKey) + options.forward(buildRelayHookEnvelope(event, event.source ?? 'opencode', options.env)) + }, + takeLegacyIdentity: (paneKey) => { + const prior = this.getLegacyIdentity(paneKey) + this.clearLegacyProjection(paneKey) + return prior + }, + forwardUnavailable: options.forwardUnavailable + }) + } + + protected ingestCanonicalTmuxHook( + source: AgentHookSource, + body: unknown, + env: string + ): Promise { + return this.tmuxOwner?.ingest(source, body, env) ?? Promise.resolve(false) + } + + protected isCanonicalPane(paneKey: string): boolean { + return this.canonicalStatusStore + .getParents() + .some((parent) => parent.subject.kind === 'pty' && parent.subject.paneKey === paneKey) + } + + protected clearCanonicalPane(paneKey: string): void { + this.tmuxOwner?.clearPane(paneKey) + for (const parent of this.canonicalStatusStore.getParents()) { + if (parent.subject.kind === 'pty' && parent.subject.paneKey === paneKey) { + this.canonicalStatusStore.applyMutation({ removeParent: parent.subject }) + } + } + } + + protected replayCanonicalHooks(): number { + const options = this.options + if (!options) { + return 0 + } + let count = 0 + for (const parent of this.canonicalStatusStore.getParents()) { + if (parent.subject.kind !== 'pty' || isTmuxInnerSubject(parent.subject)) { + continue + } + if (options.isPaneSurfaceRetired?.(parent.subject.paneKey)) { + this.clearCanonicalPane(parent.subject.paneKey) + continue + } + const unavailable = readTmuxUnavailable(this.canonicalStatusStore, parent.subject) + if (unavailable) { + options.forwardUnavailable?.(unavailable) + } else if (parent.status) { + const event = tmuxCanonicalStatusEvent(parent.status) + options.forward( + buildRelayHookEnvelope(event, event.source ?? 'opencode', options.env, undefined, { + isReplay: true + }) + ) + } else { + continue + } + count++ + } + return count + } + + protected stopCanonicalHooks(): void { + this.tmuxOwner?.stop() + this.tmuxOwner = undefined + this.canonicalStatusStore = createAgentStatusStore({ epoch: randomUUID(), mode: 'authority' }) + } +} diff --git a/src/relay/agent-hook-envelope-build.ts b/src/relay/agent-hook-envelope-build.ts index ba2779c684a1..e85a634f9cb4 100644 --- a/src/relay/agent-hook-envelope-build.ts +++ b/src/relay/agent-hook-envelope-build.ts @@ -16,6 +16,9 @@ export function buildRelayHookEnvelope( ): AgentHookRelayEnvelope { return { source, + ...(event.hostEvidenceObservedAt !== undefined + ? { evidenceAgeMs: Math.max(0, Date.now() - event.hostEvidenceObservedAt) } + : {}), agentPresence: event.agentPresence, paneKey: event.paneKey, ...(event.launchToken ? { launchToken: event.launchToken } : {}), diff --git a/src/relay/agent-hook-envelope-publication.ts b/src/relay/agent-hook-envelope-publication.ts index 96a413a02c70..684ad8f7dde2 100644 --- a/src/relay/agent-hook-envelope-publication.ts +++ b/src/relay/agent-hook-envelope-publication.ts @@ -2,6 +2,7 @@ import { AGENT_HOOK_NOTIFICATION_METHOD, AGENT_HOOK_SHED_FIELDS_KEY, createShedSubagentsField, + type AgentHookUnavailableEnvelope, type AgentHookRelayEnvelope } from '../shared/agent-hook-relay' import type { RelayDispatcher } from './dispatcher' @@ -156,12 +157,27 @@ function logUnsendableEnvelope( * background; one that no shedding can fit is dropped. */ export function publishAgentHookEnvelope( dispatcher: RelayDispatcher, - envelope: AgentHookRelayEnvelope + envelope: AgentHookRelayEnvelope | AgentHookUnavailableEnvelope ): void { const clientIds = dispatcher.activeClientIds() if (clientIds.length === 0) { return } + if (envelope.payload === null) { + const params = { ...envelope } + if (!fitsProducerFrame(dispatcher, params)) { + clearPendingEnvelope(dispatcher, envelope.paneKey) + logUnsendableEnvelope(dispatcher, params, clientIds) + return + } + const rejected = publishToClients(dispatcher, params, clientIds) + if (rejected.length === 0) { + clearPendingEnvelope(dispatcher, envelope.paneKey) + } else { + setPendingEnvelope(dispatcher, envelope.paneKey, params, rejected) + } + return + } // Why: a fan-out must choose one payload before it writes anything, or the first client keeps a // frame a smaller later client forces us to shed. A single sink writes nothing when it rejects, // so there the attempt itself is the measurement — one encode instead of a probe plus a publish. diff --git a/src/relay/agent-hook-request.ts b/src/relay/agent-hook-request.ts index e852f66cf69b..a5bfe9190766 100644 --- a/src/relay/agent-hook-request.ts +++ b/src/relay/agent-hook-request.ts @@ -25,6 +25,7 @@ export async function handleRelayHookRequest( env?: string, version?: string ) => AgentHookEventPayload | undefined + ingestTmuxHook?: (source: AgentHookSource, body: unknown) => Promise retryScheduler: AgentHookResultRetryScheduler transportInterference: ReturnType } @@ -55,6 +56,11 @@ export async function handleRelayHookRequest( } const body = await readRequestBody(req) const hookBody = mergeAgentHookRequestHeaders(body, req.headers) + if (await options.ingestTmuxHook?.(source, hookBody)) { + res.writeHead(204) + res.end() + return + } const event = normalizeHookPayload(options.state, source, hookBody, options.env, { deferCompactOwnershipToClient: true }) diff --git a/src/relay/agent-hook-server-contract.ts b/src/relay/agent-hook-server-contract.ts new file mode 100644 index 000000000000..7f9e1cf439ba --- /dev/null +++ b/src/relay/agent-hook-server-contract.ts @@ -0,0 +1,32 @@ +import type { + AgentHookRelayEnvelope, + AgentHookUnavailableEnvelope +} from '../shared/agent-hook-relay' +import type { TmuxManagedPty } from '../shared/tmux-agent-hook-owner' + +export type RelayHookForward = (envelope: AgentHookRelayEnvelope) => void + +export type RelayHookServerOptions = { + /** Where to put endpoint.env / endpoint.cmd. Defaults to `$HOME/.orca-relay/agent-hooks`. */ + endpointDir?: string + /** Env tag forwarded into hook payloads. Defaults to "remote", which main excludes from dev-vs-prod mismatch warnings. */ + env?: string + /** Fixed auth token. WSL relay passes the host-issued token (already in guest env via WSLENV) so unmodified hook clients authenticate. Defaults to a fresh UUID. */ + token?: string + /** Preferred bind port. WSL relay passes the Windows listener's port so env-sourced client coords stay truthful; falls back to :0 if occupied. Defaults to :0. */ + preferredPort?: number + forwardUnavailable?: (envelope: AgentHookUnavailableEnvelope) => void + forward: RelayHookForward + /** + * True when the host has been told this pane's tab is gone and no PTY has re-bound the paneKey. + * Posts from such a pane come from a process the user already closed, so they describe no surface + * any client owns. Defaults to "never retired", which is the pre-existing behaviour — a listener + * with no PTY handler behind it (the WSL relay) keeps forwarding everything. + */ + getTmuxManagedPty?: (paneKey: string) => Promise + isPaneSurfaceRetired?: (paneKey: string) => boolean +} + +export type RelayHookServerStartOptions = { + publishEndpoint?: boolean +} diff --git a/src/relay/agent-hook-server.ts b/src/relay/agent-hook-server.ts index 126d0470abb0..2b88dfdc86a6 100644 --- a/src/relay/agent-hook-server.ts +++ b/src/relay/agent-hook-server.ts @@ -1,3 +1,14 @@ +import { RelayAgentHookCanonicalStatus } from './agent-hook-canonical-status' +import type { + RelayHookForward, + RelayHookServerOptions, + RelayHookServerStartOptions +} from './agent-hook-server-contract' +export type { + RelayHookForward, + RelayHookServerOptions, + RelayHookServerStartOptions +} from './agent-hook-server-contract' import { handleRelayHookRequest } from './agent-hook-request' import { transitionHookPresence } from '../shared/agent-hook-presence-transition' import { RelayAgentPresence } from './relay-agent-presence' @@ -21,53 +32,20 @@ import { getEndpointFileName, writeEndpointFile } from '../shared/agent-hook-listener/endpoint-publication' -import { normalizeHookPayload } from '../shared/agent-hook-listener' import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' import { createHookTransportInterferenceTracker, describeHookTransportInterference } from '../shared/agent-hook-transport-interference' -import { - isAgentHookSource, - REMOTE_AGENT_HOOK_ENV, - type AgentHookRelayEnvelope, - type AgentHookSource -} from '../shared/agent-hook-relay' -import { - buildSpoolHookBody, - drainAgentHookSpool, - type SpoolRecord -} from '../shared/agent-hook-spool' +import { REMOTE_AGENT_HOOK_ENV, type AgentHookSource } from '../shared/agent-hook-relay' +import { drainAgentHookSpool, type SpoolRecord } from '../shared/agent-hook-spool' import { buildRelayHookPtyEnv, defaultEndpointDir } from './agent-hook-endpoint-coordinates' -import { buildRelayHookEnvelope, hookBodyEnv, hookBodyVersion } from './agent-hook-envelope-build' +import { buildRelayHookEnvelope } from './agent-hook-envelope-build' +import { ingestRelayHookSpoolRecord } from './agent-hook-spool-ingest' import { AgentHookResultRetryScheduler } from './agent-hook-result-retry-scheduler' import { MAX_CACHED_PANES, selectReplayableCachedPanes } from './agent-hook-cached-pane-status' -export type RelayHookForward = (envelope: AgentHookRelayEnvelope) => void - -export type RelayHookServerOptions = { - /** Where to put endpoint.env / endpoint.cmd. Defaults to `$HOME/.orca-relay/agent-hooks`. */ - endpointDir?: string - /** Env tag forwarded into hook payloads. Defaults to "remote", which main excludes from dev-vs-prod mismatch warnings. */ - env?: string - /** Fixed auth token. WSL relay passes the host-issued token (already in guest env via WSLENV) so unmodified hook clients authenticate. Defaults to a fresh UUID. */ - token?: string - /** Preferred bind port. WSL relay passes the Windows listener's port so env-sourced client coords stay truthful; falls back to :0 if occupied. Defaults to :0. */ - preferredPort?: number - forward: RelayHookForward - /** - * True when the host has been told this pane's tab is gone and no PTY has re-bound the paneKey. - * Posts from such a pane come from a process the user already closed, so they describe no surface - * any client owns. Defaults to "never retired", which is the pre-existing behaviour — a listener - * with no PTY handler behind it (the WSL relay) keeps forwarding everything. - */ - isPaneSurfaceRetired?: (paneKey: string) => boolean -} - -export type RelayHookServerStartOptions = { - publishEndpoint?: boolean -} -export class RelayAgentHookServer { +export class RelayAgentHookServer extends RelayAgentHookCanonicalStatus { private server: ReturnType | null = null private port = 0 private token = '' @@ -94,6 +72,7 @@ export class RelayAgentHookServer { private retryScheduler: AgentHookResultRetryScheduler constructor(options: RelayHookServerOptions) { + super() this.env = options.env ?? REMOTE_AGENT_HOOK_ENV this.endpointDir = options.endpointDir ?? defaultEndpointDir() this.endpointFilePath = join(this.endpointDir, getEndpointFileName()) @@ -101,6 +80,16 @@ export class RelayAgentHookServer { this.preferredPort = options.preferredPort ?? 0 this.forward = options.forward this.isPaneSurfaceRetired = options.isPaneSurfaceRetired ?? (() => false) + this.configureCanonicalHooks( + options, + (paneKey) => this.clearPaneState(paneKey, true), + (paneKey) => { + const row = this.state.lastStatusByPaneKey.get(paneKey) + return row + ? { ...row, source: this.lastEnvelopeMetaByPaneKey.get(paneKey)?.source ?? row.source } + : undefined + } + ) this.retryScheduler = new AgentHookResultRetryScheduler({ state: this.state, env: this.env, @@ -115,6 +104,7 @@ export class RelayAgentHookServer { if (this.server) { return } + this.startCanonicalHooks() this.token = this.fixedToken ?? randomUUID() this.endpointFileWritten = false this.portFallbackApplied = false @@ -198,6 +188,7 @@ export class RelayAgentHookServer { this.port = 0 this.token = '' this.endpointFileWritten = false + this.stopCanonicalHooks() this.retryScheduler.clearAll() clearAllListenerCaches(this.state) this.lastEnvelopeMetaByPaneKey.clear() @@ -219,7 +210,7 @@ export class RelayAgentHookServer { buildRelayHookEnvelope(event, meta.source, meta.env, meta.version, { isReplay: true }) ) } - return replayable.length + return replayable.length + this.replayCanonicalHooks() } checkAgentPresence(paneKey: string): Promise { @@ -237,7 +228,10 @@ export class RelayAgentHookServer { } /** Drop a paneKey's cached entries on PTY exit so a terminated pane can't resurface as a ghost event on reconnect. */ - clearPaneState(paneKey: string): void { + clearPaneState(paneKey: string, preserveTmuxInnerSubjects = false): void { + if (!preserveTmuxInnerSubjects) { + this.clearCanonicalPane(paneKey) + } this.retryScheduler.clearAssistantMessageRetry(paneKey) this.retryScheduler.clearTranscriptPoll(paneKey) clearPaneCacheState(this.state, paneKey) @@ -268,6 +262,7 @@ export class RelayAgentHookServer { env: this.env, state: this.state, applyEvent: (event, source, env, version) => this.applyEvent(event, source, env, version), + ingestTmuxHook: (source, body) => this.ingestCanonicalTmuxHook(source, body, this.env), retryScheduler: this.retryScheduler, transportInterference: this.transportInterference }) @@ -280,6 +275,9 @@ export class RelayAgentHookServer { version?: string, options: { isReplay?: boolean; checkPresence?: boolean } = {} ): AgentHookEventPayload | undefined { + if (this.isCanonicalPane(incoming.paneKey)) { + return undefined + } const transitioned = transitionHookPresence( incoming, this.state.lastStatusByPaneKey.get(incoming.paneKey) @@ -331,18 +329,8 @@ export class RelayAgentHookServer { } private ingestSpoolRecord(record: SpoolRecord): void { - if (!isAgentHookSource(record.source)) { - return - } - const body = buildSpoolHookBody(record) - const event = normalizeHookPayload(this.state, record.source, body, this.env, { - deferCompactOwnershipToClient: true - }) - if (!event) { - return - } - this.applyEvent(event, record.source, hookBodyEnv(body), hookBodyVersion(body), { - isReplay: true + ingestRelayHookSpoolRecord(record, this.state, this.env, (event, source, env, version) => { + this.applyEvent(event, source, env, version, { isReplay: true }) }) } } diff --git a/src/relay/agent-hook-spool-ingest.ts b/src/relay/agent-hook-spool-ingest.ts new file mode 100644 index 000000000000..bdc184804266 --- /dev/null +++ b/src/relay/agent-hook-spool-ingest.ts @@ -0,0 +1,29 @@ +import { normalizeHookPayload } from '../shared/agent-hook-listener' +import type { HookListenerState } from '../shared/agent-hook-listener/listener-state' +import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' +import { isAgentHookSource, type AgentHookSource } from '../shared/agent-hook-relay' +import { buildSpoolHookBody, type SpoolRecord } from '../shared/agent-hook-spool' +import { hookBodyEnv, hookBodyVersion } from './agent-hook-envelope-build' + +export function ingestRelayHookSpoolRecord( + record: SpoolRecord, + state: HookListenerState, + env: string, + apply: ( + event: AgentHookEventPayload, + source: AgentHookSource, + env?: string, + version?: string + ) => void +): void { + if (!isAgentHookSource(record.source)) { + return + } + const body = buildSpoolHookBody(record) + const event = normalizeHookPayload(state, record.source, body, env, { + deferCompactOwnershipToClient: true + }) + if (event) { + apply(event, record.source, hookBodyEnv(body), hookBodyVersion(body)) + } +} diff --git a/src/relay/pty-handler.ts b/src/relay/pty-handler.ts index fbad4a212a0d..b80b7ffdbd14 100644 --- a/src/relay/pty-handler.ts +++ b/src/relay/pty-handler.ts @@ -1,3 +1,4 @@ +import type { TmuxManagedPty } from '../shared/tmux-agent-hook-owner' /* oxlint-disable max-lines */ import { resolveSynchronizedOutputSafeSplit } from '../shared/terminal-synchronized-output-scan' import { createTerminalTitleTracker } from '../shared/terminal-output-side-effects' @@ -751,6 +752,29 @@ export class PtyHandler { return this.retiredPaneSurfaces.isRetired(paneKey) } + getTmuxManagedPty(paneKey: string): TmuxManagedPty | null { + if (process.platform === 'win32' || this.isPaneSurfaceRetired(paneKey)) { + return null + } + const candidates = [...this.ptys.values()].filter( + (pty) => !pty.disposed && (pty.paneKey ?? pty.attachIdentity?.paneKey) === paneKey + ) + const root = candidates.length === 1 ? candidates[0] : undefined + if (!root?.worktreeId || !root.pty.pid) { + return null + } + return { + pid: root.pty.pid, + incarnation: root.incarnationId, + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: root.worktreeId, + workspaceKind: root.worktreeId.startsWith('folder:') ? 'folder' : 'git-worktree' + } + } + } + /** Notified when the last PTY leaves the pool, so the relay can re-arm its idle grace. */ onPtyPoolEmpty(listener: () => void): () => void { this.ptyPoolEmptyListener = listener diff --git a/src/relay/relay-agent-hook-runtime.ts b/src/relay/relay-agent-hook-runtime.ts index 1dcda669f7d9..67f29c1cba29 100644 --- a/src/relay/relay-agent-hook-runtime.ts +++ b/src/relay/relay-agent-hook-runtime.ts @@ -39,9 +39,11 @@ export class RelayAgentHookRuntime { this.hookServer = new RelayAgentHookServer({ endpointDir: endpointDir ?? endpointDirForRelaySocket(sockPath), forward: (envelope) => publishAgentHookEnvelope(dispatcher, envelope), + forwardUnavailable: (envelope) => publishAgentHookEnvelope(dispatcher, envelope), // Why: the PTY handler is the only component that knows which panes still have a client // surface, so it — not the client — decides whether a hook post describes a live pane. - isPaneSurfaceRetired: (paneKey) => ptyHandler.isPaneSurfaceRetired(paneKey) + isPaneSurfaceRetired: (paneKey) => ptyHandler.isPaneSurfaceRetired(paneKey), + getTmuxManagedPty: async (paneKey) => ptyHandler.getTmuxManagedPty(paneKey) }) } diff --git a/src/relay/relay-tmux-hook-owner.ts b/src/relay/relay-tmux-hook-owner.ts new file mode 100644 index 000000000000..478b4d2a20c9 --- /dev/null +++ b/src/relay/relay-tmux-hook-owner.ts @@ -0,0 +1,46 @@ +import type { AgentStatusStore } from '../shared/agent-status-store' +import { commitTmuxSelectedStatus, commitTmuxUnavailable } from '../shared/tmux-selected-status' +import { TmuxAgentHookOwner, type TmuxManagedPty } from '../shared/tmux-agent-hook-owner' +import type { AgentHookEventPayload } from '../shared/agent-hook-listener/listener-event' +import type { AgentHookUnavailableEnvelope } from '../shared/agent-hook-relay' + +export function createRelayTmuxHookOwner(options: { + getRoot?: (paneKey: string) => Promise + isRetired: (paneKey: string) => boolean + store: () => AgentStatusStore + publish: (event: AgentHookEventPayload) => void + takeLegacyIdentity: (paneKey: string) => AgentHookEventPayload | undefined + forwardUnavailable?: (envelope: AgentHookUnavailableEnvelope) => void +}): TmuxAgentHookOwner | undefined { + if (!options.getRoot) { + return undefined + } + return new TmuxAgentHookOwner({ + store: options.store, + getRoot: options.getRoot, + isRetired: options.isRetired, + publish: (event, observedAt, subject, stateStartedAt) => { + const status = commitTmuxSelectedStatus( + options.store(), + subject, + event, + observedAt, + stateStartedAt + ) + if (!status) { + return + } + options.publish({ ...event, hostEvidenceObservedAt: status.evidenceObservedAt }) + }, + unavailable: (paneKey, subject, identity) => { + if (!subject) { + return + } + const prior = options.takeLegacyIdentity(paneKey) + const envelope = commitTmuxUnavailable(options.store(), subject, prior ?? identity) + if (envelope) { + options.forwardUnavailable?.(envelope) + } + } + }) +} diff --git a/src/renderer/src/hooks/ipc-events/agent-status-listeners.ts b/src/renderer/src/hooks/ipc-events/agent-status-listeners.ts index 70b13e22a41f..62d9f9c3f036 100644 --- a/src/renderer/src/hooks/ipc-events/agent-status-listeners.ts +++ b/src/renderer/src/hooks/ipc-events/agent-status-listeners.ts @@ -82,7 +82,10 @@ export function registerAgentStatusListeners(args: { } } const store = useAppStore.getState() - if (store.agentStatusByPaneKey[data.paneKey]?.state === 'done') { + if ( + data.statusUnavailable !== true && + store.agentStatusByPaneKey[data.paneKey]?.state === 'done' + ) { return } store.removeAgentStatus(data.paneKey) diff --git a/src/renderer/src/hooks/ipc-events/agent-status-unavailable.test.ts b/src/renderer/src/hooks/ipc-events/agent-status-unavailable.test.ts new file mode 100644 index 000000000000..b3924a583ea5 --- /dev/null +++ b/src/renderer/src/hooks/ipc-events/agent-status-unavailable.test.ts @@ -0,0 +1,42 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { AgentStatusClearIpcPayload } from '../../../../shared/agent-status-types' +import { registerAgentStatusListeners } from './agent-status-listeners' + +const store = vi.hoisted(() => ({ + agentStatusByPaneKey: { pane: { state: 'done' } }, + removeAgentStatus: vi.fn() +})) +vi.mock('../../store', () => ({ useAppStore: { getState: () => store } })) +afterEach(() => { + vi.unstubAllGlobals() + vi.clearAllMocks() +}) + +describe('host-selected status unavailable', () => { + it('removes a completed projection only for an explicit host unavailable clear', () => { + let clear: ((event: AgentStatusClearIpcPayload) => void) | undefined + vi.stubGlobal('window', { + api: { + agentStatus: { + onSet: () => () => {}, + onClear: (callback: typeof clear) => { + clear = callback + return () => {} + } + } + } + }) + registerAgentStatusListeners({ + unsubs: [], + enqueueLiveAgentStatus: vi.fn(), + drainQueuedLiveAgentStatusesForPane: vi.fn(), + pendingAgentStatusEvents: [], + transientClearWatermarkByConnectionId: new Map(), + liveAgentStatusBurstQueue: [] + }) + clear?.({ paneKey: 'pane' }) + expect(store.removeAgentStatus).not.toHaveBeenCalled() + clear?.({ paneKey: 'pane', statusUnavailable: true }) + expect(store.removeAgentStatus).toHaveBeenCalledWith('pane') + }) +}) diff --git a/src/shared/agent-hook-listener.ts b/src/shared/agent-hook-listener.ts index 99b78fcfc3e3..3932164f4c22 100644 --- a/src/shared/agent-hook-listener.ts +++ b/src/shared/agent-hook-listener.ts @@ -1,5 +1,5 @@ import { readAgentProcessIdentity } from './agent-process-presence' -import { normalizeAgentStatusPayload } from './agent-status-types' +import { normalizeAgentStatusPayload, type AgentMainAgentStatus } from './agent-status-types' import type { AgentHookSource } from './agent-hook-relay' import { extractAgentProviderSession } from './agent-session-resume' import { @@ -37,7 +37,10 @@ export function normalizeHookPayload( source: AgentHookSource, body: unknown, expectedEnv: string, - options: { deferCompactOwnershipToClient?: boolean } = {} + options: { + deferCompactOwnershipToClient?: boolean + previousOpenCodeMainAgent?: AgentMainAgentStatus + } = {} ): AgentHookEventPayload | null { const envelope = parseHookEnvelope(state, source, body, expectedEnv) if (!envelope) { @@ -189,7 +192,8 @@ export function normalizeHookPayload( paneKey, hookPayload: hookPayloadRecord, envelope: record, - extractedPrompt + extractedPrompt, + previousOpenCodeMainAgent: options.previousOpenCodeMainAgent }) const providerSessionOnly = (source === 'pi' || source === 'prime-agent') && diff --git a/src/shared/agent-hook-listener/listener-event.ts b/src/shared/agent-hook-listener/listener-event.ts index a954313cf158..8157b47c798f 100644 --- a/src/shared/agent-hook-listener/listener-event.ts +++ b/src/shared/agent-hook-listener/listener-event.ts @@ -52,6 +52,8 @@ export type AgentHookEventPayload = { * Lets a reader rejoin the row to its terminal after the pane key moved. Never persisted: * a handle belongs to the runtime that issued it. */ terminalHandle?: string + /** Execution-host clock retained for age-preserving projection replay; never sent verbatim. */ + hostEvidenceObservedAt?: number payload: ParsedAgentStatusPayload } diff --git a/src/shared/agent-hook-listener/provider-dispatch.ts b/src/shared/agent-hook-listener/provider-dispatch.ts index ab7e847c4cad..b7fffa09f164 100644 --- a/src/shared/agent-hook-listener/provider-dispatch.ts +++ b/src/shared/agent-hook-listener/provider-dispatch.ts @@ -1,6 +1,6 @@ import { normalizeCompatibleLifecycleEvent } from './providers/compatible-lifecycle-events' import { normalizeQoderEvent } from './providers/qoder-events' -import type { ParsedAgentStatusPayload } from '../agent-status-types' +import type { AgentMainAgentStatus, ParsedAgentStatusPayload } from '../agent-status-types' import type { AgentHookSource } from '../agent-hook-relay' import { readLastCommandCodeUserPromptEntryFromTranscript } from './command-code-transcript' import { readGrokHomeEnvelope } from './grok-result-discovery' @@ -45,6 +45,7 @@ export function normalizeProviderEvent(input: { hookPayload: Record envelope: Record extractedPrompt: ExtractedPromptText + previousOpenCodeMainAgent?: AgentMainAgentStatus }): ProviderDispatchResult { const { state, source, eventName, promptText, paneKey, hookPayload, envelope, extractedPrompt } = input @@ -101,7 +102,8 @@ export function normalizeProviderEvent(input: { eventName, promptText, paneKey, - hookPayload + hookPayload, + input.previousOpenCodeMainAgent ) break } diff --git a/src/shared/agent-hook-listener/providers/opencode-family-events.ts b/src/shared/agent-hook-listener/providers/opencode-family-events.ts index ba64837e42b4..55b3e7753467 100644 --- a/src/shared/agent-hook-listener/providers/opencode-family-events.ts +++ b/src/shared/agent-hook-listener/providers/opencode-family-events.ts @@ -1,6 +1,7 @@ import { continueMainAgentStatus } from '../../agent-lead-status-fold' import { normalizeAgentStatusPayload, + type AgentMainAgentStatus, type ParsedAgentStatusPayload } from '../../agent-status-types' import type { HookListenerState } from '../listener-state' @@ -13,7 +14,8 @@ export function normalizeOpenCodeFamilyEvent( eventName: unknown, promptText: string, paneKey: string, - hookPayload: Record + hookPayload: Record, + previousMainAgent?: AgentMainAgentStatus ): ParsedAgentStatusPayload | null { const resetsTurn = isNewTurnEvent(source, eventName) || @@ -50,7 +52,7 @@ export function normalizeOpenCodeFamilyEvent( ? continueMainAgentStatus( eventName === 'SessionStart' ? undefined - : state.lastStatusByPaneKey.get(paneKey)?.payload.mainAgent, + : (previousMainAgent ?? state.lastStatusByPaneKey.get(paneKey)?.payload.mainAgent), { state: rootState, outcome: diff --git a/src/shared/agent-hook-relay.ts b/src/shared/agent-hook-relay.ts index 4a29fca868bf..6da6f2b2df0e 100644 --- a/src/shared/agent-hook-relay.ts +++ b/src/shared/agent-hook-relay.ts @@ -123,9 +123,22 @@ export type AgentHookRelayEnvelope = { version?: string /** Pre-normalized status payload from the relay's `normalizeHookPayload`. * Orca's `ingestRemote` validates it again at the SSH trust boundary. */ + evidenceAgeMs?: number payload: ParsedAgentStatusPayload } +/** Older clients ignore the null payload; newer clients clear only the selected projection. */ +export type AgentHookUnavailableEnvelope = { + source: 'opencode' | 'opencode2' + paneKey: string + tabId?: string + worktreeId?: string + launchToken?: string + connectionId: null + statusUnavailable: true + payload: null +} + /** JSON-RPC notification method name carried over the relay control channel. */ export const AGENT_HOOK_NOTIFICATION_METHOD = 'agent.hook' as const diff --git a/src/shared/agent-status-ipc-payload.ts b/src/shared/agent-status-ipc-payload.ts index 68398a6d0def..59803d498702 100644 --- a/src/shared/agent-status-ipc-payload.ts +++ b/src/shared/agent-status-ipc-payload.ts @@ -84,7 +84,7 @@ export type AgentStatusCacheIdentity = { /** Wire shape for ordinary pane teardown or a stamped SSH disconnect batch. */ export type AgentStatusClearIpcPayload = - | { paneKey: string } + | { paneKey: string; statusUnavailable?: true } | { transient: true connectionId: string diff --git a/src/shared/agent-status-store-parent.test.ts b/src/shared/agent-status-store-parent.test.ts new file mode 100644 index 000000000000..62513faf3b0c --- /dev/null +++ b/src/shared/agent-status-store-parent.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest' +import { parseAgentStatusParentInput } from './agent-status-store-parent' +import { TMUX_TEST_PANE, TMUX_TEST_ROOT } from './tmux-status.test-fixture' +const subject = { ...TMUX_TEST_ROOT.scope, kind: 'pty', paneKey: TMUX_TEST_PANE } as const + +describe('canonical PTY attachment without a status claim', () => { + it('retains the subject when its owner reports no selected status', () => { + expect(parseAgentStatusParentInput({ subject })).toEqual({ subject }) + }) + it('still refuses a status claiming another pane', () => { + expect( + parseAgentStatusParentInput({ + subject, + status: { + paneKey: 'foreign', + connectionId: null, + worktreeId: 'workspace', + state: 'done', + prompt: '', + receivedAt: 1, + stateStartedAt: 1 + } + }) + ).toBeNull() + }) +}) diff --git a/src/shared/agent-status-store-parent.ts b/src/shared/agent-status-store-parent.ts index 12682becb9d7..86efe39fc4e8 100644 --- a/src/shared/agent-status-store-parent.ts +++ b/src/shared/agent-status-store-parent.ts @@ -50,7 +50,7 @@ function isParentScopeConsistent(parent: AgentStatusParentInput): boolean { if (status?.worktreeId !== undefined && status.worktreeId !== subject.workspaceId) { return false } - if (subject.kind === 'pty' && status?.paneKey !== subject.paneKey) { + if (subject.kind === 'pty' && status && status.paneKey !== subject.paneKey) { return false } if (subject.kind === 'pty-run' && status?.runId !== undefined && status.runId !== subject.runId) { diff --git a/src/shared/tmux-agent-hook-owner.test.ts b/src/shared/tmux-agent-hook-owner.test.ts new file mode 100644 index 000000000000..0222ee2ae97f --- /dev/null +++ b/src/shared/tmux-agent-hook-owner.test.ts @@ -0,0 +1,231 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { createAgentStatusStore } from './agent-status-store' +import type { ProcessTableRow } from './process-table-snapshot' +import { TmuxAgentHookOwner, type TmuxManagedPty } from './tmux-agent-hook-owner' + +const paneKey = '11111111-1111-4111-8111-111111111111:22222222-2222-4222-8222-222222222222' +const root: TmuxManagedPty = { + pid: 100, + incarnation: 'first', + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace', + workspaceKind: 'folder' + } +} +const rows: ProcessTableRow[] = [ + { + pid: 100, + ppid: 1, + pgid: 100, + tpgid: 101, + tty: 'pts/1', + stat: 'S', + startTime: 'root', + command: '/bin/bash' + }, + { + pid: 101, + ppid: 100, + pgid: 101, + tpgid: 101, + tty: 'pts/1', + stat: 'S+', + startTime: 'client', + command: '/usr/bin/tmux attach' + } +] +function fixture() { + let now = 1000 + let selected = '%0' + let unverifiable = false + let noClient = false + let current: TmuxManagedPty | null = root + let retired = false + const store = createAgentStatusStore({ epoch: 'tmux-test', mode: 'authority' }) + const publish = vi.fn() + const unavailable = vi.fn() + const probe = vi.fn(async (_socket: string, _roots: readonly number[]) => + unverifiable ? null : { clients: noClient ? [] : [{ pid: 101, pane: selected }], rows } + ) + const owner = new TmuxAgentHookOwner({ + store: () => store, + getRoot: async () => current, + publish, + unavailable, + probe, + now: () => now, + isRetired: () => retired + }) + const ingest = (pane: string, event: string, prompt: string) => + owner.ingest( + 'opencode', + { + paneKey, + worktreeId: 'workspace', + tmux: { socket: '/tmp/test.sock', pane }, + payload: { hook_event_name: event, prompt } + }, + 'dev' + ) + return { + store, + publish, + unavailable, + probe, + owner, + ingest, + advance: () => { + now += 1000 + }, + select: (pane: string) => { + selected = pane + }, + detach: () => { + noClient = true + }, + disconnect: () => { + unverifiable = true + current = null + }, + replace: () => { + current = { ...root, incarnation: 'second' } + }, + retire: () => { + retired = true + } + } +} +const owners: TmuxAgentHookOwner[] = [] +afterEach(() => { + for (const owner of owners.splice(0)) { + owner.stop() + } +}) +function setup() { + const f = fixture() + owners.push(f.owner) + return f +} + +describe('tmux canonical hook ownership', () => { + it('keeps inactive observations canonical and projects only the attached pane with its original age', async () => { + const f = setup() + await f.ingest('%0', 'SessionIdle', 'visible completed turn') + expect(f.publish.mock.lastCall?.[0].payload.state).toBe('done') + f.advance() + await f.ingest('%1', 'PermissionRequest', 'inactive permission') + expect(f.store.getParents()).toHaveLength(2) + expect(f.publish).toHaveBeenCalledTimes(1) + f.advance() + f.select('%1') + await f.owner.refresh() + expect(f.publish.mock.lastCall?.[0].payload.state).toBe('waiting') + expect(f.publish.mock.lastCall?.[1]).toBe(2000) + f.advance() + await f.owner.refresh() + expect(f.publish).toHaveBeenCalledTimes(2) + expect(f.publish.mock.lastCall?.[1]).toBe(2000) + }) + + it('clears an unknown selected pane without inventing completion, but retains evidence on lost contact', async () => { + const f = setup() + await f.ingest('%0', 'SessionBusy', 'working') + f.advance() + f.disconnect() + await f.owner.refresh() + expect(f.unavailable).not.toHaveBeenCalled() + expect(f.store.getParents()).toHaveLength(1) + const other = setup() + await other.ingest('%0', 'SessionBusy', 'working') + other.advance() + other.select('%9') + await other.owner.refresh() + expect(other.unavailable).toHaveBeenCalledWith( + paneKey, + expect.objectContaining({ kind: 'pty', paneKey }), + expect.objectContaining({ source: 'opencode' }) + ) + expect(other.publish).toHaveBeenCalledTimes(1) + }) + + it('clears the attachment on a successful empty client proof while preserving inner observations', async () => { + const f = setup() + await f.ingest('%0', 'SessionIdle', 'completed') + f.advance() + f.detach() + await f.owner.refresh() + expect(f.unavailable).toHaveBeenCalledTimes(1) + expect(f.store.getParents()).toHaveLength(1) + f.advance() + await f.owner.refresh() + expect(f.unavailable).toHaveBeenCalledTimes(1) + }) + + it('visits every socket through bounded rounds instead of skipping inventories larger than sixteen', async () => { + const f = setup() + for (let index = 0; index < 34; index++) { + await f.owner.ingest( + 'opencode', + { + paneKey: `tab-${index}:22222222-2222-4222-8222-222222222222`, + worktreeId: 'workspace', + tmux: { socket: `/tmp/socket-${index}`, pane: '%0' }, + payload: { hook_event_name: 'SessionBusy', prompt: 'work' } + }, + 'dev' + ) + } + for (let round = 0; round < 3; round++) { + f.advance() + await f.owner.refresh() + } + expect(new Set(f.probe.mock.calls.map((call) => call[0])).size).toBe(34) + expect(f.probe).toHaveBeenCalledTimes(49) + }) + + it('retires inner observations on certified root replacement or outer exit', async () => { + const f = setup() + await f.ingest('%0', 'SessionBusy', 'working') + f.advance() + f.replace() + await f.owner.refresh() + expect(f.store.getParents()).toHaveLength(0) + expect(f.unavailable).toHaveBeenCalledWith( + paneKey, + expect.objectContaining({ kind: 'pty', paneKey }), + expect.objectContaining({ source: 'opencode' }) + ) + const other = setup() + await other.ingest('%0', 'SessionBusy', 'working') + other.owner.clearPane(paneKey) + expect(other.store.getParents()).toHaveLength(0) + other.retire() + await other.ingest('%1', 'SessionBusy', 'late event') + expect(other.store.getParents()).toHaveLength(0) + }) + + it('rejects a hook retired while authenticated PTY resolution is pending', async () => { + const f = setup() + const ingest = f.ingest('%0', 'SessionBusy', 'late hook') + f.retire() + await ingest + expect(f.store.getParents()).toHaveLength(0) + expect(f.publish).not.toHaveBeenCalled() + }) + + it('throttles repeated hooks and coalesces concurrent attachment captures', async () => { + const f = setup() + await f.ingest('%0', 'SessionBusy', 'working') + await Promise.all([ + f.owner.refresh(), + f.owner.refresh(), + f.ingest('%1', 'SessionBusy', 'other') + ]) + expect(f.probe).toHaveBeenCalledTimes(1) + f.advance() + await Promise.all([f.owner.refresh(), f.owner.refresh()]) + expect(f.probe).toHaveBeenCalledTimes(2) + }) +}) diff --git a/src/shared/tmux-agent-hook-owner.ts b/src/shared/tmux-agent-hook-owner.ts new file mode 100644 index 000000000000..8843fe31de1c --- /dev/null +++ b/src/shared/tmux-agent-hook-owner.ts @@ -0,0 +1,312 @@ +import { normalizeHookPayload } from './agent-hook-listener' +import { + createHookListenerState, + type HookListenerState +} from './agent-hook-listener/listener-state' +import type { AgentHookEventPayload } from './agent-hook-listener/listener-event' +import type { AgentHookSource } from './agent-hook-relay' +import type { AgentStatusStore } from './agent-status-store' +import { continueMainAgentStatus } from './agent-lead-status-fold' +import { tmuxCanonicalStatusEvent, tmuxInnerSubject } from './tmux-selected-status' +import type { AgentStatusExecutionScope, AgentStatusPtySubject } from './agent-status-subject' +import { + readTmuxHookPane, + resolveTmuxClientAttachment, + type TmuxHookPane +} from './tmux-client-attachment' +import { probeTmuxHostAttachments } from './tmux-host-attachment-probe' +export { isTmuxInnerSubject } from './tmux-selected-status' + +export type TmuxManagedPty = { + pid: number + incarnation: string + scope: AgentStatusExecutionScope +} +type OuterPane = { + paneKey: string + socket: string + root: TmuxManagedPty + inner: Map + selection?: string + publication?: string +} +/** Inner observations live in the hook owner's canonical store; this index holds attachments only. */ +export class TmuxAgentHookOwner { + private readonly outers = new Map() + private timer: ReturnType | undefined + private refreshing: Promise | undefined + private stopped = false + private lastRefreshAt = -Infinity + private socketCursor = 0 + + constructor( + private readonly options: { + store: () => AgentStatusStore + getRoot: (paneKey: string) => Promise + publish: ( + event: AgentHookEventPayload, + observedAt: number, + subject: AgentStatusPtySubject, + stateStartedAt: number + ) => void + unavailable: ( + paneKey: string, + subject?: AgentStatusPtySubject, + identity?: AgentHookEventPayload + ) => void + probe?: typeof probeTmuxHostAttachments + isRetired?: (paneKey: string) => boolean + now?: () => number + } + ) {} + + private now(): number { + return this.options.now?.() ?? Date.now() + } + + async ingest(source: AgentHookSource, body: unknown, env: string): Promise { + if (source !== 'opencode' && source !== 'opencode2') { + return false + } + if (typeof body !== 'object' || body === null || !('tmux' in body)) { + return false + } + const tmux = readTmuxHookPane(body.tmux) + if (!tmux || !('paneKey' in body) || typeof body.paneKey !== 'string' || this.stopped) { + return true + } + const paneKey = body.paneKey + const root = await this.options.getRoot(paneKey).catch(() => null) + if ( + !root || + this.stopped || + this.options.isRetired?.(paneKey) || + !('worktreeId' in body) || + body.worktreeId !== root.scope.workspaceId + ) { + return true + } + let outer = this.outers.get(paneKey) + if ( + outer && + (outer.root.incarnation !== root.incarnation || + outer.root.pid !== root.pid || + outer.socket !== tmux.socket) + ) { + this.clearPane(paneKey) + outer = undefined + } + if (!outer) { + if (this.outers.size >= 64) { + return true + } + outer = { paneKey, socket: tmux.socket, root, inner: new Map() } + this.outers.set(paneKey, outer) + } + const inner = this.inner(outer, tmux) + if (!inner) { + return true + } + const store = this.options.store() + const previousParent = store.getParent(inner.subject) + const previous = previousParent?.status + const event = normalizeHookPayload(inner.normalization, source, body, env, { + previousOpenCodeMainAgent: previous?.mainAgent + }) + if (!event || event.paneKey !== paneKey) { + return true + } + const observedAt = this.now() + const stateStartedAt = + previous?.state === event.payload.state && previous.workingMode === event.payload.workingMode + ? previous.stateStartedAt + : observedAt + store.applyMutation({ + parent: { + subject: inner.subject, + firstObservedAt: previousParent?.firstObservedAt ?? observedAt, + status: { + ...event.payload, + ...(event.payload.mainAgent + ? { + mainAgent: continueMainAgentStatus( + previous?.mainAgent, + event.payload.mainAgent, + observedAt + ) + } + : {}), + paneKey: inner.subject.paneKey, + tabId: event.tabId, + worktreeId: root.scope.workspaceId, + connectionId: null, + receivedAt: observedAt, + evidenceObservedAt: observedAt, + stateStartedAt, + providerSession: event.providerSession, + promptInteractionKey: event.promptInteractionKey, + launchToken: event.launchToken + } + } + }) + if (!this.timer) { + this.timer = setInterval(() => { + void this.refresh() + }, 1000) + this.timer.unref?.() + } + this.project(outer) + await this.refresh() + return true + } + + private inner(outer: OuterPane, tmux: TmuxHookPane) { + let inner = outer.inner.get(tmux.pane) + if (!inner && outer.inner.size < 32) { + const subject = tmuxInnerSubject(outer.root.scope, outer.paneKey, tmux) + inner = { subject, normalization: createHookListenerState() } + outer.inner.set(tmux.pane, inner) + } + return inner + } + + refresh(): Promise { + if (this.refreshing) { + return this.refreshing + } + if (this.stopped || this.now() - this.lastRefreshAt < 1000) { + return Promise.resolve() + } + this.lastRefreshAt = this.now() + const work = this.refreshAttachments().finally(() => { + if (this.refreshing === work) { + this.refreshing = undefined + } + }) + this.refreshing = work + return work + } + + private async refreshAttachments(): Promise { + const groups = new Map() + for (const outer of this.outers.values()) { + const group = groups.get(outer.socket) ?? [] + group.push(outer) + groups.set(outer.socket, group) + } + const entries = [...groups] + const start = this.socketCursor % Math.max(entries.length, 1) + const selected = entries.slice(start).concat(entries.slice(0, start)).slice(0, 16) + this.socketCursor = (start + selected.length) % Math.max(entries.length, 1) + for (let index = 0; index < selected.length; index += 2) { + await Promise.all( + selected.slice(index, index + 2).map(async ([socket, outers]) => { + const proof = await (this.options.probe ?? probeTmuxHostAttachments)( + socket, + outers.map((outer) => outer.root.pid) + ).catch(() => null) + if (!proof || this.stopped) { + return + } + for (const outer of outers) { + if (this.outers.get(outer.paneKey) !== outer) { + continue + } + const current = await this.options.getRoot(outer.paneKey).catch(() => null) + if (this.stopped || this.outers.get(outer.paneKey) !== outer) { + continue + } + if (this.options.isRetired?.(outer.paneKey)) { + this.clearPane(outer.paneKey) + continue + } + if (!current) { + continue + } + if (current.incarnation !== outer.root.incarnation || current.pid !== outer.root.pid) { + this.unavailable(outer) + this.clearPane(outer.paneKey) + continue + } + const client = resolveTmuxClientAttachment(outer.root.pid, proof.clients, proof.rows) + if (!client) { + outer.selection = undefined + if (outer.publication !== 'unattached') { + outer.publication = 'unattached' + this.unavailable(outer) + } + continue + } + outer.selection = client.pane + this.project(outer) + } + }) + ) + } + } + + private unavailable(outer: OuterPane): void { + const status = [...outer.inner.values()] + .map(({ subject }) => this.options.store().getParent(subject)?.status) + .find(Boolean) + this.options.unavailable( + outer.paneKey, + { ...outer.root.scope, kind: 'pty', paneKey: outer.paneKey }, + status ? tmuxCanonicalStatusEvent(status) : undefined + ) + } + + private project(outer: OuterPane): void { + if (!outer.selection || this.outers.get(outer.paneKey) !== outer) { + return + } + const inner = outer.inner.get(outer.selection) + const parent = inner && this.options.store().getParent(inner.subject) + const status = parent?.status + const key = `${outer.selection}:${parent?.revision ?? 'unavailable'}` + if (outer.publication === key) { + return + } + outer.publication = key + if (!status) { + this.unavailable(outer) + return + } + this.options.publish( + { ...tmuxCanonicalStatusEvent(status), paneKey: outer.paneKey }, + status.evidenceObservedAt ?? status.receivedAt, + { ...outer.root.scope, kind: 'pty', paneKey: outer.paneKey }, + status.stateStartedAt + ) + } + + clearPane(paneKey: string): void { + const outer = this.outers.get(paneKey) + if (!outer) { + return + } + this.outers.delete(paneKey) + for (const { subject } of outer.inner.values()) { + this.options.store().applyMutation({ removeParent: subject }) + } + if (this.outers.size === 0) { + clearInterval(this.timer) + this.timer = undefined + } + } + + clearTab(tabId: string): void { + for (const paneKey of this.outers.keys()) { + if (paneKey.startsWith(`${tabId}:`)) { + this.clearPane(paneKey) + } + } + } + + stop(): void { + this.stopped = true + for (const paneKey of this.outers.keys()) { + this.clearPane(paneKey) + } + } +} diff --git a/src/shared/tmux-client-attachment.test.ts b/src/shared/tmux-client-attachment.test.ts new file mode 100644 index 000000000000..caf725c0d5bb --- /dev/null +++ b/src/shared/tmux-client-attachment.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, it } from 'vitest' +import type { ProcessTableRow } from './process-table-snapshot' +import { + parseTmuxAttachedClients, + readTmuxHookPane, + resolveTmuxClientAttachment +} from './tmux-client-attachment' + +const ROOT: ProcessTableRow = { + pid: 100, + ppid: 1, + pgid: 100, + tpgid: 101, + tty: 'ttys149', + startTime: 'root-start', + stat: 'S', + command: '/bin/bash' +} +const CLIENT: ProcessTableRow = { + ...ROOT, + pid: 101, + ppid: 100, + pgid: 101, + stat: 'S+', + startTime: 'client-start', + command: '/opt/homebrew/bin/tmux -S /private/tmp/status.sock attach' +} + +describe('execution-host tmux client attachment', () => { + it('selects by the outer foreground client, independently of another attached client', () => { + const clients = [ + { pid: 101, pane: '%0' }, + { pid: 201, pane: '%1' } + ] + const otherRoot = { ...ROOT, pid: 200, tpgid: 201, tty: 'ttys150' } + const otherClient = { ...CLIENT, pid: 201, ppid: 200, pgid: 201, tty: 'ttys150' } + const rows = [ROOT, CLIENT, otherRoot, otherClient] + expect(resolveTmuxClientAttachment(100, clients, rows)?.pane).toBe('%0') + expect(resolveTmuxClientAttachment(200, clients, rows)?.pane).toBe('%1') + }) + + it('accepts tmux replacing the managed shell itself', () => { + expect(resolveTmuxClientAttachment(101, [{ pid: 101, pane: '%2' }], [CLIENT])?.pane).toBe('%2') + }) + + it.each([ + { tty: 'ttys150' }, + { pgid: 103 }, + { ppid: 999 }, + { startTime: undefined }, + { command: '/bin/bash tmux' }, + { stat: 'T+' } + ])('refuses a client without the full foreground attachment proof: %j', (changes) => { + expect( + resolveTmuxClientAttachment( + 100, + [{ pid: 101, pane: '%0' }], + [ROOT, { ...CLIENT, ...changes }] + ) + ).toBeNull() + }) + + it('refuses ambiguous clients and broken parent chains', () => { + expect( + resolveTmuxClientAttachment( + 100, + [ + { pid: 101, pane: '%0' }, + { pid: 101, pane: '%1' } + ], + [ROOT, CLIENT] + ) + ).toBeNull() + const cycle = { ...CLIENT, ppid: 102 } + expect( + resolveTmuxClientAttachment( + 100, + [{ pid: 101, pane: '%0' }], + [ROOT, cycle, { ...ROOT, pid: 102, ppid: 101 }] + ) + ).toBeNull() + }) + + it('parses bounded client output without accepting a partial malformed table', () => { + expect(parseTmuxAttachedClients('101:%0\n201:%1\n')).toEqual([ + { pid: 101, pane: '%0' }, + { pid: 201, pane: '%1' } + ]) + expect(parseTmuxAttachedClients('101:%0\ngarbage')).toBeNull() + expect(parseTmuxAttachedClients('')).toEqual([]) + }) + + it('accepts a socket containing commas but refuses malformed provider metadata', () => { + expect(readTmuxHookPane({ socket: '/tmp/a,b.sock', pane: '%1' })).toEqual({ + socket: '/tmp/a,b.sock', + pane: '%1' + }) + expect(readTmuxHookPane({ socket: 'relative.sock', pane: '%1' })).toBeNull() + expect(readTmuxHookPane({ socket: '/tmp/x\n.sock', pane: '%1' })).toBeNull() + expect(readTmuxHookPane({ socket: '/tmp/x.sock', pane: '%1;run' })).toBeNull() + }) +}) diff --git a/src/shared/tmux-client-attachment.ts b/src/shared/tmux-client-attachment.ts new file mode 100644 index 000000000000..6184ebfc247f --- /dev/null +++ b/src/shared/tmux-client-attachment.ts @@ -0,0 +1,80 @@ +import type { ProcessTableRow } from './process-table-snapshot' + +export type TmuxHookPane = { socket: string; pane: string } +export type TmuxAttachedClient = { pid: number; pane: string } + +export function readTmuxHookPane(value: unknown): TmuxHookPane | null { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + return null + } + if (!('socket' in value) || !('pane' in value)) { + return null + } + const { socket, pane } = value + if ( + typeof socket !== 'string' || + !socket.startsWith('/') || + socket.length > 1024 || + /[\0\r\n]/.test(socket) || + typeof pane !== 'string' || + !/^%\d{1,12}$/.test(pane) + ) { + return null + } + return { socket, pane } +} + +export function parseTmuxAttachedClients(output: string): TmuxAttachedClient[] | null { + if (output.length > 65536) { + return null + } + const clients: TmuxAttachedClient[] = [] + for (const line of output.trim().split('\n')) { + if (!line) { + continue + } + const match = /^(\d+):(%\d{1,12})$/.exec(line) + const pid = Number(match?.[1]) + if (!match || !Number.isSafeInteger(pid) || pid <= 0 || clients.length >= 128) { + return null + } + clients.push({ pid, pane: match[2] }) + } + return clients +} + +/** Only this outer terminal's foreground client selects its projected inner pane. */ +export function resolveTmuxClientAttachment( + rootPid: number, + clients: readonly TmuxAttachedClient[], + rows: readonly ProcessTableRow[] +): TmuxAttachedClient | null { + const byPid = new Map(rows.map((row) => [row.pid, row])) + const root = byPid.get(rootPid) + if (!root?.startTime || !root.tty || root.tty === '?' || !root.tpgid || root.tpgid <= 0) { + return null + } + const attached = clients.filter((client) => { + const row = byPid.get(client.pid) + if ( + !row?.startTime || + row.tty !== root.tty || + row.pgid !== root.tpgid || + !/^(?:\S*\/)?tmux(?:\s|$)/.test(row.command) || + /[ZT]/.test(row.stat) + ) { + return false + } + const visited = new Set() + let current: ProcessTableRow | undefined = row + while (current && visited.size < 32 && !visited.has(current.pid)) { + if (current.pid === rootPid) { + return true + } + visited.add(current.pid) + current = byPid.get(current.ppid) + } + return false + }) + return attached.length === 1 ? attached[0] : null +} diff --git a/src/shared/tmux-hook-refresh-retirement.test.ts b/src/shared/tmux-hook-refresh-retirement.test.ts new file mode 100644 index 000000000000..47cd3f8fa66e --- /dev/null +++ b/src/shared/tmux-hook-refresh-retirement.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, it, vi } from 'vitest' +import { createAgentStatusStore } from './agent-status-store' +import { TmuxAgentHookOwner } from './tmux-agent-hook-owner' +import { commitTmuxSelectedStatus } from './tmux-selected-status' +import { TMUX_TEST_PANE, TMUX_TEST_ROOT, tmuxTestBody } from './tmux-status.test-fixture' +import type { AgentStatusPtySubject } from './agent-status-subject' + +describe('tmux refresh retirement during root resolution', () => { + it.each([ + ['clear-pane', 'same-root'], + ['clear-pane', 'replacement-root'], + ['stop', 'same-root'], + ['stop', 'replacement-root'] + ])('preserves newer authority after %s with %s', async (boundary, rootCase) => { + const store = createAgentStatusStore({ epoch: 'tmux-retirement', mode: 'authority' }) + let finish = (_root: typeof TMUX_TEST_ROOT) => {} + const pending = new Promise((resolve) => { + finish = resolve + }) + const getRoot = vi + .fn(async () => TMUX_TEST_ROOT) + .mockResolvedValueOnce(TMUX_TEST_ROOT) + .mockImplementationOnce(() => pending) + const unavailable = vi.fn((_pane: string, subject?: AgentStatusPtySubject) => { + if (subject) { + store.applyMutation({ parent: { subject } }) + } + }) + const owner = new TmuxAgentHookOwner({ + store: () => store, + getRoot, + probe: async () => ({ clients: [], rows: [] }), + isRetired: () => false, + publish: (event, time, subject, start) => { + commitTmuxSelectedStatus(store, subject, event, time, start) + }, + unavailable + }) + try { + const refresh = owner.ingest('opencode', tmuxTestBody(), 'production') + await vi.waitFor(() => expect(getRoot).toHaveBeenCalledTimes(2)) + if (boundary === 'stop') { + owner.stop() + } else { + owner.clearPane(TMUX_TEST_PANE) + } + const current = + rootCase === 'same-root' + ? TMUX_TEST_ROOT + : { ...TMUX_TEST_ROOT, pid: 200, incarnation: 'replacement' } + const subject = { ...current.scope, kind: 'pty' as const, paneKey: TMUX_TEST_PANE } + commitTmuxSelectedStatus( + store, + subject, + { + paneKey: TMUX_TEST_PANE, + tabId: 'tab-tmux', + worktreeId: 'workspace', + source: 'opencode2', + connectionId: null, + launchToken: 'new-generation', + payload: { + agentType: 'opencode2', + state: 'waiting', + toolName: 'shell', + prompt: 'new turn' + } + }, + 100 + ) + const before = store.getParent(subject) + expect(before?.status?.state).toBe('waiting') + finish(current) + await refresh + expect(store.getParent(subject)).toEqual(before) + expect(unavailable).not.toHaveBeenCalled() + } finally { + finish(TMUX_TEST_ROOT) + owner.stop() + } + }) +}) diff --git a/src/shared/tmux-host-attachment-probe.test.ts b/src/shared/tmux-host-attachment-probe.test.ts new file mode 100644 index 000000000000..f78c214a66ca --- /dev/null +++ b/src/shared/tmux-host-attachment-probe.test.ts @@ -0,0 +1,50 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { probeTmuxHostAttachments } from './tmux-host-attachment-probe' +const run = vi.hoisted(() => vi.fn()) +vi.mock('./child-process/run-process', () => ({ runProcess: run })) +vi.mock('node:fs/promises', () => ({ + stat: async () => ({ isSocket: () => true, uid: process.getuid?.() }) +})) +vi.mock('./agent-process-presence-probe', () => ({ + readAgentProcess: async (pid: number) => ({ verdict: 'live', startTime: `birth-${pid}` }) +})) +let active = 0 +let maximum = 0 +beforeEach(() => { + active = 0 + maximum = 0 + run.mockReset() + run.mockImplementation(async (options: { program: string; args: string[] }) => { + if (options.program === 'tmux') { + return { + code: 0, + timedOut: false, + stdout: Array.from({ length: 16 }, (_, i) => `${101 + i}:%${i}`).join('\n') + } + } + const pid = Number(options.args[1]) + active++ + maximum = Math.max(maximum, active) + await Promise.resolve() + active-- + const start = process.platform === 'darwin' ? 'Fri Oct 2 03:00:00 2026' : '100' + return { + code: 0, + timedOut: false, + stdout: `${pid} ${pid === 100 ? 1 : 100} ${pid} ${pid === 100 ? 101 : pid} S pts/1 ${start} ${pid === 100 ? '/bin/bash' : '/usr/bin/tmux attach'}\n` + } + }) +}) +describe('bounded targeted tmux process capture', () => { + it('captures sixteen clients plus the root in bounded batches without a whole-host scan', async () => { + const proof = await probeTmuxHostAttachments('/tmp/fixture.sock', [100]) + expect(proof?.clients).toHaveLength(16) + expect(proof?.rows).toHaveLength(17) + const calls = run.mock.calls.filter(([options]) => options.program === '/bin/ps') + expect(calls).toHaveLength(17) + expect(maximum).toBeLessThanOrEqual(16) + expect( + calls.every(([options]) => options.args[0] === '-p' && /^[0-9]+$/.test(options.args[1])) + ).toBe(true) + }) +}) diff --git a/src/shared/tmux-host-attachment-probe.ts b/src/shared/tmux-host-attachment-probe.ts new file mode 100644 index 000000000000..3adcfe303cba --- /dev/null +++ b/src/shared/tmux-host-attachment-probe.ts @@ -0,0 +1,102 @@ +import { stat } from 'node:fs/promises' +import { readAgentProcess } from './agent-process-presence-probe' +import { runProcess } from './child-process/run-process' +import { + PS_ARGS, + parseStrictProcessTableRows, + type ProcessTableRow +} from './process-table-snapshot' +import { parseTmuxAttachedClients, type TmuxAttachedClient } from './tmux-client-attachment' + +const MAX_PROCESS_ROWS = 256 + +async function readSelectedRows(pids: readonly number[]): Promise { + if (pids.length > 16) { + const rows: ProcessTableRow[] = [] + for (let index = 0; index < pids.length; index += 16) { + rows.push(...(await readSelectedRows(pids.slice(index, index + 16)))) + } + return rows + } + if (pids.length > 1) { + const rows = await Promise.all(pids.map((pid) => readSelectedRows([pid]))) + return rows.flat() + } + const result = await runProcess({ + program: '/bin/ps', + args: ['-p', pids.join(','), '-o', PS_ARGS[1]], + env: { ...process.env, LC_ALL: 'C', LANG: 'C', TZ: 'UTC0' }, + timeoutMs: 1000, + maxOutputBytes: 65536 + }) + if (result.timedOut || result.code !== 0 || result.stdout.length >= 65536) { + throw new Error('tmux_process_capture_unverifiable') + } + const rows = parseStrictProcessTableRows(result.stdout) + if (process.platform === 'linux') { + return Promise.all( + rows.map(async (row) => { + const observed = await readAgentProcess(row.pid) + return { ...row, startTime: observed.verdict === 'live' ? observed.startTime : undefined } + }) + ) + } + return rows +} + +/** A bounded capture of tmux clients and their parent paths; never a whole-host scan. */ +export async function probeTmuxHostAttachments( + socket: string, + rootPids: readonly number[] +): Promise<{ clients: TmuxAttachedClient[]; rows: ProcessTableRow[] } | null> { + if (process.platform === 'win32' || rootPids.length === 0 || rootPids.length > 64) { + return null + } + try { + const socketStat = await stat(socket) + if (!socketStat.isSocket() || (process.getuid && socketStat.uid !== process.getuid())) { + return null + } + const result = await runProcess({ + program: 'tmux', + args: ['-S', socket, 'list-clients', '-F', '#{client_pid}:#{pane_id}'], + timeoutMs: 1000, + maxOutputBytes: 65536 + }) + const clients = + !result.timedOut && result.code === 0 ? parseTmuxAttachedClients(result.stdout) : null + if (!clients) { + return null + } + const byPid = new Map() + let pids = [...new Set([...rootPids, ...clients.map((client) => client.pid)])] + for (let depth = 0; depth < 8 && pids.length > 0; depth++) { + if (byPid.size + pids.length > MAX_PROCESS_ROWS) { + return null + } + for (const row of await readSelectedRows(pids)) { + byPid.set(row.pid, row) + } + const missingParents = new Set() + for (const client of clients) { + const visited = new Set() + let row = byPid.get(client.pid) + while (row && !rootPids.includes(row.pid) && !visited.has(row.pid)) { + visited.add(row.pid) + if (row.ppid <= 1) { + break + } + if (!byPid.has(row.ppid)) { + missingParents.add(row.ppid) + break + } + row = byPid.get(row.ppid) + } + } + pids = [...missingParents] + } + return { clients, rows: [...byPid.values()] } + } catch { + return null + } +} diff --git a/src/shared/tmux-selected-status.ts b/src/shared/tmux-selected-status.ts new file mode 100644 index 000000000000..76b9dc11efa2 --- /dev/null +++ b/src/shared/tmux-selected-status.ts @@ -0,0 +1,150 @@ +import { createHash } from 'node:crypto' +import type { TmuxHookPane } from './tmux-client-attachment' +import { normalizeAgentStatusPayload, type AgentStatusIpcPayload } from './agent-status-types' +import type { AgentHookEventPayload } from './agent-hook-listener/listener-event' +import type { AgentStatusStore } from './agent-status-store' +import type { AgentHookUnavailableEnvelope } from './agent-hook-relay' +import { + serializeAgentStatusSubject, + type AgentStatusExecutionScope, + type AgentStatusPtySubject +} from './agent-status-subject' + +/** Commit the attachment projection in its owner's store, retaining the inner observation clock. */ +export function commitTmuxSelectedStatus( + store: AgentStatusStore, + subject: AgentStatusPtySubject, + event: AgentHookEventPayload, + observedAt: number, + stateStartedAt = observedAt, + now = Date.now() +) { + const previous = store.getParent(subject) + const status = { + ...event.payload, + paneKey: subject.paneKey, + tabId: event.tabId, + worktreeId: subject.workspaceId, + connectionId: event.connectionId, + launchToken: event.launchToken, + providerSession: event.providerSession, + promptInteractionKey: event.promptInteractionKey, + receivedAt: Math.max(now, (previous?.status?.receivedAt ?? -1) + 1), + evidenceObservedAt: observedAt, + stateStartedAt + } + const mutation = store.applyMutation({ + removeFacts: ['tmux.unavailable', 'tmux.source', 'tmux.tabId', 'tmux.launchToken'].map( + (key) => ({ subject, key }) + ), + parent: { subject, status, firstObservedAt: previous?.firstObservedAt ?? observedAt } + }) + return mutation ? store.getParent(subject)?.status : undefined +} + +/** Legacy wire projection is derived from the committed record, never a writable row cache. */ +export function tmuxCanonicalStatusEvent(status: AgentStatusIpcPayload): AgentHookEventPayload { + const payload = normalizeAgentStatusPayload(status) + if (!payload) { + throw new Error('Invalid committed tmux status') + } + return { + paneKey: status.paneKey, + tabId: status.tabId, + worktreeId: status.worktreeId, + connectionId: status.connectionId, + source: status.agentType === 'opencode2' ? 'opencode2' : 'opencode', + launchToken: status.launchToken, + providerSession: status.providerSession, + promptInteractionKey: status.promptInteractionKey, + hasExplicitPrompt: status.prompt.length > 0, + hostEvidenceObservedAt: status.evidenceObservedAt ?? status.receivedAt, + hookEventName: + status.state === 'done' + ? 'SessionIdle' + : status.state === 'waiting' + ? status.toolName + ? 'PermissionRequest' + : 'AskUserQuestion' + : 'SessionBusy', + payload + } +} + +export function commitTmuxUnavailable( + store: AgentStatusStore, + subject: AgentStatusPtySubject, + identity?: Pick +) { + const previous = store.getParent(subject) + const status = previous?.status + const correlation = status + ? { + source: status.agentType === 'opencode2' ? 'opencode2' : 'opencode', + tabId: status.tabId, + launchToken: status.launchToken + } + : identity + const facts = [{ subject, key: 'tmux.unavailable', value: true }] + store.applyMutation({ + parent: { subject, firstObservedAt: previous?.firstObservedAt ?? Date.now() }, + facts: [ + ...facts, + ...(correlation + ? [ + { + subject, + key: 'tmux.source', + value: correlation.source ?? 'opencode' + }, + { subject, key: 'tmux.tabId', value: correlation.tabId ?? null }, + { subject, key: 'tmux.launchToken', value: correlation.launchToken ?? null } + ] + : []) + ] + }) + return readTmuxUnavailable(store, subject) +} + +export function readTmuxUnavailable( + store: AgentStatusStore, + subject: AgentStatusPtySubject +): AgentHookUnavailableEnvelope | undefined { + const key = serializeAgentStatusSubject(subject) + const facts = new Map( + store + .getSnapshot() + .facts.filter((fact) => serializeAgentStatusSubject(fact.subject) === key) + .map((fact) => [fact.key, fact.value]) + ) + if (facts.get('tmux.unavailable') !== true) { + return undefined + } + const tabId = facts.get('tmux.tabId') + const launchToken = facts.get('tmux.launchToken') + return { + source: facts.get('tmux.source') === 'opencode2' ? 'opencode2' : 'opencode', + paneKey: subject.paneKey, + worktreeId: subject.workspaceId, + ...(typeof tabId === 'string' ? { tabId } : {}), + ...(typeof launchToken === 'string' ? { launchToken } : {}), + connectionId: null, + statusUnavailable: true, + payload: null + } +} + +export function isTmuxInnerSubject(subject: { kind: string; paneKey?: string }): boolean { + return subject.kind === 'pty' && subject.paneKey?.startsWith('tmux-inner:') === true +} + +export function tmuxInnerSubject( + scope: AgentStatusExecutionScope, + outerPaneKey: string, + tmux: TmuxHookPane +): AgentStatusPtySubject { + const digest = createHash('sha256') + .update(`${outerPaneKey}\0${tmux.socket}\0${tmux.pane}`) + .digest('hex') + return { ...scope, kind: 'pty', paneKey: `tmux-inner:${digest}` } +} diff --git a/src/shared/tmux-status.test-fixture.ts b/src/shared/tmux-status.test-fixture.ts new file mode 100644 index 000000000000..517b38a5f41e --- /dev/null +++ b/src/shared/tmux-status.test-fixture.ts @@ -0,0 +1,46 @@ +import type { ProcessTableRow } from './process-table-snapshot' +import type { TmuxManagedPty } from './tmux-agent-hook-owner' + +export const TMUX_TEST_PANE = 'tab-tmux:22222222-2222-4222-8222-222222222222' +export const TMUX_TEST_ROOT: TmuxManagedPty = { + pid: 100, + incarnation: 'first', + scope: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace', + workspaceKind: 'folder' + } +} +export const TMUX_TEST_ROWS: ProcessTableRow[] = [ + { + pid: 100, + ppid: 1, + pgid: 100, + tpgid: 101, + tty: 'pts/1', + stat: 'S', + startTime: 'root', + command: '/bin/bash' + }, + { + pid: 101, + ppid: 100, + pgid: 101, + tpgid: 101, + tty: 'pts/1', + stat: 'S+', + startTime: 'client', + command: '/usr/bin/tmux attach' + } +] +export function tmuxTestBody(pane = '%0') { + return { + paneKey: TMUX_TEST_PANE, + tabId: 'tab-tmux', + worktreeId: 'workspace', + launchToken: 'generation', + tmux: { socket: '/tmp/test.sock', pane }, + payload: { hook_event_name: 'SessionIdle', prompt: 'completed', session_id: 'session' } + } +}