diff --git a/src/relay/agent-status-store-relay-context.test.ts b/src/relay/agent-status-store-relay-context.test.ts new file mode 100644 index 00000000000..92319a2e644 --- /dev/null +++ b/src/relay/agent-status-store-relay-context.test.ts @@ -0,0 +1,79 @@ +import { readFileSync } from 'node:fs' +import { describe, expect, it } from 'vitest' +import { createAgentChildWorkAdmission } from '../shared/agent-status-child-work-admission' +import { createAgentStatusStore } from '../shared/agent-status-store' +import { makeStructuredAgentStatusSubject } from '../shared/agent-status-subject' + +const SHARED_CORE_FILES = [ + 'agent-status-child-work.ts', + 'agent-status-child-work-codec.ts', + 'agent-status-child-work-admission.ts', + 'agent-status-child-work-admission-core.ts', + 'agent-status-child-work-admission-operations.ts', + 'agent-status-child-work-alias.ts', + 'agent-status-child-work-freshness.ts', + 'agent-status-child-work-projection.ts', + 'agent-status-store.ts', + 'agent-status-store-codec.ts', + 'agent-status-store-mutation.ts', + 'agent-status-store-contract.ts', + 'agent-status-store-fact-codec.ts', + 'agent-status-store-parent.ts', + 'agent-status-store-persistence.ts', + 'agent-status-store-state.ts', + 'agent-status-store-status-codec.ts', + 'agent-status-transport-envelope.ts' +] + +const trustedSubject = makeStructuredAgentStatusSubject( + { + executionHostId: 'ssh:relay-host-a', + wslDistro: null, + workspaceId: 'folder-workspace-a', + workspaceKind: 'folder' + }, + 'session_11111111-1111-4111-8111-111111111111' +) + +describe('agent status store relay context', () => { + it('instantiates the same shared core and completes an admission/snapshot round-trip', () => { + const authority = createAgentStatusStore({ epoch: 'relay-epoch-a', mode: 'authority' }) + expect( + authority.applyMutation({ parent: { subject: trustedSubject, firstObservedAt: 10 } }) + ).not.toBeNull() + const admission = createAgentChildWorkAdmission(authority, { + mintChildWorkId: () => 'relay-child-1' + }) + + expect( + admission.announce({ + parent: trustedSubject, + provider: 'claude', + aliases: [{ segmentId: 'segment-1', aliasKind: 'task_id', alias: 'task-1' }], + fence: { invocationId: 'invocation-1', generation: 1 }, + lifetime: 'current', + kind: 'agent', + state: 'working', + membership: 'live', + observedAt: 20, + stoppable: true, + provenance: { source: 'transport', producerId: 'relay-fixture' } + }) + ).toMatchObject({ accepted: true, childWorkId: 'relay-child-1' }) + + const replica = createAgentStatusStore({ epoch: 'replica-placeholder', mode: 'replica' }) + expect(replica.applySnapshot(authority.getSnapshot())).toBe(true) + expect(replica.getParent(trustedSubject)?.firstObservedAt).toBe(10) + expect(replica.getChildren(trustedSubject)[0]?.childWorkId).toBe('relay-child-1') + }) + + it('keeps the relay-consumed core free of main, renderer and Electron imports', () => { + for (const filename of SHARED_CORE_FILES) { + const source = readFileSync(new URL(`../shared/${filename}`, import.meta.url), 'utf8') + expect(source, filename).not.toMatch( + /from\s+['"](?:electron|\.\.\/(?:main|renderer))(?:\/|['"])/ + ) + expect(source, filename).not.toMatch(/require\(['"]electron['"]\)/) + } + }) +}) diff --git a/src/shared/agent-status-child-work-admission-collision.test.ts b/src/shared/agent-status-child-work-admission-collision.test.ts new file mode 100644 index 00000000000..c5ac2eec935 --- /dev/null +++ b/src/shared/agent-status-child-work-admission-collision.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it } from 'vitest' +import { createAgentChildWorkAdmission } from './agent-status-child-work-admission' +import { createAgentStatusStore } from './agent-status-store' +import { + makeStructuredAgentStatusSubject, + type AgentStatusExecutionScope, + type AgentStatusSubject +} from './agent-status-subject' + +const SESSION_ID = 'session_11111111-1111-4111-8111-111111111111' + +function subject(overrides: Partial = {}): AgentStatusSubject { + return makeStructuredAgentStatusSubject( + { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree', + ...overrides + }, + SESSION_ID + ) +} + +function request(parent: AgentStatusSubject, kind: 'agent' | 'unknown') { + return { + parent, + provider: 'claude', + aliases: [{ segmentId: 'segment-1', aliasKind: 'task_id' as const, alias: 'task-1' }], + fence: { invocationId: `invocation-${kind}`, generation: kind === 'unknown' ? 1 : 2 }, + lifetime: 'current' as const, + kind, + state: 'working' as const, + membership: 'live' as const, + observedAt: 10, + stoppable: true, + provenance: { source: 'structured-session' as const, producerId: 'journal-1' } + } +} + +describe('agent child-work admission collisions', () => { + it('rejects an adopt whose alias reclassification would capture another child', () => { + const parent = subject() + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + store.applyMutation({ parent: { subject: parent } }) + const ids = ['child-1', 'child-2'] + const admission = createAgentChildWorkAdmission(store, { + mintChildWorkId: () => ids.shift() ?? 'unexpected-child' + }) + admission.announce(request(parent, 'unknown')) + admission.announce(request(parent, 'agent')) + const before = store.getSnapshot() + + expect( + admission.adopt({ + ...request(parent, 'unknown'), + childWorkId: 'child-1', + expectedFence: { invocationId: 'invocation-unknown', generation: 1 }, + kind: 'agent', + observedAt: 20 + }) + ).toEqual({ accepted: false, reason: 'ambiguous' }) + expect(store.getSnapshot()).toEqual(before) + }) + + it('rejects reparenting into an occupied alias scope without moving the child', () => { + const firstParent = subject() + const nextParent = subject({ workspaceId: 'workspace-2' }) + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + store.applyMutation({ parent: { subject: firstParent } }) + store.applyMutation({ parent: { subject: nextParent } }) + const ids = ['child-1', 'child-2'] + const admission = createAgentChildWorkAdmission(store, { + mintChildWorkId: () => ids.shift() ?? 'unexpected-child' + }) + admission.announce(request(firstParent, 'agent')) + admission.announce(request(nextParent, 'agent')) + const before = store.getSnapshot() + + expect( + admission.reparent({ + childWorkId: 'child-1', + fromParent: firstParent, + toParent: nextParent, + expectedFence: { invocationId: 'invocation-agent', generation: 2 }, + observedAt: 20 + }) + ).toEqual({ accepted: false, reason: 'ambiguous' }) + expect(store.getSnapshot()).toEqual(before) + }) +}) diff --git a/src/shared/agent-status-child-work-admission-core.ts b/src/shared/agent-status-child-work-admission-core.ts new file mode 100644 index 00000000000..ce52f9c0692 --- /dev/null +++ b/src/shared/agent-status-child-work-admission-core.ts @@ -0,0 +1,171 @@ +import { + serializeAgentChildWorkAliasKey, + type AgentChildWorkAliasInput, + type AgentChildWorkAliasRecord +} from './agent-status-child-work-alias' +import { + agentChildWorkFencesEqual, + type AgentChildWorkId, + type AgentChildWorkInput, + type AgentChildWorkInvocationFence, + type AgentChildWorkKind, + type AgentChildWorkRecord +} from './agent-status-child-work' +import { parseAgentChildWorkInput } from './agent-status-child-work-codec' +import type { + AgentChildWorkAdmissionResult, + AgentChildWorkAdoptRequest, + AgentChildWorkAnnounceRequest, + AgentChildWorkObservationAlias, + AgentChildWorkObservationFields +} from './agent-status-child-work-admission' +import type { AgentStatusStore } from './agent-status-store' +import { agentStatusSubjectsEqual, type AgentStatusSubject } from './agent-status-subject' + +const MAX_ALIASES_PER_ADMISSION = 32 + +export function rejectAgentChildWorkAdmission( + reason: Extract['reason'] +) { + return { accepted: false, reason } as const +} + +export function findAgentChildWork( + store: AgentStatusStore, + childWorkId: string +): AgentChildWorkRecord | null { + return store.getSnapshot().children.find((child) => child.childWorkId === childWorkId) ?? null +} + +export function agentChildWorkAliasesForChild( + store: AgentStatusStore, + childWorkId: string +): AgentChildWorkAliasRecord[] { + return store.getSnapshot().aliases.filter((alias) => alias.childWorkId === childWorkId) +} + +export function buildAgentChildWorkAliases( + parent: AgentStatusSubject, + provider: string, + kind: AgentChildWorkKind, + aliases: AgentChildWorkObservationAlias[], + childWorkId: AgentChildWorkId, + fence: AgentChildWorkInvocationFence +): AgentChildWorkAliasInput[] | null { + if (aliases.length === 0 || aliases.length > MAX_ALIASES_PER_ADMISSION) { + return null + } + const built: AgentChildWorkAliasInput[] = [] + const keys = new Set() + try { + for (const alias of aliases) { + const candidate = { parent, provider, kind, ...alias, childWorkId, fence } + const key = serializeAgentChildWorkAliasKey(candidate) + if (keys.has(key)) { + return null + } + keys.add(key) + built.push(candidate) + } + } catch { + return null + } + return built +} + +export function buildAgentChildWork( + request: AgentChildWorkObservationFields & { + parent: AgentStatusSubject + provider: string + }, + childWorkId: string, + firstObservedAt: number, + invocation: AgentChildWorkInvocationFence, + previousInvocations?: AgentChildWorkInput['previousInvocations'] +): AgentChildWorkInput | null { + return parseAgentChildWorkInput({ + childWorkId, + parent: request.parent, + provider: request.provider, + kind: request.kind, + state: request.state, + membership: request.membership, + ...(request.outcome !== undefined ? { outcome: request.outcome } : {}), + ...(request.name !== undefined ? { name: request.name } : {}), + ...(request.description !== undefined ? { description: request.description } : {}), + ...(request.agentType !== undefined ? { agentType: request.agentType } : {}), + ...(request.model !== undefined ? { model: request.model } : {}), + ...(request.totalTokens !== undefined ? { totalTokens: request.totalTokens } : {}), + ...(request.providerTiming !== undefined ? { providerTiming: request.providerTiming } : {}), + firstObservedAt, + observedAt: request.observedAt, + stoppable: request.stoppable, + invocation, + ...(previousInvocations !== undefined ? { previousInvocations } : {}), + provenance: request.provenance + }) +} + +export function commitAgentChildWork( + store: AgentStatusStore, + child: AgentChildWorkInput, + aliases: AgentChildWorkAliasInput[], + created: boolean, + removeAliases: string[] = [] +): AgentChildWorkAdmissionResult { + const envelope = store.applyMutation({ + children: [child], + aliases, + ...(removeAliases.length > 0 ? { removeAliases } : {}) + }) + return envelope + ? { accepted: true, childWorkId: child.childWorkId, revision: envelope.revision, created } + : rejectAgentChildWorkAdmission('store-rejected') +} + +export function updateExistingAgentChildWork( + store: AgentStatusStore, + request: AgentChildWorkAnnounceRequest | AgentChildWorkAdoptRequest, + child: AgentChildWorkRecord, + aliases: AgentChildWorkAliasInput[], + removeAliases: string[] = [] +): AgentChildWorkAdmissionResult { + const updated = buildAgentChildWork( + request, + child.childWorkId, + child.firstObservedAt, + child.invocation, + child.previousInvocations + ) + return updated + ? commitAgentChildWork(store, updated, aliases, false, removeAliases) + : rejectAgentChildWorkAdmission('invalid') +} + +export function resolveAgentChildWorkAliasRecords( + store: AgentStatusStore, + aliases: AgentChildWorkAliasInput[] +): AgentChildWorkAliasRecord[] { + const keys = new Set(aliases.map(serializeAgentChildWorkAliasKey)) + return store + .getSnapshot() + .aliases.filter((alias) => keys.has(serializeAgentChildWorkAliasKey(alias))) +} + +export function validateExistingAgentChildWork( + child: AgentChildWorkRecord | null, + parent: AgentStatusSubject, + provider: string, + expectedFence: AgentChildWorkInvocationFence +): AgentChildWorkAdmissionResult | null { + if (!child) { + return rejectAgentChildWorkAdmission('unknown-child') + } + if (!agentStatusSubjectsEqual(child.parent, parent) || child.provider !== provider) { + return rejectAgentChildWorkAdmission('ambiguous') + } + if (!agentChildWorkFencesEqual(child.invocation, expectedFence)) { + return rejectAgentChildWorkAdmission('stale-invocation') + } + return null +} diff --git a/src/shared/agent-status-child-work-admission-operations.ts b/src/shared/agent-status-child-work-admission-operations.ts new file mode 100644 index 00000000000..ded779cd342 --- /dev/null +++ b/src/shared/agent-status-child-work-admission-operations.ts @@ -0,0 +1,293 @@ +import { serializeAgentChildWorkAliasKey } from './agent-status-child-work-alias' +import { + AGENT_CHILD_WORK_INVOCATION_HISTORY_MAX, + agentChildWorkFencesEqual, + type AgentChildWorkId, + type AgentChildWorkRecord +} from './agent-status-child-work' +import { + agentChildWorkAliasesForChild, + buildAgentChildWork, + buildAgentChildWorkAliases, + commitAgentChildWork, + findAgentChildWork, + rejectAgentChildWorkAdmission, + resolveAgentChildWorkAliasRecords, + updateExistingAgentChildWork, + validateExistingAgentChildWork +} from './agent-status-child-work-admission-core' +import type { + AgentChildWorkAdmissionResult, + AgentChildWorkAdoptRequest, + AgentChildWorkAnnounceRequest, + AgentChildWorkReparentRequest, + AgentChildWorkResumeRequest, + AgentChildWorkStopRequest +} from './agent-status-child-work-admission' +import { + parseAgentChildWorkInput, + parseAgentChildWorkInvocationFence +} from './agent-status-child-work-codec' +import type { AgentStatusStore } from './agent-status-store' +import { agentStatusSubjectsEqual, parseAgentStatusSubject } from './agent-status-subject' + +export function announceAgentChildWork( + store: AgentStatusStore, + mintChildWorkId: () => AgentChildWorkId, + request: AgentChildWorkAnnounceRequest +): AgentChildWorkAdmissionResult { + const parent = parseAgentStatusSubject(request.parent) + const fence = parseAgentChildWorkInvocationFence(request.fence) + if (!parent || !fence || !store.getParent(parent)) { + return rejectAgentChildWorkAdmission('invalid') + } + const lookupAliases = buildAgentChildWorkAliases( + parent, + request.provider, + request.kind, + request.aliases, + 'unresolved-child', + fence + ) + if (!lookupAliases) { + return rejectAgentChildWorkAdmission('invalid') + } + const bindings = resolveAgentChildWorkAliasRecords(store, lookupAliases) + const exact = bindings.filter((binding) => agentChildWorkFencesEqual(binding.fence, fence)) + const exactIds = new Set(exact.map((binding) => binding.childWorkId)) + if (request.lifetime === 'proven-new') { + if (exact.length > 0) { + return rejectAgentChildWorkAdmission('id-collision') + } + const candidateId = mintChildWorkId() + const aliases = buildAgentChildWorkAliases( + parent, + request.provider, + request.kind, + request.aliases, + candidateId, + fence + ) + if (!aliases || findAgentChildWork(store, candidateId)) { + return rejectAgentChildWorkAdmission(aliases ? 'id-collision' : 'invalid') + } + const child = buildAgentChildWork(request, candidateId, request.observedAt, fence) + return child + ? commitAgentChildWork(store, child, aliases, true) + : rejectAgentChildWorkAdmission('invalid') + } + if (bindings.length !== exact.length || exactIds.size > 1) { + return rejectAgentChildWorkAdmission(exactIds.size > 1 ? 'ambiguous' : 'stale-invocation') + } + const existingId = exact[0]?.childWorkId + if (existingId) { + const child = findAgentChildWork(store, existingId) + if (!child) { + return rejectAgentChildWorkAdmission('ambiguous') + } + const aliases = buildAgentChildWorkAliases( + parent, + request.provider, + request.kind, + request.aliases, + child.childWorkId, + fence + ) + return aliases + ? updateExistingAgentChildWork(store, request, child, aliases) + : rejectAgentChildWorkAdmission('invalid') + } + const candidateId = mintChildWorkId() + if (findAgentChildWork(store, candidateId)) { + return rejectAgentChildWorkAdmission('id-collision') + } + const aliases = buildAgentChildWorkAliases( + parent, + request.provider, + request.kind, + request.aliases, + candidateId, + fence + ) + const child = buildAgentChildWork(request, candidateId, request.observedAt, fence) + return aliases && child + ? commitAgentChildWork(store, child, aliases, true) + : rejectAgentChildWorkAdmission('invalid') +} + +export function adoptAgentChildWork( + store: AgentStatusStore, + request: AgentChildWorkAdoptRequest +): AgentChildWorkAdmissionResult { + const child = findAgentChildWork(store, request.childWorkId) + const invalid = validateExistingAgentChildWork( + child, + request.parent, + request.provider, + request.expectedFence + ) + if (invalid || !child) { + return invalid ?? rejectAgentChildWorkAdmission('unknown-child') + } + const aliases = buildAgentChildWorkAliases( + request.parent, + request.provider, + request.kind, + request.aliases, + child.childWorkId, + child.invocation + ) + if (!aliases) { + return rejectAgentChildWorkAdmission('invalid') + } + const collisions = resolveAgentChildWorkAliasRecords(store, aliases).filter( + (binding) => binding.childWorkId !== child.childWorkId + ) + if (collisions.length > 0) { + return rejectAgentChildWorkAdmission('ambiguous') + } + const oldAliases = agentChildWorkAliasesForChild(store, child.childWorkId) + const removeAliases = oldAliases + .filter((alias) => alias.kind !== request.kind) + .map(serializeAgentChildWorkAliasKey) + const reclassified = oldAliases.map((alias) => ({ + parent: alias.parent, + provider: alias.provider, + segmentId: alias.segmentId, + kind: request.kind, + aliasKind: alias.aliasKind, + alias: alias.alias, + childWorkId: alias.childWorkId, + fence: alias.fence + })) + const unique = new Map( + [...reclassified, ...aliases].map((alias) => [serializeAgentChildWorkAliasKey(alias), alias]) + ) + const reclassifiedCollisions = resolveAgentChildWorkAliasRecords(store, [ + ...unique.values() + ]).filter((binding) => binding.childWorkId !== child.childWorkId) + if (reclassifiedCollisions.length > 0) { + return rejectAgentChildWorkAdmission('ambiguous') + } + return updateExistingAgentChildWork(store, request, child, [...unique.values()], removeAliases) +} + +export function resumeAgentChildWork( + store: AgentStatusStore, + request: AgentChildWorkResumeRequest +): AgentChildWorkAdmissionResult { + const child = findAgentChildWork(store, request.childWorkId) + const invalid = validateExistingAgentChildWork( + child, + request.parent, + request.provider, + request.expectedFence + ) + const nextFence = parseAgentChildWorkInvocationFence(request.nextFence) + if (invalid || !child) { + return invalid ?? rejectAgentChildWorkAdmission('unknown-child') + } + if (!nextFence || agentChildWorkFencesEqual(child.invocation, nextFence)) { + return rejectAgentChildWorkAdmission('invalid') + } + const aliases = buildAgentChildWorkAliases( + request.parent, + request.provider, + request.kind, + request.aliases, + child.childWorkId, + nextFence + ) + if (!aliases) { + return rejectAgentChildWorkAdmission('invalid') + } + const collisions = resolveAgentChildWorkAliasRecords(store, aliases).filter( + (binding) => binding.childWorkId !== child.childWorkId + ) + if (collisions.length > 0) { + return rejectAgentChildWorkAdmission('ambiguous') + } + const previousInvocations = [ + ...(child.previousInvocations ?? []), + { + fence: child.invocation, + ...(child.outcome !== undefined ? { outcome: child.outcome } : {}), + ...(child.membership === 'settled' ? { settledAt: child.observedAt } : {}) + } + ].slice(-AGENT_CHILD_WORK_INVOCATION_HISTORY_MAX) + const resumed = buildAgentChildWork( + request, + child.childWorkId, + child.firstObservedAt, + nextFence, + previousInvocations + ) + return resumed + ? commitAgentChildWork(store, resumed, aliases, false) + : rejectAgentChildWorkAdmission('invalid') +} + +export function reparentAgentChildWork( + store: AgentStatusStore, + request: AgentChildWorkReparentRequest +): AgentChildWorkAdmissionResult { + const child = findAgentChildWork(store, request.childWorkId) + if ( + !child || + !agentStatusSubjectsEqual(child.parent, request.fromParent) || + !agentChildWorkFencesEqual(child.invocation, request.expectedFence) || + !store.getParent(request.toParent) || + request.observedAt < child.observedAt + ) { + return rejectAgentChildWorkAdmission(child ? 'stale-invocation' : 'unknown-child') + } + const oldAliases = agentChildWorkAliasesForChild(store, child.childWorkId) + const aliases = oldAliases.map((alias) => ({ + parent: request.toParent, + provider: alias.provider, + segmentId: alias.segmentId, + kind: alias.kind, + aliasKind: alias.aliasKind, + alias: alias.alias, + childWorkId: alias.childWorkId, + fence: alias.fence + })) + const collisions = resolveAgentChildWorkAliasRecords(store, aliases).filter( + (binding) => binding.childWorkId !== child.childWorkId + ) + if (collisions.length > 0) { + return rejectAgentChildWorkAdmission('ambiguous') + } + const { revision: _revision, ...childInput } = child + const moved = parseAgentChildWorkInput({ + ...childInput, + parent: request.toParent, + observedAt: request.observedAt + }) + return moved + ? commitAgentChildWork( + store, + moved, + aliases, + false, + oldAliases.map(serializeAgentChildWorkAliasKey) + ) + : rejectAgentChildWorkAdmission('invalid') +} + +export function authorizeAgentChildWorkStop( + store: AgentStatusStore, + request: AgentChildWorkStopRequest +): AgentChildWorkRecord | null { + const child = findAgentChildWork(store, request.childWorkId) + if ( + !child || + !agentStatusSubjectsEqual(child.parent, request.parent) || + !agentChildWorkFencesEqual(child.invocation, request.expectedFence) || + child.membership !== 'live' || + child.stoppable !== true + ) { + return null + } + return child +} diff --git a/src/shared/agent-status-child-work-admission.test.ts b/src/shared/agent-status-child-work-admission.test.ts new file mode 100644 index 00000000000..5740b36409d --- /dev/null +++ b/src/shared/agent-status-child-work-admission.test.ts @@ -0,0 +1,256 @@ +import { describe, expect, it, vi } from 'vitest' +import { + createAgentChildWorkAdmission, + type AgentChildWorkAnnounceRequest +} from './agent-status-child-work-admission' +import { serializeAgentChildWorkAliasKey } from './agent-status-child-work-alias' +import { createAgentStatusStore } from './agent-status-store' +import { + makeStructuredAgentStatusSubject, + type AgentStatusExecutionScope, + type AgentStatusSubject +} from './agent-status-subject' + +const SESSION_ID = 'session_11111111-1111-4111-8111-111111111111' + +function subject(overrides: Partial = {}): AgentStatusSubject { + return makeStructuredAgentStatusSubject( + { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree', + ...overrides + }, + SESSION_ID + ) +} + +function announce( + parent: AgentStatusSubject, + overrides: Partial = {} +): AgentChildWorkAnnounceRequest { + return { + parent, + provider: 'claude', + aliases: [{ segmentId: 'segment-1', aliasKind: 'task_id', alias: 'task-1' }], + fence: { invocationId: 'invocation-1', generation: 1 }, + lifetime: 'current', + kind: 'agent', + state: 'working', + membership: 'live', + observedAt: 10, + stoppable: true, + provenance: { source: 'structured-session', producerId: 'journal-1' }, + ...overrides + } +} + +function setup(parents: AgentStatusSubject[] = [subject()]) { + const ids = ['child-1', 'child-2', 'child-3', 'child-4'] + const mintChildWorkId = vi.fn(() => ids.shift() ?? 'child-overflow') + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + for (const parent of parents) { + expect(store.applyMutation({ parent: { subject: parent } })).not.toBeNull() + } + return { + store, + mintChildWorkId, + admission: createAgentChildWorkAdmission(store, { mintChildWorkId }) + } +} + +describe('agent child-work admission', () => { + it('keeps one host id across re-announcement with another tool-use alias', () => { + const parent = subject() + const { admission, mintChildWorkId, store } = setup([parent]) + + expect(admission.announce(announce(parent))).toMatchObject({ + accepted: true, + childWorkId: 'child-1', + created: true + }) + expect( + admission.announce( + announce(parent, { + aliases: [ + { segmentId: 'segment-1', aliasKind: 'task_id', alias: 'task-1' }, + { segmentId: 'segment-1', aliasKind: 'tool_use_id', alias: 'tool-2' } + ], + observedAt: 11 + }) + ) + ).toMatchObject({ accepted: true, childWorkId: 'child-1', created: false }) + expect(mintChildWorkId).toHaveBeenCalledTimes(1) + expect(store.getChildren(parent)).toHaveLength(1) + expect(store.getSnapshot().aliases).toHaveLength(2) + }) + + it('adopts and reclassifies a provisional child without changing its id', () => { + const parent = subject() + const { admission, store } = setup([parent]) + const first = admission.announce( + announce(parent, { + kind: 'unknown', + aliases: [{ segmentId: 'segment-1', aliasKind: 'tool_use_id', alias: 'tool-1' }] + }) + ) + expect(first).toMatchObject({ accepted: true, childWorkId: 'child-1' }) + + expect( + admission.adopt({ + ...announce(parent, { kind: 'agent', observedAt: 12 }), + childWorkId: 'child-1', + expectedFence: { invocationId: 'invocation-1', generation: 1 }, + aliases: [{ segmentId: 'segment-1', aliasKind: 'task_id', alias: 'task-1' }] + }) + ).toMatchObject({ accepted: true, childWorkId: 'child-1', created: false }) + expect(store.getChildren(parent)).toMatchObject([{ childWorkId: 'child-1', kind: 'agent' }]) + expect(store.getSnapshot().aliases.every((alias) => alias.kind === 'agent')).toBe(true) + }) + + it('preserves the logical id and prior outcome across an explicit resume fence', () => { + const parent = subject() + const { admission, store } = setup([parent]) + admission.announce( + announce(parent, { + state: 'done', + membership: 'settled', + outcome: 'failed', + observedAt: 20 + }) + ) + + expect( + admission.resume({ + ...announce(parent, { + aliases: [{ segmentId: 'segment-2', aliasKind: 'task_id', alias: 'task-1' }], + observedAt: 30 + }), + childWorkId: 'child-1', + expectedFence: { invocationId: 'invocation-1', generation: 1 }, + nextFence: { invocationId: 'invocation-2', generation: 2 } + }) + ).toMatchObject({ accepted: true, childWorkId: 'child-1', created: false }) + + expect(store.getChildren(parent)[0]).toMatchObject({ + childWorkId: 'child-1', + firstObservedAt: 20, + invocation: { invocationId: 'invocation-2', generation: 2 }, + previousInvocations: [ + { + fence: { invocationId: 'invocation-1', generation: 1 }, + outcome: 'failed', + settledAt: 20 + } + ] + }) + }) + + it('mints a distinct id for proven reuse and rejects delayed predecessor updates and stops', () => { + const parent = subject() + const { admission, store } = setup([parent]) + admission.announce( + announce(parent, { + state: 'done', + membership: 'settled', + outcome: 'succeeded', + observedAt: 20 + }) + ) + const successor = announce(parent, { + fence: { invocationId: 'invocation-2', generation: 2 }, + lifetime: 'proven-new', + observedAt: 30 + }) + expect(admission.announce(successor)).toMatchObject({ + accepted: true, + childWorkId: 'child-2', + created: true + }) + expect(store.getChildren(parent).map((child) => child.childWorkId)).toEqual([ + 'child-1', + 'child-2' + ]) + + expect(admission.announce(announce(parent, { observedAt: 40 }))).toEqual({ + accepted: false, + reason: 'stale-invocation' + }) + expect( + admission.authorizeStop({ + parent, + childWorkId: 'child-2', + expectedFence: { invocationId: 'invocation-1', generation: 1 } + }) + ).toBeNull() + expect( + admission.authorizeStop({ + parent, + childWorkId: 'child-2', + expectedFence: { invocationId: 'invocation-2', generation: 2 } + })?.childWorkId + ).toBe('child-2') + }) + + it('fails closed for settled or non-stoppable stop targets', () => { + const parent = subject() + const { admission } = setup([parent]) + admission.announce(announce(parent, { stoppable: false })) + const stop = { + parent, + childWorkId: 'child-1', + expectedFence: { invocationId: 'invocation-1', generation: 1 } + } + expect(admission.authorizeStop(stop)).toBeNull() + + const second = setup([parent]) + second.admission.announce( + announce(parent, { state: 'done', membership: 'settled', outcome: 'succeeded' }) + ) + expect(second.admission.authorizeStop(stop)).toBeNull() + }) + + it('scopes identical aliases by parent, provider, segment and kind', () => { + const firstParent = subject() + const secondParent = subject({ executionHostId: 'ssh:host-a' }) + const { admission, store } = setup([firstParent, secondParent]) + + const requests = [ + announce(firstParent), + announce(secondParent), + announce(firstParent, { provider: 'codex' }), + announce(firstParent, { + aliases: [{ segmentId: 'segment-2', aliasKind: 'task_id', alias: 'task-1' }] + }), + announce(firstParent, { kind: 'workflow' }) + ] + for (const request of requests) { + expect(admission.announce(request).accepted).toBe(true) + } + + expect(store.getSnapshot().children).toHaveLength(5) + expect(new Set(store.getSnapshot().aliases.map(serializeAgentChildWorkAliasKey)).size).toBe(5) + }) + + it('reparents a child and all aliases without reminting identity', () => { + const firstParent = subject() + const nextParent = subject({ workspaceId: 'workspace-2' }) + const { admission, store, mintChildWorkId } = setup([firstParent, nextParent]) + admission.announce(announce(firstParent)) + + expect( + admission.reparent({ + childWorkId: 'child-1', + fromParent: firstParent, + toParent: nextParent, + expectedFence: { invocationId: 'invocation-1', generation: 1 }, + observedAt: 20 + }) + ).toMatchObject({ accepted: true, childWorkId: 'child-1', created: false }) + expect(store.getChildren(firstParent)).toEqual([]) + expect(store.getChildren(nextParent)[0]?.childWorkId).toBe('child-1') + expect(store.getSnapshot().aliases[0]?.parent).toEqual(nextParent) + expect(mintChildWorkId).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/shared/agent-status-child-work-admission.ts b/src/shared/agent-status-child-work-admission.ts new file mode 100644 index 00000000000..2cf29699580 --- /dev/null +++ b/src/shared/agent-status-child-work-admission.ts @@ -0,0 +1,116 @@ +import type { AgentChildWorkAliasKind } from './agent-status-child-work-alias' +import type { + AgentChildWorkId, + AgentChildWorkInvocationFence, + AgentChildWorkKind, + AgentChildWorkMembership, + AgentChildWorkOutcome, + AgentChildWorkProviderTiming, + AgentChildWorkProvenance, + AgentChildWorkRecord, + AgentChildWorkState +} from './agent-status-child-work' +import { + adoptAgentChildWork, + announceAgentChildWork, + authorizeAgentChildWorkStop, + reparentAgentChildWork, + resumeAgentChildWork +} from './agent-status-child-work-admission-operations' +import type { AgentStatusStore } from './agent-status-store' +import type { AgentStatusSubject } from './agent-status-subject' + +export type AgentChildWorkObservationAlias = { + segmentId: string + aliasKind: AgentChildWorkAliasKind + alias: string +} + +export type AgentChildWorkObservationFields = { + kind: AgentChildWorkKind + state: AgentChildWorkState + membership: AgentChildWorkMembership + outcome?: AgentChildWorkOutcome + name?: string + description?: string + agentType?: string + model?: string + totalTokens?: number + providerTiming?: AgentChildWorkProviderTiming + observedAt: number + stoppable: boolean + provenance: AgentChildWorkProvenance +} + +export type AgentChildWorkAnnounceRequest = AgentChildWorkObservationFields & { + parent: AgentStatusSubject + provider: string + aliases: AgentChildWorkObservationAlias[] + fence: AgentChildWorkInvocationFence + lifetime: 'current' | 'proven-new' +} + +export type AgentChildWorkAdoptRequest = AgentChildWorkObservationFields & { + parent: AgentStatusSubject + provider: string + childWorkId: AgentChildWorkId + expectedFence: AgentChildWorkInvocationFence + aliases: AgentChildWorkObservationAlias[] +} + +export type AgentChildWorkResumeRequest = AgentChildWorkObservationFields & { + parent: AgentStatusSubject + provider: string + childWorkId: AgentChildWorkId + expectedFence: AgentChildWorkInvocationFence + nextFence: AgentChildWorkInvocationFence + aliases: AgentChildWorkObservationAlias[] +} + +export type AgentChildWorkReparentRequest = { + childWorkId: AgentChildWorkId + fromParent: AgentStatusSubject + toParent: AgentStatusSubject + expectedFence: AgentChildWorkInvocationFence + observedAt: number +} + +export type AgentChildWorkStopRequest = { + parent: AgentStatusSubject + childWorkId: AgentChildWorkId + expectedFence: AgentChildWorkInvocationFence +} + +export type AgentChildWorkAdmissionResult = + | { accepted: true; childWorkId: AgentChildWorkId; revision: number; created: boolean } + | { + accepted: false + reason: + | 'invalid' + | 'ambiguous' + | 'stale-invocation' + | 'unknown-child' + | 'id-collision' + | 'store-rejected' + } + +export type AgentChildWorkAdmission = { + announce(request: AgentChildWorkAnnounceRequest): AgentChildWorkAdmissionResult + adopt(request: AgentChildWorkAdoptRequest): AgentChildWorkAdmissionResult + resume(request: AgentChildWorkResumeRequest): AgentChildWorkAdmissionResult + reparent(request: AgentChildWorkReparentRequest): AgentChildWorkAdmissionResult + authorizeStop(request: AgentChildWorkStopRequest): AgentChildWorkRecord | null +} + +export function createAgentChildWorkAdmission( + store: AgentStatusStore, + options: { mintChildWorkId: () => AgentChildWorkId } +): AgentChildWorkAdmission { + return { + announce: (request) => announceAgentChildWork(store, options.mintChildWorkId, request), + adopt: (request) => adoptAgentChildWork(store, request), + resume: (request) => resumeAgentChildWork(store, request), + reparent: (request) => reparentAgentChildWork(store, request), + authorizeStop: (request) => authorizeAgentChildWorkStop(store, request) + } +} diff --git a/src/shared/agent-status-child-work-alias.ts b/src/shared/agent-status-child-work-alias.ts new file mode 100644 index 00000000000..323ea6d50b5 --- /dev/null +++ b/src/shared/agent-status-child-work-alias.ts @@ -0,0 +1,214 @@ +import { + agentChildWorkFencesEqual, + type AgentChildWorkId, + type AgentChildWorkInvocationFence, + type AgentChildWorkKind +} from './agent-status-child-work' +import { parseAgentChildWorkInvocationFence } from './agent-status-child-work-codec' +import { + deserializeAgentStatusSubject, + parseAgentStatusSubject, + serializeAgentStatusSubject, + type AgentStatusSubject +} from './agent-status-subject' + +const CHILD_ALIAS_KEY_PREFIX = 'agent-child-work-alias-v1:' +const MAX_ALIAS_PART_LENGTH = 512 + +export type AgentChildWorkAliasKind = 'task_id' | 'tool_use_id' + +export type AgentChildWorkAliasIdentity = Pick< + AgentChildWorkAliasInput, + 'parent' | 'provider' | 'segmentId' | 'kind' | 'aliasKind' | 'alias' +> + +export type AgentChildWorkAliasInput = { + parent: AgentStatusSubject + provider: string + segmentId: string + kind: AgentChildWorkKind + aliasKind: AgentChildWorkAliasKind + alias: string + childWorkId: AgentChildWorkId + fence: AgentChildWorkInvocationFence +} + +export type AgentChildWorkAliasRecord = AgentChildWorkAliasInput & { + revision: number +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasOnlyKeys( + record: Record, + required: readonly string[], + optional: readonly string[] = [] +): boolean { + const keys = Object.keys(record) + return ( + required.every((key) => Object.hasOwn(record, key)) && + keys.every((key) => required.includes(key) || optional.includes(key)) + ) +} + +function isBoundedString(value: unknown): value is string { + if ( + typeof value !== 'string' || + value.length === 0 || + value.length > MAX_ALIAS_PART_LENGTH || + value !== value.trim() + ) { + return false + } + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index) + if (code <= 0x1f || code === 0x7f) { + return false + } + } + return true +} + +function isRevision(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +function isKind(value: unknown): value is AgentChildWorkKind { + return ( + value === 'agent' || + value === 'workflow' || + value === 'command' || + value === 'monitor' || + value === 'unknown' + ) +} + +export function parseAgentChildWorkAliasInput(value: unknown): AgentChildWorkAliasInput | null { + if ( + !isRecord(value) || + !hasOnlyKeys(value, [ + 'parent', + 'provider', + 'segmentId', + 'kind', + 'aliasKind', + 'alias', + 'childWorkId', + 'fence' + ]) || + !isBoundedString(value.provider) || + !isBoundedString(value.segmentId) || + !isKind(value.kind) || + (value.aliasKind !== 'task_id' && value.aliasKind !== 'tool_use_id') || + !isBoundedString(value.alias) || + !isBoundedString(value.childWorkId) + ) { + return null + } + const parent = parseAgentStatusSubject(value.parent) + const fence = parseAgentChildWorkInvocationFence(value.fence) + if (!parent || !fence) { + return null + } + return { + parent, + provider: value.provider, + segmentId: value.segmentId, + kind: value.kind, + aliasKind: value.aliasKind, + alias: value.alias, + childWorkId: value.childWorkId, + fence + } +} + +export function parseAgentChildWorkAliasRecord(value: unknown): AgentChildWorkAliasRecord | null { + if (!isRecord(value) || !isRevision(value.revision)) { + return null + } + const input = { ...value } + delete input.revision + const parsed = parseAgentChildWorkAliasInput(input) + return parsed ? { ...parsed, revision: value.revision } : null +} + +export function serializeAgentChildWorkAliasKey(alias: AgentChildWorkAliasIdentity): string { + const parsed = parseAgentChildWorkAliasInput({ + parent: alias.parent, + provider: alias.provider, + segmentId: alias.segmentId, + kind: alias.kind, + aliasKind: alias.aliasKind, + alias: alias.alias, + childWorkId: 'key-only', + fence: { invocationId: 'key-only', generation: 0 } + }) + if (!parsed) { + throw new Error('Invalid agent child-work alias') + } + return `${CHILD_ALIAS_KEY_PREFIX}${JSON.stringify([ + serializeAgentStatusSubject(parsed.parent), + parsed.provider, + parsed.segmentId, + parsed.kind, + parsed.aliasKind, + parsed.alias + ])}` +} + +export function deserializeAgentChildWorkAliasKey( + value: string +): AgentChildWorkAliasIdentity | null { + if (!value.startsWith(CHILD_ALIAS_KEY_PREFIX)) { + return null + } + let tuple: unknown + try { + tuple = JSON.parse(value.slice(CHILD_ALIAS_KEY_PREFIX.length)) + } catch { + return null + } + if (!Array.isArray(tuple) || tuple.length !== 6) { + return null + } + const [parentKey, provider, segmentId, kind, aliasKind, alias] = tuple + if (typeof parentKey !== 'string') { + return null + } + const parent = deserializeAgentStatusSubject(parentKey) + const parsed = parseAgentChildWorkAliasInput({ + parent, + provider, + segmentId, + kind, + aliasKind, + alias, + childWorkId: 'key-only', + fence: { invocationId: 'key-only', generation: 0 } + }) + if (!parsed) { + return null + } + const identity = { + parent: parsed.parent, + provider: parsed.provider, + segmentId: parsed.segmentId, + kind: parsed.kind, + aliasKind: parsed.aliasKind, + alias: parsed.alias + } + return serializeAgentChildWorkAliasKey(identity) === value ? identity : null +} + +export function agentChildWorkAliasesMatch( + left: AgentChildWorkAliasInput, + right: AgentChildWorkAliasInput +): boolean { + return ( + serializeAgentChildWorkAliasKey(left) === serializeAgentChildWorkAliasKey(right) && + left.childWorkId === right.childWorkId && + agentChildWorkFencesEqual(left.fence, right.fence) + ) +} diff --git a/src/shared/agent-status-child-work-codec.ts b/src/shared/agent-status-child-work-codec.ts new file mode 100644 index 00000000000..616a7430022 --- /dev/null +++ b/src/shared/agent-status-child-work-codec.ts @@ -0,0 +1,273 @@ +import { + AGENT_CHILD_WORK_INVOCATION_HISTORY_MAX, + AGENT_CHILD_WORK_KINDS, + AGENT_CHILD_WORK_MEMBERSHIPS, + AGENT_CHILD_WORK_OUTCOMES, + AGENT_CHILD_WORK_STATES, + agentChildWorkFencesEqual, + type AgentChildWorkInput, + type AgentChildWorkInvocationFence, + type AgentChildWorkInvocationHistory, + type AgentChildWorkKind, + type AgentChildWorkMembership, + type AgentChildWorkOutcome, + type AgentChildWorkProviderTiming, + type AgentChildWorkProvenance, + type AgentChildWorkRecord, + type AgentChildWorkState +} from './agent-status-child-work' +import { parseAgentStatusSubject } from './agent-status-subject' + +const MAX_ID_LENGTH = 256 +const MAX_LABEL_LENGTH = 512 +const MAX_DESCRIPTION_LENGTH = 8_000 +const CHILD_WORK_KIND_SET: ReadonlySet = new Set(AGENT_CHILD_WORK_KINDS) +const CHILD_WORK_STATE_SET: ReadonlySet = new Set(AGENT_CHILD_WORK_STATES) +const CHILD_WORK_MEMBERSHIP_SET: ReadonlySet = new Set(AGENT_CHILD_WORK_MEMBERSHIPS) +const CHILD_WORK_OUTCOME_SET: ReadonlySet = new Set(AGENT_CHILD_WORK_OUTCOMES) + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasOnlyKeys( + record: Record, + required: readonly string[], + optional: readonly string[] = [] +): boolean { + const keys = Object.keys(record) + return ( + required.every((key) => Object.hasOwn(record, key)) && + keys.every((key) => required.includes(key) || optional.includes(key)) + ) +} + +function isBoundedString(value: unknown, maxLength = MAX_ID_LENGTH): value is string { + if ( + typeof value !== 'string' || + value.length === 0 || + value.length > maxLength || + value !== value.trim() + ) { + return false + } + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index) + if (code <= 0x1f || code === 0x7f) { + return false + } + } + return true +} + +function isTimestamp(value: unknown): value is number { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 +} + +function isRevision(value: unknown): value is number { + return Number.isSafeInteger(value) && typeof value === 'number' && value >= 0 +} + +function isKind(value: unknown): value is AgentChildWorkKind { + return typeof value === 'string' && CHILD_WORK_KIND_SET.has(value) +} + +function isState(value: unknown): value is AgentChildWorkState { + return typeof value === 'string' && CHILD_WORK_STATE_SET.has(value) +} + +function isMembership(value: unknown): value is AgentChildWorkMembership { + return typeof value === 'string' && CHILD_WORK_MEMBERSHIP_SET.has(value) +} + +function isOutcome(value: unknown): value is AgentChildWorkOutcome { + return typeof value === 'string' && CHILD_WORK_OUTCOME_SET.has(value) +} + +export function parseAgentChildWorkInvocationFence( + value: unknown +): AgentChildWorkInvocationFence | null { + if ( + !isRecord(value) || + !hasOnlyKeys(value, ['invocationId', 'generation']) || + !isBoundedString(value.invocationId) || + !isRevision(value.generation) + ) { + return null + } + return { invocationId: value.invocationId, generation: value.generation } +} + +function parseProviderTiming(value: unknown): AgentChildWorkProviderTiming | null { + if (!isRecord(value) || !hasOnlyKeys(value, [], ['startedAt', 'completedAt'])) { + return null + } + if ( + (value.startedAt !== undefined && !isTimestamp(value.startedAt)) || + (value.completedAt !== undefined && !isTimestamp(value.completedAt)) + ) { + return null + } + return { + ...(isTimestamp(value.startedAt) ? { startedAt: value.startedAt } : {}), + ...(isTimestamp(value.completedAt) ? { completedAt: value.completedAt } : {}) + } +} + +function parseProvenance(value: unknown): AgentChildWorkProvenance | null { + if ( + !isRecord(value) || + !hasOnlyKeys(value, ['source', 'producerId']) || + (value.source !== 'hook' && + value.source !== 'structured-session' && + value.source !== 'restore' && + value.source !== 'transport') || + !isBoundedString(value.producerId) + ) { + return null + } + return { source: value.source, producerId: value.producerId } +} + +function parseInvocationHistory(value: unknown): AgentChildWorkInvocationHistory[] | null { + if (!Array.isArray(value) || value.length > AGENT_CHILD_WORK_INVOCATION_HISTORY_MAX) { + return null + } + const history: AgentChildWorkInvocationHistory[] = [] + for (const candidate of value) { + if ( + !isRecord(candidate) || + !hasOnlyKeys(candidate, ['fence'], ['outcome', 'settledAt']) || + (candidate.outcome !== undefined && !isOutcome(candidate.outcome)) || + (candidate.settledAt !== undefined && !isTimestamp(candidate.settledAt)) + ) { + return null + } + const fence = parseAgentChildWorkInvocationFence(candidate.fence) + if (!fence) { + return null + } + history.push({ + fence, + ...(isOutcome(candidate.outcome) ? { outcome: candidate.outcome } : {}), + ...(isTimestamp(candidate.settledAt) ? { settledAt: candidate.settledAt } : {}) + }) + } + return history +} + +function parseOptionalLabel(value: unknown, maxLength = MAX_LABEL_LENGTH): string | null { + return value === undefined ? '' : isBoundedString(value, maxLength) ? value : null +} + +export function parseAgentChildWorkInput(value: unknown): AgentChildWorkInput | null { + if ( + !isRecord(value) || + !hasOnlyKeys( + value, + [ + 'childWorkId', + 'parent', + 'provider', + 'kind', + 'state', + 'membership', + 'firstObservedAt', + 'observedAt', + 'stoppable', + 'invocation', + 'provenance' + ], + [ + 'outcome', + 'name', + 'description', + 'agentType', + 'model', + 'totalTokens', + 'providerTiming', + 'previousInvocations' + ] + ) || + !isBoundedString(value.childWorkId) || + !isBoundedString(value.provider) || + !isKind(value.kind) || + !isState(value.state) || + !isMembership(value.membership) || + (value.outcome !== undefined && !isOutcome(value.outcome)) || + (value.outcome !== undefined && value.membership !== 'settled') || + !isTimestamp(value.firstObservedAt) || + !isTimestamp(value.observedAt) || + value.firstObservedAt > value.observedAt || + typeof value.stoppable !== 'boolean' || + (value.totalTokens !== undefined && + (typeof value.totalTokens !== 'number' || + !Number.isSafeInteger(value.totalTokens) || + value.totalTokens < 0)) + ) { + return null + } + const parent = parseAgentStatusSubject(value.parent) + const invocation = parseAgentChildWorkInvocationFence(value.invocation) + const provenance = parseProvenance(value.provenance) + const timing = + value.providerTiming === undefined ? undefined : parseProviderTiming(value.providerTiming) + const history = + value.previousInvocations === undefined + ? undefined + : parseInvocationHistory(value.previousInvocations) + const labels = { + name: parseOptionalLabel(value.name), + description: parseOptionalLabel(value.description, MAX_DESCRIPTION_LENGTH), + agentType: parseOptionalLabel(value.agentType), + model: parseOptionalLabel(value.model) + } + if (!parent || !invocation || !provenance || timing === null || history === null) { + return null + } + if (Object.values(labels).includes(null)) { + return null + } + const historyFenceKeys = history?.map( + (entry) => `${entry.fence.invocationId}\0${entry.fence.generation}` + ) + if ( + history && + historyFenceKeys && + (new Set(historyFenceKeys).size !== historyFenceKeys.length || + history.some((entry) => agentChildWorkFencesEqual(entry.fence, invocation))) + ) { + return null + } + return { + childWorkId: value.childWorkId, + parent, + provider: value.provider, + kind: value.kind, + state: value.state, + membership: value.membership, + ...(isOutcome(value.outcome) ? { outcome: value.outcome } : {}), + ...(labels.name ? { name: labels.name } : {}), + ...(labels.description ? { description: labels.description } : {}), + ...(labels.agentType ? { agentType: labels.agentType } : {}), + ...(labels.model ? { model: labels.model } : {}), + ...(typeof value.totalTokens === 'number' ? { totalTokens: value.totalTokens } : {}), + ...(timing ? { providerTiming: timing } : {}), + firstObservedAt: value.firstObservedAt, + observedAt: value.observedAt, + stoppable: value.stoppable, + invocation, + ...(history ? { previousInvocations: history } : {}), + provenance + } +} + +export function parseAgentChildWorkRecord(value: unknown): AgentChildWorkRecord | null { + if (!isRecord(value) || !isRevision(value.revision)) { + return null + } + const input = { ...value } + delete input.revision + const parsed = parseAgentChildWorkInput(input) + return parsed ? { ...parsed, revision: value.revision } : null +} diff --git a/src/shared/agent-status-child-work-freshness.ts b/src/shared/agent-status-child-work-freshness.ts new file mode 100644 index 00000000000..ca513c81143 --- /dev/null +++ b/src/shared/agent-status-child-work-freshness.ts @@ -0,0 +1,20 @@ +import type { AgentChildWorkMembership, AgentChildWorkState } from './agent-status-child-work' + +export type AgentChildWorkFreshnessInput = { + state: AgentChildWorkState + membership: AgentChildWorkMembership + parentEvidenceFresh: boolean + transportObservation: 'live' | 'unverifiable' +} + +/** One decay rule for CLI and structured children; freshness never rewrites settled history. */ +export function resolveAgentChildWorkFreshness( + input: AgentChildWorkFreshnessInput +): AgentChildWorkState { + if (input.membership === 'settled') { + return input.state + } + return input.parentEvidenceFresh && input.transportObservation === 'live' + ? input.state + : 'unverifiable' +} diff --git a/src/shared/agent-status-child-work-projection.test.ts b/src/shared/agent-status-child-work-projection.test.ts new file mode 100644 index 00000000000..91200735498 --- /dev/null +++ b/src/shared/agent-status-child-work-projection.test.ts @@ -0,0 +1,166 @@ +import { describe, expect, it } from 'vitest' +import { AGENT_STATUS_MAX_SUBAGENTS } from './agent-status-types' +import { resolveAgentChildWorkFreshness } from './agent-status-child-work-freshness' +import { + projectAgentChildWorkLegacyBackgroundTasks, + projectAgentChildWorkLegacySubagents, + type AgentChildWorkLegacyProjectionCandidate +} from './agent-status-child-work-projection' +import type { AgentChildWorkState } from './agent-status-child-work' + +function candidate( + providerId: string, + overrides: Partial = {} +): AgentChildWorkLegacyProjectionCandidate { + return { + providerId, + child: { + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: 123, + description: 'Investigate', + agentType: 'researcher', + model: 'model-a', + stoppable: true, + ...overrides + } + } +} + +describe('agent child-work legacy projection', () => { + it('preserves the bridge state mapping and stable host first-observation time', () => { + const states: (AgentChildWorkState | undefined)[] = [ + undefined, + 'working', + 'monitoring', + 'done', + 'idle', + 'waiting', + 'blocked', + 'unverifiable' + ] + const projected = projectAgentChildWorkLegacySubagents( + states.map((state, index) => candidate(`task-${index}`, { state })) + ) + + expect(projected?.map((item) => item.state)).toEqual([ + 'working', + 'working', + 'working', + 'idle', + 'idle', + 'waiting', + 'blocked', + 'unverifiable' + ]) + expect(projected?.every((item) => item.startedAt === 123)).toBe(true) + }) + + it('admits only agent kind and trims bounded nonempty provider ids', () => { + const projected = projectAgentChildWorkLegacySubagents([ + candidate(' valid-id '), + candidate(''), + candidate(' '.repeat(10)), + candidate('x'.repeat(65)), + candidate('workflow-id', { kind: 'workflow' }), + candidate('command-id', { kind: 'command' }), + candidate('monitor-id', { kind: 'monitor' }), + candidate('unknown-id', { kind: 'unknown' }) + ]) + + expect(projected).toEqual([ + { + id: 'valid-id', + state: 'working', + startedAt: 123, + agentType: 'researcher', + model: 'model-a', + description: 'Investigate' + } + ]) + }) + + it.each([31, 32, 33])('caps after accepting %i valid rows in source order', (count) => { + const interleaved = Array.from({ length: count }, (_, index) => [ + candidate('', { kind: 'agent' }), + candidate(`task-${index}`) + ]).flat() + const projected = projectAgentChildWorkLegacySubagents(interleaved) + + expect(projected).toHaveLength(Math.min(count, AGENT_STATUS_MAX_SUBAGENTS)) + expect(projected?.at(-1)?.id).toBe(`task-${Math.min(count, AGENT_STATUS_MAX_SUBAGENTS) - 1}`) + }) + + it('rejects missing host first-observation time instead of inventing zero', () => { + expect( + projectAgentChildWorkLegacySubagents([ + candidate('task-invalid', { firstObservedAt: Number.NaN }), + candidate('task-valid', { firstObservedAt: 55 }) + ]) + ).toMatchObject([{ id: 'task-valid', startedAt: 55 }]) + }) + + it('projects every kind into separate live and settled background lists', () => { + const projection = projectAgentChildWorkLegacyBackgroundTasks([ + candidate('agent-live', { kind: 'agent', totalTokens: 10 }), + candidate('workflow-settled', { + kind: 'workflow', + state: 'done', + membership: 'settled' + }), + candidate('command-live', { kind: 'command' }), + candidate('monitor-live', { kind: 'monitor', state: 'monitoring' }), + candidate('unknown-settled', { kind: 'unknown', state: 'idle', membership: 'settled' }) + ]) + + expect(projection.tasks?.map((item) => item.kind)).toEqual(['agent', 'command', 'monitor']) + expect(projection.settledTasks?.map((item) => item.kind)).toEqual(['workflow', 'unknown']) + expect(projection.tasks?.[0]).toMatchObject({ + id: 'agent-live', + startedAt: 123, + totalTokens: 10, + stoppable: true + }) + }) +}) + +describe('resolveAgentChildWorkFreshness', () => { + it.each([ + [true, 'live', 'working'], + [false, 'live', 'unverifiable'], + [true, 'unverifiable', 'unverifiable'], + [false, 'unverifiable', 'unverifiable'] + ] as const)( + 'maps parentFresh=%s transport=%s to %s for live work', + (parentEvidenceFresh, transportObservation, expected) => { + expect( + resolveAgentChildWorkFreshness({ + state: 'working', + membership: 'live', + parentEvidenceFresh, + transportObservation + }) + ).toBe(expected) + } + ) + + it('does not turn live idle into settled or rewrite settled history on contact loss', () => { + expect( + resolveAgentChildWorkFreshness({ + state: 'idle', + membership: 'live', + parentEvidenceFresh: false, + transportObservation: 'live' + }) + ).toBe('unverifiable') + expect( + resolveAgentChildWorkFreshness({ + state: 'done', + membership: 'settled', + parentEvidenceFresh: false, + transportObservation: 'unverifiable' + }) + ).toBe('done') + }) +}) diff --git a/src/shared/agent-status-child-work-projection.ts b/src/shared/agent-status-child-work-projection.ts new file mode 100644 index 00000000000..f2787e9577c --- /dev/null +++ b/src/shared/agent-status-child-work-projection.ts @@ -0,0 +1,146 @@ +import type { AgentSessionBackgroundTask } from './agent-session-background-task-wire' +import { AGENT_STATUS_MAX_SUBAGENTS, type AgentSubagentSnapshot } from './agent-status-types' +import type { + AgentChildWorkKind, + AgentChildWorkMembership, + AgentChildWorkState +} from './agent-status-child-work' + +const LEGACY_PROVIDER_ID_MAX_LENGTH = 64 +const BACKGROUND_PROVIDER_ID_MAX_LENGTH = 512 + +export type AgentChildWorkLegacyProjectionCandidate = { + providerId: string + child: { + kind: AgentChildWorkKind + state?: AgentChildWorkState + membership: AgentChildWorkMembership + firstObservedAt: number + name?: string + description?: string + agentType?: string + model?: string + totalTokens?: number + stoppable: boolean + } +} + +function legacyProviderId(value: unknown): string | null { + if (typeof value !== 'string') { + return null + } + const trimmed = value.trim() + return trimmed.length > 0 && trimmed.length <= LEGACY_PROVIDER_ID_MAX_LENGTH ? trimmed : null +} + +function backgroundProviderId(value: unknown): string | null { + if (typeof value !== 'string') { + return null + } + const trimmed = value.trim() + return trimmed.length > 0 && trimmed.length <= BACKGROUND_PROVIDER_ID_MAX_LENGTH ? trimmed : null +} + +function legacySubagentState( + state: AgentChildWorkState | undefined +): AgentSubagentSnapshot['state'] | null { + if (state === undefined || state === 'working' || state === 'monitoring') { + return 'working' + } + if (state === 'done' || state === 'idle') { + return 'idle' + } + if (state === 'waiting' || state === 'blocked' || state === 'unverifiable') { + return state + } + return null +} + +export function projectAgentChildWorkLegacySubagents( + candidates: readonly AgentChildWorkLegacyProjectionCandidate[] +): AgentSubagentSnapshot[] | undefined { + const projected: AgentSubagentSnapshot[] = [] + for (const candidate of candidates) { + if (candidate.child.kind !== 'agent') { + continue + } + const id = legacyProviderId(candidate.providerId) + const state = legacySubagentState(candidate.child.state) + if ( + !id || + !state || + !Number.isFinite(candidate.child.firstObservedAt) || + candidate.child.firstObservedAt < 0 + ) { + continue + } + projected.push({ + id, + state, + startedAt: candidate.child.firstObservedAt, + ...(candidate.child.agentType !== undefined ? { agentType: candidate.child.agentType } : {}), + ...(candidate.child.model !== undefined ? { model: candidate.child.model } : {}), + ...(candidate.child.description !== undefined + ? { description: candidate.child.description } + : {}) + }) + if (projected.length === AGENT_STATUS_MAX_SUBAGENTS) { + break + } + } + return projected.length > 0 ? projected : undefined +} + +export type AgentChildWorkLegacyBackgroundProjection = { + tasks?: AgentSessionBackgroundTask[] + settledTasks?: AgentSessionBackgroundTask[] +} + +function projectBackgroundTask( + candidate: AgentChildWorkLegacyProjectionCandidate +): AgentSessionBackgroundTask | null { + const id = backgroundProviderId(candidate.providerId) + if ( + !id || + !Number.isFinite(candidate.child.firstObservedAt) || + candidate.child.firstObservedAt < 0 + ) { + return null + } + return { + id, + kind: candidate.child.kind, + ...(candidate.child.description !== undefined + ? { description: candidate.child.description } + : {}), + ...(candidate.child.name !== undefined ? { name: candidate.child.name } : {}), + ...(candidate.child.state !== undefined ? { state: candidate.child.state } : {}), + startedAt: candidate.child.firstObservedAt, + ...(candidate.child.totalTokens !== undefined + ? { totalTokens: candidate.child.totalTokens } + : {}), + stoppable: candidate.child.stoppable + } +} + +export function projectAgentChildWorkLegacyBackgroundTasks( + candidates: readonly AgentChildWorkLegacyProjectionCandidate[] +): AgentChildWorkLegacyBackgroundProjection { + const tasks: AgentSessionBackgroundTask[] = [] + const settledTasks: AgentSessionBackgroundTask[] = [] + for (const candidate of candidates) { + const projected = projectBackgroundTask(candidate) + if (!projected) { + continue + } + if (candidate.child.membership === 'live') { + tasks.push(projected) + } else { + settledTasks.push(projected) + } + } + return { + ...(tasks.length > 0 ? { tasks } : {}), + ...(settledTasks.length > 0 ? { settledTasks } : {}) + } +} diff --git a/src/shared/agent-status-child-work.test.ts b/src/shared/agent-status-child-work.test.ts new file mode 100644 index 00000000000..3d5d933615c --- /dev/null +++ b/src/shared/agent-status-child-work.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from 'vitest' +import { + AGENT_CHILD_WORK_KINDS, + AGENT_CHILD_WORK_STATES, + type AgentChildWorkInput +} from './agent-status-child-work' +import { + parseAgentChildWorkInput, + parseAgentChildWorkRecord +} from './agent-status-child-work-codec' +import { createAgentStatusStore } from './agent-status-store' +import { makeStructuredAgentStatusSubject } from './agent-status-subject' + +const parent = makeStructuredAgentStatusSubject( + { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder-1', + workspaceKind: 'folder' + }, + 'session_11111111-1111-4111-8111-111111111111' +) + +function child(overrides: Partial = {}): AgentChildWorkInput { + return { + childWorkId: 'child-1', + parent, + provider: 'claude', + kind: 'agent', + state: 'working', + membership: 'live', + name: 'Research', + description: 'Investigate the contract', + agentType: 'researcher', + model: 'model-a', + totalTokens: 42, + providerTiming: { startedAt: 5, completedAt: 8 }, + firstObservedAt: 10, + observedAt: 20, + stoppable: true, + invocation: { invocationId: 'invocation-1', generation: 1 }, + previousInvocations: [ + { + fence: { invocationId: 'invocation-0', generation: 0 }, + outcome: 'cancelled', + settledAt: 9 + } + ], + provenance: { source: 'structured-session', producerId: 'journal-1' }, + ...overrides + } +} + +describe('AgentChildWorkRecord', () => { + it('round-trips the complete canonical vocabulary through the store snapshot', () => { + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + expect(store.applyMutation({ parent: { subject: parent } })).not.toBeNull() + const children = AGENT_CHILD_WORK_STATES.map((state, index) => + child({ + childWorkId: `child-${index}`, + kind: AGENT_CHILD_WORK_KINDS[index % AGENT_CHILD_WORK_KINDS.length], + state, + membership: state === 'done' ? 'settled' : 'live', + ...(state === 'done' ? { outcome: 'failed' } : {}), + invocation: { invocationId: `invocation-${index}`, generation: index }, + previousInvocations: undefined + }) + ) + + expect(store.applyMutation({ children })).not.toBeNull() + const snapshot = store.getSnapshot() + expect(snapshot.children.map((item) => item.kind)).toEqual([ + 'agent', + 'workflow', + 'command', + 'monitor', + 'unknown', + 'agent', + 'workflow' + ]) + expect(snapshot.children.map((item) => item.state)).toEqual(AGENT_CHILD_WORK_STATES) + expect(snapshot.children.find((item) => item.state === 'done')).toMatchObject({ + membership: 'settled', + outcome: 'failed', + totalTokens: 42, + providerTiming: { startedAt: 5, completedAt: 8 }, + firstObservedAt: 10 + }) + }) + + it('parses a copied record with outcome, tokens, timing and bounded invocation history', () => { + const parsed = parseAgentChildWorkRecord({ ...child(), revision: 7 }) + + expect(parsed).toEqual({ ...child(), revision: 7 }) + expect(parsed).not.toBe(child()) + expect(parsed?.parent).not.toBe(parent) + }) + + it.each([ + child({ childWorkId: 'bad\nid' }), + child({ provider: 'claude\0forged' }), + child({ firstObservedAt: 21 }), + child({ totalTokens: -1 }), + child({ membership: 'live', outcome: 'succeeded' }), + child({ + previousInvocations: [ + { fence: { invocationId: 'invocation-1', generation: 1 }, outcome: 'failed' } + ] + }), + child({ + previousInvocations: [ + { fence: { invocationId: 'old', generation: 1 } }, + { fence: { invocationId: 'old', generation: 1 } } + ] + }) + ])('rejects malformed canonical child input %#', (value) => { + expect(parseAgentChildWorkInput(value)).toBeNull() + }) +}) diff --git a/src/shared/agent-status-child-work.ts b/src/shared/agent-status-child-work.ts new file mode 100644 index 00000000000..c82d25b42c8 --- /dev/null +++ b/src/shared/agent-status-child-work.ts @@ -0,0 +1,88 @@ +import { agentStatusSubjectsEqual, type AgentStatusSubject } from './agent-status-subject' + +export const AGENT_CHILD_WORK_KINDS = [ + 'agent', + 'workflow', + 'command', + 'monitor', + 'unknown' +] as const +export const AGENT_CHILD_WORK_STATES = [ + 'working', + 'monitoring', + 'waiting', + 'blocked', + 'done', + 'idle', + 'unverifiable' +] as const +export const AGENT_CHILD_WORK_MEMBERSHIPS = ['live', 'settled'] as const +export const AGENT_CHILD_WORK_OUTCOMES = ['succeeded', 'failed', 'cancelled', 'unknown'] as const +export const AGENT_CHILD_WORK_INVOCATION_HISTORY_MAX = 32 + +export type AgentChildWorkId = string +export type AgentChildWorkKind = (typeof AGENT_CHILD_WORK_KINDS)[number] +export type AgentChildWorkState = (typeof AGENT_CHILD_WORK_STATES)[number] +export type AgentChildWorkMembership = (typeof AGENT_CHILD_WORK_MEMBERSHIPS)[number] +export type AgentChildWorkOutcome = (typeof AGENT_CHILD_WORK_OUTCOMES)[number] + +export type AgentChildWorkInvocationFence = { + invocationId: string + generation: number +} + +export type AgentChildWorkInvocationHistory = { + fence: AgentChildWorkInvocationFence + outcome?: AgentChildWorkOutcome + settledAt?: number +} + +export type AgentChildWorkProviderTiming = { + startedAt?: number + completedAt?: number +} + +export type AgentChildWorkProvenance = { + source: 'hook' | 'structured-session' | 'restore' | 'transport' + producerId: string +} + +export type AgentChildWorkInput = { + childWorkId: AgentChildWorkId + parent: AgentStatusSubject + provider: string + kind: AgentChildWorkKind + state: AgentChildWorkState + membership: AgentChildWorkMembership + outcome?: AgentChildWorkOutcome + name?: string + description?: string + agentType?: string + model?: string + totalTokens?: number + providerTiming?: AgentChildWorkProviderTiming + firstObservedAt: number + observedAt: number + stoppable: boolean + invocation: AgentChildWorkInvocationFence + previousInvocations?: AgentChildWorkInvocationHistory[] + provenance: AgentChildWorkProvenance +} + +export type AgentChildWorkRecord = AgentChildWorkInput & { + revision: number +} + +export function agentChildWorkFencesEqual( + left: AgentChildWorkInvocationFence, + right: AgentChildWorkInvocationFence +): boolean { + return left.invocationId === right.invocationId && left.generation === right.generation +} + +export function agentChildWorkBelongsTo( + child: Pick, + parent: AgentStatusSubject +): boolean { + return agentStatusSubjectsEqual(child.parent, parent) +} diff --git a/src/shared/agent-status-store-codec.ts b/src/shared/agent-status-store-codec.ts new file mode 100644 index 00000000000..5a0d6fc2878 --- /dev/null +++ b/src/shared/agent-status-store-codec.ts @@ -0,0 +1,284 @@ +import { + parseAgentChildWorkAliasInput, + parseAgentChildWorkAliasRecord +} from './agent-status-child-work-alias' +import { + parseAgentChildWorkInput, + parseAgentChildWorkRecord +} from './agent-status-child-work-codec' +import { + AGENT_STATUS_STORE_LIMITS, + AGENT_STATUS_STORE_SNAPSHOT_VERSION, + type AgentStatusStoreMutation, + type AgentStatusStoreSnapshot, + type AgentStatusTombstoneEntity, + type AgentStatusTombstoneInput, + type AgentStatusTombstoneRecord +} from './agent-status-store-contract' +import { + parseAgentStatusFactIdentity, + parseAgentStatusFactInput, + parseAgentStatusFactRecord +} from './agent-status-store-fact-codec' +import { + parseAgentStatusParentInput, + parseAgentStatusParentRecord +} from './agent-status-store-parent' +import { parseAgentStatusSubject } from './agent-status-subject' +import { measureUtf8ByteLength } from './utf8-byte-limits' + +const MAX_EPOCH_LENGTH = 256 +const MAX_TOMBSTONE_KEY_LENGTH = 4_096 + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasOnlyKeys( + record: Record, + required: readonly string[], + optional: readonly string[] = [] +): boolean { + const keys = Object.keys(record) + return ( + required.every((key) => Object.hasOwn(record, key)) && + keys.every((key) => required.includes(key) || optional.includes(key)) + ) +} + +function isRevision(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +function isWithinSerializedLimit(value: unknown): boolean { + try { + return !measureUtf8ByteLength(JSON.stringify(value), { + stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes + }).exceededLimit + } catch { + return false + } +} + +function isBoundedKey(value: unknown, maxLength: number): value is string { + if ( + typeof value !== 'string' || + value.length === 0 || + value.length > maxLength || + value !== value.trim() + ) { + return false + } + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index) + if (code <= 0x1f || code === 0x7f) { + return false + } + } + return true +} + +export function isAgentStatusStoreEpoch(value: unknown): value is string { + return isBoundedKey(value, MAX_EPOCH_LENGTH) +} + +function isTombstoneEntity(value: unknown): value is AgentStatusTombstoneEntity { + return value === 'parent' || value === 'child' || value === 'alias' || value === 'fact' +} + +export function parseAgentStatusTombstoneInput(value: unknown): AgentStatusTombstoneInput | null { + if ( + !isRecord(value) || + !hasOnlyKeys(value, ['entity', 'key']) || + !isTombstoneEntity(value.entity) || + !isBoundedKey(value.key, MAX_TOMBSTONE_KEY_LENGTH) + ) { + return null + } + return { entity: value.entity, key: value.key } +} + +export function parseAgentStatusTombstoneRecord(value: unknown): AgentStatusTombstoneRecord | null { + if (!isRecord(value) || !isRevision(value.revision)) { + return null + } + const input = { ...value } + delete input.revision + const tombstone = parseAgentStatusTombstoneInput(input) + return tombstone ? { ...tombstone, revision: value.revision } : null +} + +function parseArray( + value: unknown, + parser: (candidate: unknown) => T | null, + maxLength: number +): T[] | null { + if (!Array.isArray(value) || value.length > maxLength) { + return null + } + const parsed: T[] = [] + for (const candidate of value) { + const item = parser(candidate) + if (!item) { + return null + } + parsed.push(item) + } + return parsed +} + +function parseMutationArray( + value: unknown, + parser: (candidate: unknown) => T | null +): T[] | null { + return parseArray(value, parser, AGENT_STATUS_STORE_LIMITS.mutationEntries) +} + +function parseStringArray(value: unknown): string[] | null { + return parseMutationArray(value, (candidate) => + isBoundedKey(candidate, MAX_TOMBSTONE_KEY_LENGTH) ? candidate : null + ) +} + +export function parseAgentStatusStoreMutation(value: unknown): AgentStatusStoreMutation | null { + const optionalKeys = [ + 'parent', + 'removeParent', + 'children', + 'removeChildren', + 'aliases', + 'removeAliases', + 'facts', + 'removeFacts', + 'tombstones' + ] + if ( + !isWithinSerializedLimit(value) || + !isRecord(value) || + !hasOnlyKeys(value, [], optionalKeys) || + Object.keys(value).length === 0 + ) { + return null + } + const parent = value.parent === undefined ? undefined : parseAgentStatusParentInput(value.parent) + const removeParent = + value.removeParent === undefined ? undefined : parseAgentStatusSubject(value.removeParent) + const children = + value.children === undefined + ? undefined + : parseMutationArray(value.children, parseAgentChildWorkInput) + const removeChildren = + value.removeChildren === undefined ? undefined : parseStringArray(value.removeChildren) + const aliases = + value.aliases === undefined + ? undefined + : parseMutationArray(value.aliases, parseAgentChildWorkAliasInput) + const removeAliases = + value.removeAliases === undefined ? undefined : parseStringArray(value.removeAliases) + const facts = + value.facts === undefined + ? undefined + : parseMutationArray(value.facts, parseAgentStatusFactInput) + const removeFacts = + value.removeFacts === undefined + ? undefined + : parseMutationArray(value.removeFacts, parseAgentStatusFactIdentity) + const tombstones = + value.tombstones === undefined + ? undefined + : parseMutationArray(value.tombstones, parseAgentStatusTombstoneInput) + const parsedValues = [ + parent, + removeParent, + children, + removeChildren, + aliases, + removeAliases, + facts, + removeFacts, + tombstones + ] + const sourceValues = optionalKeys.map((key) => value[key]) + if (sourceValues.some((item, index) => item !== undefined && !parsedValues[index])) { + return null + } + const mutationEntryCount = [ + children, + removeChildren, + aliases, + removeAliases, + facts, + removeFacts, + tombstones + ].reduce((sum, items) => sum + (items?.length ?? 0), 0) + if (mutationEntryCount > AGENT_STATUS_STORE_LIMITS.mutationEntries) { + return null + } + return { + ...(parent ? { parent } : {}), + ...(removeParent ? { removeParent } : {}), + ...(children ? { children } : {}), + ...(removeChildren ? { removeChildren } : {}), + ...(aliases ? { aliases } : {}), + ...(removeAliases ? { removeAliases } : {}), + ...(facts ? { facts } : {}), + ...(removeFacts ? { removeFacts } : {}), + ...(tombstones ? { tombstones } : {}) + } +} + +export function parseAgentStatusStoreSnapshot(value: unknown): AgentStatusStoreSnapshot | null { + const keys = [ + 'version', + 'epoch', + 'revision', + 'parents', + 'children', + 'aliases', + 'facts', + 'tombstones' + ] + if ( + !isWithinSerializedLimit(value) || + !isRecord(value) || + !hasOnlyKeys(value, keys) || + value.version !== AGENT_STATUS_STORE_SNAPSHOT_VERSION || + !isAgentStatusStoreEpoch(value.epoch) || + !isRevision(value.revision) + ) { + return null + } + const parents = parseArray( + value.parents, + parseAgentStatusParentRecord, + AGENT_STATUS_STORE_LIMITS.parents + ) + const children = parseArray( + value.children, + parseAgentChildWorkRecord, + AGENT_STATUS_STORE_LIMITS.children + ) + const aliases = parseArray( + value.aliases, + parseAgentChildWorkAliasRecord, + AGENT_STATUS_STORE_LIMITS.aliases + ) + const facts = parseArray(value.facts, parseAgentStatusFactRecord, AGENT_STATUS_STORE_LIMITS.facts) + const tombstones = parseArray( + value.tombstones, + parseAgentStatusTombstoneRecord, + AGENT_STATUS_STORE_LIMITS.tombstones + ) + return parents && children && aliases && facts && tombstones + ? { + version: AGENT_STATUS_STORE_SNAPSHOT_VERSION, + epoch: value.epoch, + revision: value.revision, + parents, + children, + aliases, + facts, + tombstones + } + : null +} diff --git a/src/shared/agent-status-store-contract.ts b/src/shared/agent-status-store-contract.ts new file mode 100644 index 00000000000..0ece08286c2 --- /dev/null +++ b/src/shared/agent-status-store-contract.ts @@ -0,0 +1,60 @@ +import type { + AgentChildWorkAliasInput, + AgentChildWorkAliasRecord +} from './agent-status-child-work-alias' +import type { AgentChildWorkInput, AgentChildWorkRecord } from './agent-status-child-work' +import type { AgentStatusParentInput, AgentStatusParentRecord } from './agent-status-store-parent' +import type { AgentStatusSubject } from './agent-status-subject' + +export const AGENT_STATUS_STORE_SNAPSHOT_VERSION = 1 as const +export const AGENT_STATUS_STORE_LIMITS = { + parents: 2_048, + children: 8_192, + aliases: 16_384, + facts: 16_384, + tombstones: 16_384, + mutationEntries: 2_048, + serializedBytes: 16 * 1024 * 1024 +} as const + +export type AgentStatusFactValue = string | number | boolean | null + +export type AgentStatusFactInput = { + subject: AgentStatusSubject + key: string + value: AgentStatusFactValue +} + +export type AgentStatusFactRecord = AgentStatusFactInput & { revision: number } +export type AgentStatusFactIdentity = Pick +export type AgentStatusTombstoneEntity = 'parent' | 'child' | 'alias' | 'fact' + +export type AgentStatusTombstoneInput = { + entity: AgentStatusTombstoneEntity + key: string +} + +export type AgentStatusTombstoneRecord = AgentStatusTombstoneInput & { revision: number } + +export type AgentStatusStoreMutation = { + parent?: AgentStatusParentInput + removeParent?: AgentStatusSubject + children?: AgentChildWorkInput[] + removeChildren?: string[] + aliases?: AgentChildWorkAliasInput[] + removeAliases?: string[] + facts?: AgentStatusFactInput[] + removeFacts?: AgentStatusFactIdentity[] + tombstones?: AgentStatusTombstoneInput[] +} + +export type AgentStatusStoreSnapshot = { + version: typeof AGENT_STATUS_STORE_SNAPSHOT_VERSION + epoch: string + revision: number + parents: AgentStatusParentRecord[] + children: AgentChildWorkRecord[] + aliases: AgentChildWorkAliasRecord[] + facts: AgentStatusFactRecord[] + tombstones: AgentStatusTombstoneRecord[] +} diff --git a/src/shared/agent-status-store-fact-codec.ts b/src/shared/agent-status-store-fact-codec.ts new file mode 100644 index 00000000000..09b5d76a236 --- /dev/null +++ b/src/shared/agent-status-store-fact-codec.ts @@ -0,0 +1,119 @@ +import type { + AgentStatusFactIdentity, + AgentStatusFactInput, + AgentStatusFactRecord, + AgentStatusFactValue +} from './agent-status-store-contract' +import { + deserializeAgentStatusSubject, + parseAgentStatusSubject, + serializeAgentStatusSubject +} from './agent-status-subject' + +const MAX_FACT_KEY_LENGTH = 256 +const MAX_FACT_STRING_LENGTH = 4_096 +const FACT_KEY_PREFIX = 'agent-status-fact-v1:' + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasExactKeys(record: Record, expected: readonly string[]): boolean { + const keys = Object.keys(record) + return keys.length === expected.length && keys.every((key) => expected.includes(key)) +} + +function isRevision(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +function isFactKey(value: unknown): value is string { + if ( + typeof value !== 'string' || + value.length === 0 || + value.length > MAX_FACT_KEY_LENGTH || + value !== value.trim() + ) { + return false + } + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index) + if (code <= 0x1f || code === 0x7f) { + return false + } + } + return true +} + +function parseFactValue(value: unknown): AgentStatusFactValue | undefined { + if (value === null || typeof value === 'boolean') { + return value + } + if (typeof value === 'string') { + return value.length <= MAX_FACT_STRING_LENGTH ? value : undefined + } + return typeof value === 'number' && Number.isFinite(value) ? value : undefined +} + +export function parseAgentStatusFactInput(value: unknown): AgentStatusFactInput | null { + if (!isRecord(value) || !hasExactKeys(value, ['subject', 'key', 'value'])) { + return null + } + const subject = parseAgentStatusSubject(value.subject) + const factValue = parseFactValue(value.value) + if (!subject || !isFactKey(value.key) || factValue === undefined) { + return null + } + return { subject, key: value.key, value: factValue } +} + +export function parseAgentStatusFactRecord(value: unknown): AgentStatusFactRecord | null { + if (!isRecord(value) || !isRevision(value.revision)) { + return null + } + const input = { ...value } + delete input.revision + const fact = parseAgentStatusFactInput(input) + return fact ? { ...fact, revision: value.revision } : null +} + +export function parseAgentStatusFactIdentity(value: unknown): AgentStatusFactIdentity | null { + if (!isRecord(value) || !hasExactKeys(value, ['subject', 'key'])) { + return null + } + const subject = parseAgentStatusSubject(value.subject) + return subject && isFactKey(value.key) ? { subject, key: value.key } : null +} + +export function serializeAgentStatusFactKey(fact: AgentStatusFactIdentity): string { + const parsed = parseAgentStatusFactIdentity({ subject: fact.subject, key: fact.key }) + if (!parsed) { + throw new Error('Invalid agent status fact identity') + } + return `${FACT_KEY_PREFIX}${JSON.stringify([ + serializeAgentStatusSubject(parsed.subject), + parsed.key + ])}` +} + +export function deserializeAgentStatusFactKey(value: string): AgentStatusFactIdentity | null { + if (!value.startsWith(FACT_KEY_PREFIX)) { + return null + } + let tuple: unknown + try { + tuple = JSON.parse(value.slice(FACT_KEY_PREFIX.length)) + } catch { + return null + } + if (!Array.isArray(tuple) || tuple.length !== 2) { + return null + } + const [subjectKey, key] = tuple + if (typeof subjectKey !== 'string') { + return null + } + const subject = deserializeAgentStatusSubject(subjectKey) + const parsed = parseAgentStatusFactIdentity({ subject, key }) + return parsed && serializeAgentStatusFactKey(parsed) === value ? parsed : null +} diff --git a/src/shared/agent-status-store-mutation.ts b/src/shared/agent-status-store-mutation.ts new file mode 100644 index 00000000000..fcaafbc4b0b --- /dev/null +++ b/src/shared/agent-status-store-mutation.ts @@ -0,0 +1,226 @@ +import { + deserializeAgentChildWorkAliasKey, + parseAgentChildWorkAliasRecord, + serializeAgentChildWorkAliasKey +} from './agent-status-child-work-alias' +import { parseAgentChildWorkRecord } from './agent-status-child-work-codec' +import type { + AgentStatusStoreMutation, + AgentStatusTombstoneEntity +} from './agent-status-store-contract' +import { parseAgentStatusFactRecord } from './agent-status-store-fact-codec' +import { parseAgentStatusParentRecord } from './agent-status-store-parent' +import { + agentStatusFactMapKey, + agentStatusTombstoneMapKey, + cloneAgentStatusStoreState, + deepFreezeAgentStatusStoreValue, + validateAgentStatusStoreState, + type AgentStatusStoreState +} from './agent-status-store-state' +import { + agentStatusSubjectsEqual, + deserializeAgentStatusSubject, + serializeAgentStatusSubject, + type AgentStatusSubject +} from './agent-status-subject' + +function addTombstone( + state: AgentStatusStoreState, + entity: AgentStatusTombstoneEntity, + key: string, + revision: number +): void { + const record = deepFreezeAgentStatusStoreValue({ entity, key, revision }) + state.tombstones.set(agentStatusTombstoneMapKey(entity, key), record) +} + +function removeAlias(state: AgentStatusStoreState, key: string, revision: number): void { + state.aliases.delete(key) + addTombstone(state, 'alias', key, revision) +} + +function removeFact(state: AgentStatusStoreState, key: string, revision: number): void { + state.facts.delete(key) + addTombstone(state, 'fact', key, revision) +} + +function removeChild(state: AgentStatusStoreState, childWorkId: string, revision: number): void { + state.children.delete(childWorkId) + addTombstone(state, 'child', childWorkId, revision) + for (const [key, alias] of state.aliases) { + if (alias.childWorkId === childWorkId) { + removeAlias(state, key, revision) + } + } +} + +function removeParent( + state: AgentStatusStoreState, + subject: AgentStatusSubject, + revision: number +): void { + const key = serializeAgentStatusSubject(subject) + state.parents.delete(key) + addTombstone(state, 'parent', key, revision) + for (const child of state.children.values()) { + if (agentStatusSubjectsEqual(child.parent, subject)) { + removeChild(state, child.childWorkId, revision) + } + } + for (const [factMapKey, fact] of state.facts) { + if (agentStatusSubjectsEqual(fact.subject, subject)) { + removeFact(state, factMapKey, revision) + } + } +} + +function applyExplicitTombstone( + state: AgentStatusStoreState, + tombstone: { entity: AgentStatusTombstoneEntity; key: string }, + revision: number +): boolean { + if (tombstone.entity === 'parent') { + const subject = deserializeAgentStatusSubject(tombstone.key) + if (!subject) { + return false + } + removeParent(state, subject, revision) + return true + } + if (tombstone.entity === 'child') { + removeChild(state, tombstone.key, revision) + } else if (tombstone.entity === 'alias') { + if (!deserializeAgentChildWorkAliasKey(tombstone.key)) { + return false + } + removeAlias(state, tombstone.key, revision) + } else { + removeFact(state, tombstone.key, revision) + } + addTombstone(state, tombstone.entity, tombstone.key, revision) + return true +} + +function upsertParent( + state: AgentStatusStoreState, + input: NonNullable, + revision: number +): boolean { + const key = serializeAgentStatusSubject(input.subject) + if (state.tombstones.has(agentStatusTombstoneMapKey('parent', key))) { + return false + } + const previous = state.parents.get(key) + const record = parseAgentStatusParentRecord({ + ...input, + ...(input.firstObservedAt === undefined && previous?.firstObservedAt !== undefined + ? { firstObservedAt: previous.firstObservedAt } + : {}), + revision + }) + if (!record) { + return false + } + state.parents.set(key, deepFreezeAgentStatusStoreValue(record)) + return true +} + +function upsertChildren( + state: AgentStatusStoreState, + children: NonNullable, + revision: number +): boolean { + for (const input of children) { + const previous = state.children.get(input.childWorkId) + if ( + state.tombstones.has(agentStatusTombstoneMapKey('child', input.childWorkId)) || + !state.parents.has(serializeAgentStatusSubject(input.parent)) || + (previous !== undefined && previous.firstObservedAt !== input.firstObservedAt) || + (previous !== undefined && input.observedAt < previous.observedAt) + ) { + return false + } + const record = parseAgentChildWorkRecord({ ...input, revision }) + if (!record) { + return false + } + state.children.set(input.childWorkId, deepFreezeAgentStatusStoreValue(record)) + } + return true +} + +function upsertAliases( + state: AgentStatusStoreState, + aliases: NonNullable, + revision: number +): boolean { + for (const input of aliases) { + const record = parseAgentChildWorkAliasRecord({ ...input, revision }) + if (!record) { + return false + } + state.aliases.set( + serializeAgentChildWorkAliasKey(record), + deepFreezeAgentStatusStoreValue(record) + ) + } + return true +} + +function upsertFacts( + state: AgentStatusStoreState, + facts: NonNullable, + revision: number +): boolean { + for (const input of facts) { + const record = parseAgentStatusFactRecord({ ...input, revision }) + if (!record || !state.parents.has(serializeAgentStatusSubject(record.subject))) { + return false + } + state.facts.set(agentStatusFactMapKey(record), deepFreezeAgentStatusStoreValue(record)) + } + return true +} + +export function applyAgentStatusStoreMutation( + current: AgentStatusStoreState, + mutation: AgentStatusStoreMutation, + revision: number +): AgentStatusStoreState | null { + const next = cloneAgentStatusStoreState(current) + next.revision = revision + if (mutation.removeParent) { + removeParent(next, mutation.removeParent, revision) + } + for (const childWorkId of mutation.removeChildren ?? []) { + removeChild(next, childWorkId, revision) + } + for (const key of mutation.removeAliases ?? []) { + if (!deserializeAgentChildWorkAliasKey(key)) { + return null + } + removeAlias(next, key, revision) + } + for (const identity of mutation.removeFacts ?? []) { + removeFact(next, agentStatusFactMapKey(identity), revision) + } + for (const tombstone of mutation.tombstones ?? []) { + if (!applyExplicitTombstone(next, tombstone, revision)) { + return null + } + } + if (mutation.parent && !upsertParent(next, mutation.parent, revision)) { + return null + } + if (mutation.children && !upsertChildren(next, mutation.children, revision)) { + return null + } + if (mutation.aliases && !upsertAliases(next, mutation.aliases, revision)) { + return null + } + if (mutation.facts && !upsertFacts(next, mutation.facts, revision)) { + return null + } + return validateAgentStatusStoreState(next) ? next : null +} diff --git a/src/shared/agent-status-store-parent.ts b/src/shared/agent-status-store-parent.ts new file mode 100644 index 00000000000..12682becb9d --- /dev/null +++ b/src/shared/agent-status-store-parent.ts @@ -0,0 +1,117 @@ +import type { AgentStatusIpcPayload } from './agent-status-ipc-payload' +import { parseAgentStatusPtyRunRecord, type AgentStatusPtyRunRecord } from './agent-status-run' +import { parseAgentStatusIpcPayloadCopy } from './agent-status-store-status-codec' +import { parseAgentStatusSubject, type AgentStatusSubject } from './agent-status-subject' + +export type AgentStatusParentInput = { + subject: AgentStatusSubject + status?: AgentStatusIpcPayload + run?: AgentStatusPtyRunRecord + firstObservedAt?: number +} + +export type AgentStatusParentRecord = AgentStatusParentInput & { revision: number } + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasOnlyKeys( + record: Record, + required: readonly string[], + optional: readonly string[] = [] +): boolean { + const keys = Object.keys(record) + return ( + required.every((key) => Object.hasOwn(record, key)) && + keys.every((key) => required.includes(key) || optional.includes(key)) + ) +} + +function isTimestamp(value: unknown): value is number { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 +} + +function isRevision(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +function isParentScopeConsistent(parent: AgentStatusParentInput): boolean { + const { subject, status, run } = parent + if (run && (subject.kind !== 'pty-run' || run.runId !== subject.runId)) { + return false + } + if ( + subject.kind !== 'pty-run' && + (status?.runId !== undefined || status?.executionId !== undefined) + ) { + return false + } + if (status?.worktreeId !== undefined && status.worktreeId !== subject.workspaceId) { + return false + } + if (subject.kind === 'pty' && status?.paneKey !== subject.paneKey) { + return false + } + if (subject.kind === 'pty-run' && status?.runId !== undefined && status.runId !== subject.runId) { + return false + } + if (run && status?.paneKey !== undefined && status.paneKey !== run.paneKey) { + return false + } + if ( + run && + status?.executionId !== undefined && + status.executionId !== run.attachment.executionId + ) { + return false + } + if ( + run && + status?.providerAlias && + run.providerSessions.length > 0 && + !run.providerSessions.some( + (session) => + session.provider === status.providerAlias?.provider && + session.sessionKeyKind === status.providerAlias.sessionKeyKind && + session.providerId === status.providerAlias.providerId + ) + ) { + return false + } + return true +} + +export function parseAgentStatusParentInput(value: unknown): AgentStatusParentInput | null { + if ( + !isRecord(value) || + !hasOnlyKeys(value, ['subject'], ['status', 'run', 'firstObservedAt']) || + (value.firstObservedAt !== undefined && !isTimestamp(value.firstObservedAt)) + ) { + return null + } + const subject = parseAgentStatusSubject(value.subject) + const status = + value.status === undefined ? undefined : parseAgentStatusIpcPayloadCopy(value.status) + const run = value.run === undefined ? undefined : parseAgentStatusPtyRunRecord(value.run) + if (!subject || (value.status !== undefined && !status) || (value.run !== undefined && !run)) { + return null + } + const parent: AgentStatusParentInput = { + subject, + ...(status ? { status } : {}), + ...(run ? { run } : {}), + ...(isTimestamp(value.firstObservedAt) ? { firstObservedAt: value.firstObservedAt } : {}) + } + return isParentScopeConsistent(parent) ? parent : null +} + +export function parseAgentStatusParentRecord(value: unknown): AgentStatusParentRecord | null { + if (!isRecord(value) || !isRevision(value.revision)) { + return null + } + const input = { ...value } + delete input.revision + const parent = parseAgentStatusParentInput(input) + return parent ? { ...parent, revision: value.revision } : null +} diff --git a/src/shared/agent-status-store-persistence.ts b/src/shared/agent-status-store-persistence.ts new file mode 100644 index 00000000000..0f72203ca3e --- /dev/null +++ b/src/shared/agent-status-store-persistence.ts @@ -0,0 +1,45 @@ +import { + AGENT_STATUS_STORE_LIMITS, + type AgentStatusStoreSnapshot +} from './agent-status-store-contract' +import { parseAgentStatusStoreSnapshot } from './agent-status-store-codec' +import { assertJsonTextStructureWithinLimits } from './json-text-structure-limit' +import { measureUtf8ByteLength } from './utf8-byte-limits' + +const SNAPSHOT_STRUCTURE_LIMITS = { + structuralTokens: 512 * 1024, + nestingDepth: 32 +} as const + +function isWithinByteLimit(value: string): boolean { + return !measureUtf8ByteLength(value, { + stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes + }).exceededLimit +} + +/** Bounded persistence form; callers write the returned string with their existing owner. */ +export function serializeAgentStatusStoreSnapshot(snapshot: AgentStatusStoreSnapshot): string { + const parsed = parseAgentStatusStoreSnapshot(snapshot) + if (!parsed) { + throw new Error('Invalid agent status store snapshot') + } + const serialized = JSON.stringify(parsed) + if (!isWithinByteLimit(serialized)) { + throw new Error('Agent status store snapshot exceeds its serialized-byte limit') + } + return serialized +} + +export function deserializeAgentStatusStoreSnapshot( + serialized: string +): AgentStatusStoreSnapshot | null { + if (!isWithinByteLimit(serialized)) { + return null + } + try { + assertJsonTextStructureWithinLimits(serialized, SNAPSHOT_STRUCTURE_LIMITS) + return parseAgentStatusStoreSnapshot(JSON.parse(serialized)) + } catch { + return null + } +} diff --git a/src/shared/agent-status-store-state.ts b/src/shared/agent-status-store-state.ts new file mode 100644 index 00000000000..1ffd1cceb3f --- /dev/null +++ b/src/shared/agent-status-store-state.ts @@ -0,0 +1,249 @@ +import { + deserializeAgentChildWorkAliasKey, + parseAgentChildWorkAliasRecord, + serializeAgentChildWorkAliasKey, + type AgentChildWorkAliasRecord +} from './agent-status-child-work-alias' +import { + agentChildWorkBelongsTo, + agentChildWorkFencesEqual, + type AgentChildWorkRecord +} from './agent-status-child-work' +import { parseAgentChildWorkRecord } from './agent-status-child-work-codec' +import { + AGENT_STATUS_STORE_LIMITS, + AGENT_STATUS_STORE_SNAPSHOT_VERSION, + type AgentStatusFactIdentity, + type AgentStatusFactRecord, + type AgentStatusStoreSnapshot, + type AgentStatusTombstoneEntity, + type AgentStatusTombstoneRecord +} from './agent-status-store-contract' +import { + parseAgentStatusStoreSnapshot, + parseAgentStatusTombstoneRecord +} from './agent-status-store-codec' +import { + deserializeAgentStatusFactKey, + parseAgentStatusFactRecord, + serializeAgentStatusFactKey +} from './agent-status-store-fact-codec' +import { + parseAgentStatusParentRecord, + type AgentStatusParentRecord +} from './agent-status-store-parent' +import { deserializeAgentStatusSubject, serializeAgentStatusSubject } from './agent-status-subject' + +export type AgentStatusStoreState = { + epoch: string + revision: number + parents: Map + children: Map + aliases: Map + facts: Map + tombstones: Map +} + +export function deepFreezeAgentStatusStoreValue(value: T): T { + if (typeof value !== 'object' || value === null || Object.isFrozen(value)) { + return value + } + for (const nested of Object.values(value)) { + deepFreezeAgentStatusStoreValue(nested) + } + return Object.freeze(value) +} + +export function agentStatusFactMapKey(fact: AgentStatusFactIdentity): string { + return serializeAgentStatusFactKey(fact) +} + +export function agentStatusTombstoneMapKey( + entity: AgentStatusTombstoneEntity, + key: string +): string { + return `${entity}\0${key}` +} + +export function createEmptyAgentStatusStoreState(epoch: string): AgentStatusStoreState { + return { + epoch, + revision: 0, + parents: new Map(), + children: new Map(), + aliases: new Map(), + facts: new Map(), + tombstones: new Map() + } +} + +export function cloneAgentStatusStoreState(state: AgentStatusStoreState): AgentStatusStoreState { + return { + epoch: state.epoch, + revision: state.revision, + parents: new Map(state.parents), + children: new Map(state.children), + aliases: new Map(state.aliases), + facts: new Map(state.facts), + tombstones: new Map(state.tombstones) + } +} + +function hasMatchingFence(child: AgentChildWorkRecord, alias: AgentChildWorkAliasRecord): boolean { + if (agentChildWorkFencesEqual(child.invocation, alias.fence)) { + return true + } + return ( + child.previousInvocations?.some((entry) => + agentChildWorkFencesEqual(entry.fence, alias.fence) + ) === true + ) +} + +export function validateAgentStatusStoreState(state: AgentStatusStoreState): boolean { + if ( + state.parents.size > AGENT_STATUS_STORE_LIMITS.parents || + state.children.size > AGENT_STATUS_STORE_LIMITS.children || + state.aliases.size > AGENT_STATUS_STORE_LIMITS.aliases || + state.facts.size > AGENT_STATUS_STORE_LIMITS.facts || + state.tombstones.size > AGENT_STATUS_STORE_LIMITS.tombstones + ) { + return false + } + for (const [key, parent] of state.parents) { + if ( + key !== serializeAgentStatusSubject(parent.subject) || + parent.revision > state.revision || + state.tombstones.has(agentStatusTombstoneMapKey('parent', key)) + ) { + return false + } + } + for (const [childWorkId, child] of state.children) { + if ( + childWorkId !== child.childWorkId || + child.revision > state.revision || + !state.parents.has(serializeAgentStatusSubject(child.parent)) || + state.tombstones.has(agentStatusTombstoneMapKey('child', childWorkId)) + ) { + return false + } + } + for (const [key, alias] of state.aliases) { + const child = state.children.get(alias.childWorkId) + const tombstone = state.tombstones.get(agentStatusTombstoneMapKey('alias', key)) + if ( + key !== serializeAgentChildWorkAliasKey(alias) || + alias.revision > state.revision || + !child || + !agentChildWorkBelongsTo(child, alias.parent) || + child.provider !== alias.provider || + child.kind !== alias.kind || + !hasMatchingFence(child, alias) || + (tombstone !== undefined && tombstone.revision >= alias.revision) + ) { + return false + } + } + for (const [key, fact] of state.facts) { + const tombstone = state.tombstones.get(agentStatusTombstoneMapKey('fact', key)) + if ( + key !== agentStatusFactMapKey(fact) || + fact.revision > state.revision || + !state.parents.has(serializeAgentStatusSubject(fact.subject)) || + (tombstone !== undefined && tombstone.revision >= fact.revision) + ) { + return false + } + } + for (const item of state.tombstones.values()) { + if ( + item.revision > state.revision || + (item.entity === 'parent' && !deserializeAgentStatusSubject(item.key)) || + (item.entity === 'alias' && !deserializeAgentChildWorkAliasKey(item.key)) || + (item.entity === 'fact' && !deserializeAgentStatusFactKey(item.key)) + ) { + return false + } + } + return true +} + +export function snapshotFromAgentStatusStoreState( + state: AgentStatusStoreState +): AgentStatusStoreSnapshot { + const snapshot = parseAgentStatusStoreSnapshot({ + version: AGENT_STATUS_STORE_SNAPSHOT_VERSION, + epoch: state.epoch, + revision: state.revision, + parents: [...state.parents.values()], + children: [...state.children.values()], + aliases: [...state.aliases.values()], + facts: [...state.facts.values()], + tombstones: [...state.tombstones.values()] + }) + if (!snapshot) { + throw new Error('Agent status store produced an invalid snapshot') + } + return deepFreezeAgentStatusStoreValue(snapshot) +} + +export function agentStatusStoreStateFromSnapshot( + snapshot: AgentStatusStoreSnapshot, + epoch: string +): AgentStatusStoreState | null { + const state = createEmptyAgentStatusStoreState(epoch) + state.revision = snapshot.revision + for (const parent of snapshot.parents) { + const record = parseAgentStatusParentRecord(parent) + if (!record) { + return null + } + const key = serializeAgentStatusSubject(record.subject) + if (state.parents.has(key)) { + return null + } + state.parents.set(key, deepFreezeAgentStatusStoreValue(record)) + } + for (const child of snapshot.children) { + const record = parseAgentChildWorkRecord(child) + if (!record || state.children.has(record.childWorkId)) { + return null + } + state.children.set(record.childWorkId, deepFreezeAgentStatusStoreValue(record)) + } + for (const alias of snapshot.aliases) { + const record = parseAgentChildWorkAliasRecord(alias) + if (!record) { + return null + } + const key = serializeAgentChildWorkAliasKey(record) + if (state.aliases.has(key)) { + return null + } + state.aliases.set(key, deepFreezeAgentStatusStoreValue(record)) + } + for (const fact of snapshot.facts) { + const record = parseAgentStatusFactRecord(fact) + if (!record) { + return null + } + const key = agentStatusFactMapKey(record) + if (state.facts.has(key)) { + return null + } + state.facts.set(key, deepFreezeAgentStatusStoreValue(record)) + } + for (const tombstone of snapshot.tombstones) { + const record = parseAgentStatusTombstoneRecord(tombstone) + if (!record) { + return null + } + const key = agentStatusTombstoneMapKey(record.entity, record.key) + if (state.tombstones.has(key)) { + return null + } + state.tombstones.set(key, deepFreezeAgentStatusStoreValue(record)) + } + return validateAgentStatusStoreState(state) ? state : null +} diff --git a/src/shared/agent-status-store-status-codec.ts b/src/shared/agent-status-store-status-codec.ts new file mode 100644 index 00000000000..21f5bcb135e --- /dev/null +++ b/src/shared/agent-status-store-status-codec.ts @@ -0,0 +1,245 @@ +import { normalizeAgentProviderSession } from './agent-session-resume' +import type { AgentStatusIpcPayload } from './agent-status-ipc-payload' +import { + isAgentStatusExecutionId, + isAgentStatusRunId, + parseAgentStatusProviderAlias +} from './agent-status-run' +import { normalizeAgentStatusPayload } from './agent-status-types' +import { assertJsonTextStructureWithinLimits } from './json-text-structure-limit' +import { measureUtf8ByteLength } from './utf8-byte-limits' + +const MAX_STATUS_BYTES = 256 * 1024 +const MAX_ID_LENGTH = 4_096 +const STATUS_STRUCTURE_LIMITS = { structuralTokens: 16_384, nestingDepth: 24 } as const + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasOnlyKeys( + record: Record, + required: readonly string[], + optional: readonly string[] = [] +): boolean { + const keys = Object.keys(record) + return ( + required.every((key) => Object.hasOwn(record, key)) && + keys.every((key) => required.includes(key) || optional.includes(key)) + ) +} + +function isTimestamp(value: unknown): value is number { + return typeof value === 'number' && Number.isFinite(value) && value >= 0 +} + +function isRevision(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +function isBoundedString(value: unknown): value is string { + return ( + typeof value === 'string' && + value.length > 0 && + value.length <= MAX_ID_LENGTH && + !value.includes('\0') + ) +} + +function copyJsonRecord(value: unknown): Record | null { + let serialized: string + try { + serialized = JSON.stringify(value) + } catch { + return null + } + if (measureUtf8ByteLength(serialized, { stopAfterBytes: MAX_STATUS_BYTES }).exceededLimit) { + return null + } + try { + assertJsonTextStructureWithinLimits(serialized, STATUS_STRUCTURE_LIMITS) + const parsed: unknown = JSON.parse(serialized) + return isRecord(parsed) ? parsed : null + } catch { + return null + } +} + +function parseObservation(value: unknown): Record | null { + if ( + !isRecord(value) || + !hasOnlyKeys( + value, + ['origin', 'authorityId', 'incarnation', 'revision', 'observedAt'], + ['boundary', 'kind'] + ) || + (value.origin !== 'hook' && + value.origin !== 'osc' && + value.origin !== 'title' && + value.origin !== 'process' && + value.origin !== 'launch' && + value.origin !== 'orchestration' && + value.origin !== 'structured') || + !isBoundedString(value.authorityId) || + !isRevision(value.incarnation) || + !isRevision(value.revision) || + !isTimestamp(value.observedAt) || + (value.boundary !== undefined && value.boundary !== true) || + (value.kind !== undefined && + value.kind !== 'transition' && + value.kind !== 'snapshot' && + value.kind !== 'identity-only') + ) { + return null + } + return { + origin: value.origin, + authorityId: value.authorityId, + incarnation: value.incarnation, + revision: value.revision, + observedAt: value.observedAt, + ...(value.boundary === true ? { boundary: true } : {}), + ...(value.kind !== undefined ? { kind: value.kind } : {}) + } +} + +function parseOrchestration(value: unknown): Record | null { + if (!isRecord(value) || !isBoundedString(value.taskId) || !isBoundedString(value.dispatchId)) { + return null + } + const optionalStrings = [ + 'taskTitle', + 'displayName', + 'parentTerminalHandle', + 'parentPaneKey', + 'coordinatorHandle', + 'orchestrationRunId' + ] + if (optionalStrings.some((key) => value[key] !== undefined && !isBoundedString(value[key]))) { + return null + } + if ( + value.dispatchStatus !== undefined && + value.dispatchStatus !== 'pending' && + value.dispatchStatus !== 'dispatched' && + value.dispatchStatus !== 'completed' && + value.dispatchStatus !== 'failed' && + value.dispatchStatus !== 'circuit_broken' + ) { + return null + } + if (value.attention !== undefined && !isRecord(value.attention)) { + return null + } + return { ...value } +} + +function copyOptionalString( + source: Record, + target: Record, + key: string +): boolean { + const value = source[key] + if (value === undefined) { + return true + } + if (!isBoundedString(value)) { + return false + } + target[key] = value + return true +} + +export function parseAgentStatusIpcPayloadCopy(value: unknown): AgentStatusIpcPayload | null { + const copied = copyJsonRecord(value) + const payload = normalizeAgentStatusPayload(copied) + if ( + !copied || + !payload || + !isBoundedString(copied.paneKey) || + (copied.connectionId !== null && !isBoundedString(copied.connectionId)) || + !isTimestamp(copied.receivedAt) || + !isTimestamp(copied.stateStartedAt) || + (copied.evidenceObservedAt !== undefined && !isTimestamp(copied.evidenceObservedAt)) + ) { + return null + } + const parsed: Record = { + ...payload, + paneKey: copied.paneKey, + connectionId: copied.connectionId, + receivedAt: copied.receivedAt, + stateStartedAt: copied.stateStartedAt + } + for (const key of [ + 'launchToken', + 'terminalHandle', + 'tabId', + 'worktreeId', + 'promptInteractionKey' + ]) { + if (!copyOptionalString(copied, parsed, key)) { + return null + } + } + if (copied.runId !== undefined) { + if (!isAgentStatusRunId(copied.runId)) { + return null + } + parsed.runId = copied.runId + } + if (copied.executionId !== undefined) { + if (!isAgentStatusExecutionId(copied.executionId)) { + return null + } + parsed.executionId = copied.executionId + } + if (copied.providerAlias !== undefined) { + const providerAlias = parseAgentStatusProviderAlias(copied.providerAlias) + if (!providerAlias) { + return null + } + parsed.providerAlias = providerAlias + } + if (isTimestamp(copied.evidenceObservedAt)) { + parsed.evidenceObservedAt = copied.evidenceObservedAt + } + if (copied.providerSession !== undefined) { + const providerSession = normalizeAgentProviderSession(copied.providerSession) + if (!providerSession) { + return null + } + parsed.providerSession = providerSession + } + if (copied.orchestration !== undefined) { + const orchestration = parseOrchestration(copied.orchestration) + if (!orchestration) { + return null + } + parsed.orchestration = orchestration + } + if (copied.observation !== undefined) { + const observation = parseObservation(copied.observation) + if (!observation) { + return null + } + parsed.observation = observation + } + if (copied.providerSessionOnly === true) { + parsed.providerSessionOnly = true + } else if (copied.providerSessionOnly !== undefined && copied.providerSessionOnly !== false) { + return null + } + if (copied.restoredUnconfirmed === true) { + parsed.restoredUnconfirmed = true + } else if (copied.restoredUnconfirmed !== undefined && copied.restoredUnconfirmed !== false) { + return null + } + if (copied.structuredHost === 'held' || copied.structuredHost === 'owned') { + parsed.structuredHost = copied.structuredHost + } else if (copied.structuredHost !== undefined) { + return null + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Every required and optional field is rebuilt from its canonical parser above. + return parsed as AgentStatusIpcPayload +} diff --git a/src/shared/agent-status-store.test.ts b/src/shared/agent-status-store.test.ts new file mode 100644 index 00000000000..dc7d78f931d --- /dev/null +++ b/src/shared/agent-status-store.test.ts @@ -0,0 +1,321 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AgentStatusIpcPayload } from './agent-status-ipc-payload' +import type { AgentChildWorkAliasInput } from './agent-status-child-work-alias' +import type { AgentChildWorkInput } from './agent-status-child-work' +import { createAgentStatusStore } from './agent-status-store' +import { + AGENT_STATUS_STORE_LIMITS, + type AgentStatusStoreSnapshot +} from './agent-status-store-contract' +import { + deserializeAgentStatusStoreSnapshot, + serializeAgentStatusStoreSnapshot +} from './agent-status-store-persistence' +import { + makePtyAgentStatusSubject, + makePtyRunAgentStatusSubject, + makeStructuredAgentStatusSubject, + serializeAgentStatusSubject, + type AgentStatusExecutionScope, + type AgentStatusSubject +} from './agent-status-subject' + +const SESSION_ID = 'session_11111111-1111-4111-8111-111111111111' + +function scope(overrides: Partial = {}): AgentStatusExecutionScope { + return { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree', + ...overrides + } +} + +function subject(overrides: Partial = {}): AgentStatusSubject { + return makeStructuredAgentStatusSubject(scope(overrides), SESSION_ID) +} + +function status(parent: AgentStatusSubject, overrides: Partial = {}) { + return { + state: 'working', + prompt: 'Ship it', + paneKey: 'structured-pane-key', + connectionId: null, + receivedAt: 20, + evidenceObservedAt: 18, + stateStartedAt: 10, + worktreeId: parent.workspaceId, + structuredHost: 'owned', + ...overrides + } satisfies AgentStatusIpcPayload +} + +function child(parent: AgentStatusSubject, overrides: Partial = {}) { + return { + childWorkId: 'child-1', + parent, + provider: 'claude', + kind: 'agent', + state: 'working', + membership: 'live', + firstObservedAt: 12, + observedAt: 20, + stoppable: true, + invocation: { invocationId: 'invocation-1', generation: 1 }, + provenance: { source: 'structured-session', producerId: 'journal-1' }, + ...overrides + } satisfies AgentChildWorkInput +} + +function alias(parent: AgentStatusSubject): AgentChildWorkAliasInput { + return { + parent, + provider: 'claude', + segmentId: 'segment-1', + kind: 'agent', + aliasKind: 'task_id', + alias: 'provider-task-1', + childWorkId: 'child-1', + fence: { invocationId: 'invocation-1', generation: 1 } + } +} + +function populatedStore() { + const parent = subject() + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const committed = store.applyMutation({ + parent: { subject: parent, status: status(parent), firstObservedAt: 5 }, + children: [child(parent)], + aliases: [alias(parent)], + facts: [{ subject: parent, key: 'acknowledged', value: true }] + }) + expect(committed?.revision).toBe(1) + return { parent, store } +} + +describe('AgentStatusStore', () => { + it('stores structured status on the parent record and isolates colliding scoped subjects', () => { + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const subjects = [ + subject(), + subject({ wslDistro: 'Ubuntu' }), + subject({ executionHostId: 'ssh:host-a' }), + subject({ executionHostId: 'runtime:peer-a' }), + subject({ workspaceId: 'folder-1', workspaceKind: 'folder' }) + ] + + for (const [index, scopedSubject] of subjects.entries()) { + expect( + store.applyMutation({ + parent: { + subject: scopedSubject, + status: status(scopedSubject, { prompt: `prompt-${index}` }), + firstObservedAt: index + 1 + } + }) + ).not.toBeNull() + } + + expect(store.getSnapshot().parents).toHaveLength(subjects.length) + expect(subjects.map((item) => store.getParent(item)?.status?.prompt)).toEqual([ + 'prompt-0', + 'prompt-1', + 'prompt-2', + 'prompt-3', + 'prompt-4' + ]) + }) + + it('accepts trusted PTY fixtures while enforcing run and scope consistency', () => { + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const pty = makePtyAgentStatusSubject(scope(), 'pane-1') + const runSubject = makePtyRunAgentStatusSubject(scope(), 'run-1') + + expect( + store.applyMutation({ + parent: { subject: pty, status: status(pty, { paneKey: 'pane-1' }) } + }) + ).not.toBeNull() + expect( + store.applyMutation({ + parent: { subject: pty, status: status(pty, { paneKey: 'wrong-pane' }) } + }) + ).toBeNull() + const run = { + runId: 'run-1', + paneKey: 'pane-2', + attachment: { executionId: 'execution-1' }, + attribution: 'token', + providerSessions: [ + { provider: 'claude', sessionKeyKind: 'session_id', providerId: 'provider-1' } + ], + role: 'root', + verdict: 'live' + } as const + expect( + store.applyMutation({ + parent: { + subject: runSubject, + run, + status: status(runSubject, { + paneKey: 'pane-2', + runId: 'run-1', + executionId: 'execution-1', + providerAlias: { + provider: 'claude', + sessionKeyKind: 'session_id', + providerId: 'provider-1' + } + }) + } + }) + ).not.toBeNull() + expect(store.getParent(runSubject)?.run).toEqual(run) + expect( + store.applyMutation({ + parent: { + subject: runSubject, + status: status(runSubject, { paneKey: 'pane-2', runId: 'wrong-run' }) + } + }) + ).toBeNull() + }) + + it('uses parsed immutable copies instead of caller-owned objects', () => { + const parent = subject() + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const inputStatus = status(parent, { + subagents: [{ id: 'provider-child', state: 'working', startedAt: 1, description: 'before' }] + }) + expect(store.applyMutation({ parent: { subject: parent, status: inputStatus } })).not.toBeNull() + + inputStatus.prompt = 'mutated input' + inputStatus.subagents?.splice(0) + const first = store.getParent(parent) + expect(first?.status?.prompt).toBe('Ship it') + expect(first?.status?.subagents).toHaveLength(1) + expect(Object.isFrozen(first)).toBe(true) + expect(Object.isFrozen(first?.status?.subagents)).toBe(true) + expect(() => first?.status?.subagents?.splice(0)).toThrow() + expect(store.getParent(parent)?.status?.subagents).toHaveLength(1) + }) + + it('commits parent, child, alias, fact and tombstone state atomically', () => { + const { parent, store } = populatedStore() + const before = store.getSnapshot() + const invalidAlias = { ...alias(parent), provider: 'codex' } + + expect( + store.applyMutation({ + children: [child(parent, { state: 'waiting', observedAt: 21 })], + aliases: [invalidAlias], + facts: [{ subject: parent, key: 'unread', value: true }] + }) + ).toBeNull() + expect(store.getSnapshot()).toEqual(before) + + const removal = store.applyMutation({ removeParent: parent }) + const removed = store.getSnapshot() + expect(removal?.revision).toBe(2) + expect(removed.parents).toEqual([]) + expect(removed.children).toEqual([]) + expect(removed.aliases).toEqual([]) + expect(removed.facts).toEqual([]) + expect(new Set(removed.tombstones.map((item) => item.entity))).toEqual( + new Set(['parent', 'child', 'alias', 'fact']) + ) + expect(new Set(removed.tombstones.map((item) => item.revision))).toEqual(new Set([2])) + }) + + it('never resurrects an exactly removed parent or child id', () => { + const { parent, store } = populatedStore() + expect(store.applyMutation({ removeChildren: ['child-1'] })).not.toBeNull() + expect(store.applyMutation({ children: [child(parent, { observedAt: 30 })] })).toBeNull() + expect(store.applyMutation({ removeParent: parent })).not.toBeNull() + expect(store.applyMutation({ parent: { subject: parent, status: status(parent) } })).toBeNull() + }) + + it('persists a bounded snapshot and restores child identity under a new epoch', () => { + const { parent, store } = populatedStore() + expect( + store.applyMutation({ removeFacts: [{ subject: parent, key: 'acknowledged' }] }) + ).not.toBeNull() + expect( + store.applyMutation({ facts: [{ subject: parent, key: 'acknowledged', value: true }] }) + ).not.toBeNull() + const serialized = serializeAgentStatusStoreSnapshot(store.getSnapshot()) + const persisted = deserializeAgentStatusStoreSnapshot(serialized) + const restarted = createAgentStatusStore({ epoch: 'epoch-b', mode: 'authority' }) + + expect(persisted).not.toBeNull() + expect(restarted.applySnapshot(persisted)).toBe(true) + expect(restarted.getSnapshot().epoch).toBe('epoch-b') + expect(restarted.getSnapshot().revision).toBe(3) + expect(restarted.getChildren(parent)[0]?.childWorkId).toBe('child-1') + expect(restarted.getChildren(parent)[0]?.firstObservedAt).toBe(12) + }) + + it('fails closed before parsing oversized persistence payloads or allocating huge mutations', () => { + const parse = vi.spyOn(JSON, 'parse') + const oversized = ' '.repeat(AGENT_STATUS_STORE_LIMITS.serializedBytes + 1) + + expect(deserializeAgentStatusStoreSnapshot(oversized)).toBeNull() + expect(parse).not.toHaveBeenCalled() + parse.mockRestore() + + const store = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + expect( + store.applyMutation({ + removeChildren: Array.from( + { length: AGENT_STATUS_STORE_LIMITS.mutationEntries + 1 }, + (_, index) => `child-${index}` + ) + }) + ).toBeNull() + expect(store.getSnapshot().revision).toBe(0) + const scopedSubject = subject() + expect( + store.applyMutation({ + parent: { + subject: scopedSubject, + status: { + ...status(scopedSubject), + orchestration: { + taskId: 'task-1', + dispatchId: 'dispatch-1', + attention: { oversized: 'x'.repeat(300 * 1024) } + } + } + } + }) + ).toBeNull() + }) + + it('rejects malformed and scope-mismatched snapshots without changing state', () => { + const { store } = populatedStore() + const before = store.getSnapshot() + const malformed: AgentStatusStoreSnapshot = { + ...before, + children: [ + { + ...before.children[0], + parent: subject({ workspaceId: 'other-workspace' }) + } + ] + } + + expect(store.applySnapshot(malformed)).toBe(false) + expect(store.getSnapshot()).toEqual(before) + }) + + it('serializes subject removal keys exactly in tombstones', () => { + const { parent, store } = populatedStore() + expect(store.applyMutation({ removeParent: parent })).not.toBeNull() + expect(store.getSnapshot().tombstones).toContainEqual({ + entity: 'parent', + key: serializeAgentStatusSubject(parent), + revision: 2 + }) + }) +}) diff --git a/src/shared/agent-status-store.ts b/src/shared/agent-status-store.ts new file mode 100644 index 00000000000..4037623734f --- /dev/null +++ b/src/shared/agent-status-store.ts @@ -0,0 +1,158 @@ +import { agentChildWorkBelongsTo, type AgentChildWorkRecord } from './agent-status-child-work' +import { parseAgentChildWorkRecord } from './agent-status-child-work-codec' +import type { AgentStatusStoreSnapshot } from './agent-status-store-contract' +import { + isAgentStatusStoreEpoch, + parseAgentStatusStoreMutation, + parseAgentStatusStoreSnapshot +} from './agent-status-store-codec' +import { applyAgentStatusStoreMutation } from './agent-status-store-mutation' +import { + parseAgentStatusParentRecord, + type AgentStatusParentRecord +} from './agent-status-store-parent' +import { + agentStatusStoreStateFromSnapshot, + createEmptyAgentStatusStoreState, + deepFreezeAgentStatusStoreValue, + snapshotFromAgentStatusStoreState +} from './agent-status-store-state' +import { + parseAgentStatusSubject, + serializeAgentStatusSubject, + type AgentStatusSubject +} from './agent-status-subject' +import { + parseAgentStatusTransportEnvelope, + type AgentStatusMutationEnvelope +} from './agent-status-transport-envelope' + +export type AgentStatusStoreMode = 'authority' | 'replica' + +export type AgentStatusStore = { + getParent(subject: AgentStatusSubject): AgentStatusParentRecord | null + getChildren(subject: AgentStatusSubject): AgentChildWorkRecord[] + getSnapshot(): AgentStatusStoreSnapshot + applyMutation(mutation: unknown): AgentStatusMutationEnvelope | null + applySnapshot(snapshot: unknown): boolean + applyTransportEnvelope(envelope: unknown): boolean +} + +export type CreateAgentStatusStoreOptions = { + epoch: string + mode: AgentStatusStoreMode +} + +export function createAgentStatusStore(options: CreateAgentStatusStoreOptions): AgentStatusStore { + if (!isAgentStatusStoreEpoch(options.epoch)) { + throw new Error('Invalid agent status store epoch') + } + let state = createEmptyAgentStatusStoreState(options.epoch) + let snapshotApplied = options.mode === 'authority' + + const store: AgentStatusStore = { + getParent(subject) { + const parsed = parseAgentStatusSubject(subject) + if (!parsed) { + return null + } + const record = state.parents.get(serializeAgentStatusSubject(parsed)) + return record ? deepFreezeAgentStatusStoreValue(parseAgentStatusParentRecord(record)) : null + }, + getChildren(subject) { + const parsed = parseAgentStatusSubject(subject) + if (!parsed) { + return [] + } + const children = [...state.children.values()] + .filter((child) => agentChildWorkBelongsTo(child, parsed)) + .map((child) => parseAgentChildWorkRecord(child)) + .filter((child): child is AgentChildWorkRecord => child !== null) + return deepFreezeAgentStatusStoreValue(children) + }, + getSnapshot() { + return snapshotFromAgentStatusStoreState(state) + }, + applyMutation(value) { + if (options.mode !== 'authority') { + return null + } + const mutation = parseAgentStatusStoreMutation(value) + if (!mutation || state.revision === Number.MAX_SAFE_INTEGER) { + return null + } + const previousRevision = state.revision + const next = applyAgentStatusStoreMutation(state, mutation, previousRevision + 1) + if (!next) { + return null + } + state = next + return deepFreezeAgentStatusStoreValue({ + type: 'mutation', + epoch: state.epoch, + previousRevision, + revision: state.revision, + mutation + }) + }, + applySnapshot(value) { + const snapshot = parseAgentStatusStoreSnapshot(value) + if (!snapshot) { + return false + } + if (options.mode === 'authority') { + if (state.revision !== 0) { + return false + } + const restored = agentStatusStoreStateFromSnapshot(snapshot, options.epoch) + if (!restored) { + return false + } + state = restored + snapshotApplied = true + return true + } + if ( + snapshotApplied && + snapshot.epoch === state.epoch && + snapshot.revision <= state.revision + ) { + return false + } + const mirrored = agentStatusStoreStateFromSnapshot(snapshot, snapshot.epoch) + if (!mirrored) { + return false + } + state = mirrored + snapshotApplied = true + return true + }, + applyTransportEnvelope(value) { + if (options.mode !== 'replica') { + return false + } + const envelope = parseAgentStatusTransportEnvelope(value) + if (!envelope) { + return false + } + if (envelope.type === 'snapshot') { + return store.applySnapshot(envelope.snapshot) + } + if ( + !snapshotApplied || + envelope.epoch !== state.epoch || + envelope.previousRevision !== state.revision || + envelope.revision !== state.revision + 1 + ) { + return false + } + const next = applyAgentStatusStoreMutation(state, envelope.mutation, envelope.revision) + if (!next) { + return false + } + state = next + return true + } + } + return store +} diff --git a/src/shared/agent-status-transport-envelope.test.ts b/src/shared/agent-status-transport-envelope.test.ts new file mode 100644 index 00000000000..81007bc7e39 --- /dev/null +++ b/src/shared/agent-status-transport-envelope.test.ts @@ -0,0 +1,110 @@ +import { describe, expect, it, vi } from 'vitest' +import { createAgentStatusStore } from './agent-status-store' +import { AGENT_STATUS_STORE_LIMITS } from './agent-status-store-contract' +import { makeStructuredAgentStatusSubject } from './agent-status-subject' +import { + deserializeAgentStatusTransportEnvelope, + serializeAgentStatusTransportEnvelope +} from './agent-status-transport-envelope' + +const parent = makeStructuredAgentStatusSubject( + { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree' + }, + 'session_11111111-1111-4111-8111-111111111111' +) + +describe('agent status transport envelope', () => { + it('requires a snapshot before replay and then accepts only contiguous same-epoch mutations', () => { + const authority = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const first = authority.applyMutation({ parent: { subject: parent, firstObservedAt: 10 } }) + expect(first).not.toBeNull() + + const replica = createAgentStatusStore({ epoch: 'replica-placeholder', mode: 'replica' }) + expect(replica.applyTransportEnvelope(first)).toBe(false) + expect( + replica.applyTransportEnvelope({ type: 'snapshot', snapshot: authority.getSnapshot() }) + ).toBe(true) + + const second = authority.applyMutation({ + facts: [{ subject: parent, key: 'acknowledged', value: true }] + }) + expect(replica.applyTransportEnvelope(second)).toBe(true) + expect(replica.getSnapshot()).toEqual(authority.getSnapshot()) + expect(replica.applyTransportEnvelope(second)).toBe(false) + expect( + replica.applyTransportEnvelope({ + ...second, + previousRevision: 9, + revision: 10 + }) + ).toBe(false) + }) + + it('adopts a restart snapshot before replay and rejects the predecessor epoch', () => { + const firstAuthority = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + firstAuthority.applyMutation({ parent: { subject: parent } }) + const persisted = firstAuthority.getSnapshot() + const replica = createAgentStatusStore({ epoch: 'replica-placeholder', mode: 'replica' }) + replica.applySnapshot(persisted) + const stale = firstAuthority.applyMutation({ + facts: [{ subject: parent, key: 'unread', value: true }] + }) + + const restarted = createAgentStatusStore({ epoch: 'epoch-b', mode: 'authority' }) + expect(restarted.applySnapshot(persisted)).toBe(true) + const restartedSnapshot = restarted.getSnapshot() + expect(restartedSnapshot.epoch).toBe('epoch-b') + expect(replica.applySnapshot(restartedSnapshot)).toBe(true) + expect(replica.applyTransportEnvelope(stale)).toBe(false) + + const fresh = restarted.applyMutation({ + facts: [{ subject: parent, key: 'retained', value: true }] + }) + expect(replica.applyTransportEnvelope(fresh)).toBe(true) + expect(replica.getSnapshot().facts.map((fact) => fact.key)).toEqual(['retained']) + }) + + it('round-trips a mutation envelope and fails closed on malformed or oversized input', () => { + const authority = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const mutation = authority.applyMutation({ parent: { subject: parent } }) + expect(mutation).not.toBeNull() + if (!mutation) { + return + } + + expect( + deserializeAgentStatusTransportEnvelope(serializeAgentStatusTransportEnvelope(mutation)) + ).toEqual(mutation) + expect( + deserializeAgentStatusTransportEnvelope( + JSON.stringify({ ...mutation, revision: mutation.revision + 2 }) + ) + ).toBeNull() + + const parse = vi.spyOn(JSON, 'parse') + expect( + deserializeAgentStatusTransportEnvelope( + ' '.repeat(AGENT_STATUS_STORE_LIMITS.serializedBytes + 1) + ) + ).toBeNull() + expect(parse).not.toHaveBeenCalled() + parse.mockRestore() + }) + + it('does not resurrect an exact removal through stale replay', () => { + const authority = createAgentStatusStore({ epoch: 'epoch-a', mode: 'authority' }) + const insertion = authority.applyMutation({ parent: { subject: parent } }) + const replica = createAgentStatusStore({ epoch: 'replica-placeholder', mode: 'replica' }) + expect(replica.applySnapshot(authority.getSnapshot())).toBe(true) + const removal = authority.applyMutation({ removeParent: parent }) + + expect(replica.applyTransportEnvelope(removal)).toBe(true) + expect(replica.getParent(parent)).toBeNull() + expect(replica.applyTransportEnvelope(insertion)).toBe(false) + expect(replica.getParent(parent)).toBeNull() + }) +}) diff --git a/src/shared/agent-status-transport-envelope.ts b/src/shared/agent-status-transport-envelope.ts new file mode 100644 index 00000000000..51fd0800db8 --- /dev/null +++ b/src/shared/agent-status-transport-envelope.ts @@ -0,0 +1,113 @@ +import { + AGENT_STATUS_STORE_LIMITS, + type AgentStatusStoreMutation, + type AgentStatusStoreSnapshot +} from './agent-status-store-contract' +import { + isAgentStatusStoreEpoch, + parseAgentStatusStoreMutation, + parseAgentStatusStoreSnapshot +} from './agent-status-store-codec' +import { assertJsonTextStructureWithinLimits } from './json-text-structure-limit' +import { measureUtf8ByteLength } from './utf8-byte-limits' + +const ENVELOPE_STRUCTURE_LIMITS = { + structuralTokens: 512 * 1024, + nestingDepth: 40 +} as const + +export type AgentStatusSnapshotEnvelope = { + type: 'snapshot' + snapshot: AgentStatusStoreSnapshot +} + +export type AgentStatusMutationEnvelope = { + type: 'mutation' + epoch: string + previousRevision: number + revision: number + mutation: AgentStatusStoreMutation +} + +export type AgentStatusTransportEnvelope = AgentStatusSnapshotEnvelope | AgentStatusMutationEnvelope + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasExactKeys(record: Record, expected: readonly string[]): boolean { + const keys = Object.keys(record) + return keys.length === expected.length && keys.every((key) => expected.includes(key)) +} + +function isRevision(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +export function parseAgentStatusTransportEnvelope( + value: unknown +): AgentStatusTransportEnvelope | null { + if (!isRecord(value)) { + return null + } + if (value.type === 'snapshot' && hasExactKeys(value, ['type', 'snapshot'])) { + const snapshot = parseAgentStatusStoreSnapshot(value.snapshot) + return snapshot ? { type: 'snapshot', snapshot } : null + } + if ( + value.type !== 'mutation' || + !hasExactKeys(value, ['type', 'epoch', 'previousRevision', 'revision', 'mutation']) || + !isAgentStatusStoreEpoch(value.epoch) || + !isRevision(value.previousRevision) || + !isRevision(value.revision) || + value.revision !== value.previousRevision + 1 + ) { + return null + } + const mutation = parseAgentStatusStoreMutation(value.mutation) + return mutation + ? { + type: 'mutation', + epoch: value.epoch, + previousRevision: value.previousRevision, + revision: value.revision, + mutation + } + : null +} + +export function serializeAgentStatusTransportEnvelope( + envelope: AgentStatusTransportEnvelope +): string { + const parsed = parseAgentStatusTransportEnvelope(envelope) + if (!parsed) { + throw new Error('Invalid agent status transport envelope') + } + const serialized = JSON.stringify(parsed) + if ( + measureUtf8ByteLength(serialized, { + stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes + }).exceededLimit + ) { + throw new Error('Agent status transport envelope exceeds its serialized-byte limit') + } + return serialized +} + +export function deserializeAgentStatusTransportEnvelope( + serialized: string +): AgentStatusTransportEnvelope | null { + if ( + measureUtf8ByteLength(serialized, { + stopAfterBytes: AGENT_STATUS_STORE_LIMITS.serializedBytes + }).exceededLimit + ) { + return null + } + try { + assertJsonTextStructureWithinLimits(serialized, ENVELOPE_STRUCTURE_LIMITS) + return parseAgentStatusTransportEnvelope(JSON.parse(serialized)) + } catch { + return null + } +}