diff --git a/scripts/t3layer-mcp-server.ts b/scripts/t3layer-mcp-server.ts index 1530ca9..7c010b4 100644 --- a/scripts/t3layer-mcp-server.ts +++ b/scripts/t3layer-mcp-server.ts @@ -46,6 +46,13 @@ export const T3LAYER_APP_REQUIRED_TOOL_NAMES = [ "stop", "respondToApproval", "respondToUserInput", + "archive", + "unarchive", + "settle", + "unsettle", + "snooze", + "unsnooze", + "updateMeta", ] as const satisfies readonly StockT3McpToolName[]; const LOCAL_TOOL_NAMES = new Set(T3LAYER_LOCAL_TOOL_NAMES); diff --git a/src/facade.ts b/src/facade.ts index d773cbc..33b1bbd 100644 --- a/src/facade.ts +++ b/src/facade.ts @@ -7,6 +7,7 @@ import type { SpawnResult, StockSpawnInput, T3NativeRuntime, + ThreadMetaFields, TurnReceipt, UserInputResponse, } from "./nativeRuntime"; @@ -46,6 +47,7 @@ export type { RetryState, RuntimeModelSelection, RuntimeOperationOptions, + ThreadMetaFields, SanitizedRetryError, SpawnResult, StockRuntimeErrorCode, @@ -185,6 +187,11 @@ export function createStockT3Facade( : createWorkerOverlay(options.overlay); const lifecycleSequenceByRef = new Map(); const lifecycleMutationsByRef = new Map(); + const lifecycleInitialTerminalByRef = new Map(); + const lifecycleTerminalOutcomeByRef = new Map< + string, + { readonly sequence: number; readonly terminal: boolean } + >(); function lifecycleKey(ref: AgentRef): string { return JSON.stringify([ref.environmentId, ref.threadId]); @@ -202,17 +209,61 @@ export function createStockT3Facade( else lifecycleMutationsByRef.set(key, remaining); } + function rememberLifecycleStart(key: string, wasTerminal: boolean): void { + if ((lifecycleMutationsByRef.get(key) ?? 0) === 1) { + lifecycleInitialTerminalByRef.set(key, wasTerminal); + } + } + + function rememberLifecycleOutcome( + key: string, + sequence: number, + terminal: boolean, + ): void { + const current = lifecycleTerminalOutcomeByRef.get(key); + if (current === undefined || sequence >= current.sequence) { + lifecycleTerminalOutcomeByRef.set(key, { sequence, terminal }); + } + lifecycleSequenceByRef.set( + key, + Math.max(lifecycleSequenceByRef.get(key) ?? 0, sequence), + ); + } + + function completeLifecycleMutation(ref: AgentRef, key: string): void { + endLifecycleMutation(key); + if ((lifecycleMutationsByRef.get(key) ?? 0) > 0) { + recordWorkerTerminalState(overlay, ref, false); + return; + } + const terminal = + lifecycleTerminalOutcomeByRef.get(key)?.terminal ?? + lifecycleInitialTerminalByRef.get(key) ?? + false; + lifecycleTerminalOutcomeByRef.delete(key); + lifecycleInitialTerminalByRef.delete(key); + recordWorkerTerminalState(overlay, ref, terminal); + if (terminal) pruneLifecycleTracking(ref); + } + function pruneLifecycleTracking(ref: AgentRef): void { const key = lifecycleKey(ref); lifecycleSequenceByRef.delete(key); if ((lifecycleMutationsByRef.get(key) ?? 0) === 0) { lifecycleMutationsByRef.delete(key); + lifecycleInitialTerminalByRef.delete(key); + lifecycleTerminalOutcomeByRef.delete(key); } } function recordTerminalRelease(ref: AgentRef, observedSequence: number): boolean { - const minimumSequence = lifecycleSequenceByRef.get(lifecycleKey(ref)); + const key = lifecycleKey(ref); + const minimumSequence = lifecycleSequenceByRef.get(key); if (minimumSequence !== undefined && observedSequence < minimumSequence) return false; + if ((lifecycleMutationsByRef.get(key) ?? 0) > 0) { + rememberLifecycleOutcome(key, observedSequence, true); + return true; + } recordWorkerTerminalState(overlay, ref, true); pruneLifecycleTracking(ref); return true; @@ -230,26 +281,24 @@ export function createStockT3Facade( let wasTerminal: boolean; try { wasTerminal = recordWorkerTerminalState(overlay, ref, false); + rememberLifecycleStart(mutationKey, wasTerminal); } catch (error) { endLifecycleMutation(mutationKey); throw error; } + let result: ControlOperationResult; try { - const result = await action(); - lifecycleSequenceByRef.set(mutationKey, result.snapshot.snapshotSequence); - const terminal = isTerminalSnapshot(result.snapshot); - recordWorkerTerminalState(overlay, ref, terminal); - endLifecycleMutation(mutationKey); - if (terminal) pruneLifecycleTracking(ref); - return result; + result = await action(); } catch (error) { - endLifecycleMutation(mutationKey); - if (wasTerminal) { - recordWorkerTerminalState(overlay, ref, true); - pruneLifecycleTracking(ref); - } + completeLifecycleMutation(ref, mutationKey); throw error; } + if (result.kind !== "pending") { + const terminal = isTerminalSnapshot(result.snapshot); + rememberLifecycleOutcome(mutationKey, result.snapshot.snapshotSequence, terminal); + } + completeLifecycleMutation(ref, mutationKey); + return result; } async function requireCanonical( @@ -338,23 +387,25 @@ export function createStockT3Facade( let wasTerminal: boolean; try { wasTerminal = recordWorkerTerminalState(overlay, ref, false); + rememberLifecycleStart(mutationKey, wasTerminal); } catch (error) { endLifecycleMutation(mutationKey); throw error; } + let receipt: TurnReceipt; try { - const receipt = await runtime.send(ref, message, options); - lifecycleSequenceByRef.set( - mutationKey, - Math.max(receipt.observedSequence, receipt.acceptedSequence ?? 0), - ); - endLifecycleMutation(mutationKey); - return receipt; + receipt = await runtime.send(ref, message, options); } catch (error) { - endLifecycleMutation(mutationKey); - if (wasTerminal) recordWorkerTerminalState(overlay, ref, true); + completeLifecycleMutation(ref, mutationKey); throw error; } + rememberLifecycleOutcome( + mutationKey, + Math.max(receipt.observedSequence, receipt.acceptedSequence ?? 0), + false, + ); + completeLifecycleMutation(ref, mutationKey); + return receipt; }, async wait(receipt: TurnReceipt, options?: RuntimeOperationOptions) { try { @@ -405,6 +456,23 @@ export function createStockT3Facade( ref, () => runtime.respondToUserInput(ref, response, options), ), + archive: async (ref: AgentRef, options?: RuntimeOperationOptions) => + runControlLifecycleMutation(ref, () => runtime.archive(ref, options)), + unarchive: async (ref: AgentRef, options?: RuntimeOperationOptions) => + runControlLifecycleMutation(ref, () => runtime.unarchive(ref, options)), + settle: async (ref: AgentRef, options?: RuntimeOperationOptions) => + runControlLifecycleMutation(ref, () => runtime.settle(ref, options)), + unsettle: async (ref: AgentRef, options?: RuntimeOperationOptions) => + runControlLifecycleMutation(ref, () => runtime.unsettle(ref, options)), + snooze: async (ref: AgentRef, until: string, options?: RuntimeOperationOptions) => + runControlLifecycleMutation(ref, () => runtime.snooze(ref, until, options)), + unsnooze: async (ref: AgentRef, options?: RuntimeOperationOptions) => + runControlLifecycleMutation(ref, () => runtime.unsnooze(ref, options)), + updateMeta: async ( + ref: AgentRef, + fields: ThreadMetaFields, + options?: RuntimeOperationOptions, + ) => runControlLifecycleMutation(ref, () => runtime.updateMeta(ref, fields, options)), async observe(ref: AgentRef, options?: RuntimeOperationOptions) { const snapshot = await runtime.observe(ref, options); if (snapshot !== undefined) { diff --git a/src/mcp.ts b/src/mcp.ts index 594cbc1..0b0be99 100644 --- a/src/mcp.ts +++ b/src/mcp.ts @@ -5,6 +5,7 @@ import type { ApprovalResponse, RuntimeOperationOptions, StockSpawnInput, + ThreadMetaFields, TurnReceipt, UserInputResponse, } from "./nativeRuntime"; @@ -22,6 +23,13 @@ export const MCP_TOOL_NAMES = [ "stop", "respondToApproval", "respondToUserInput", + "archive", + "unarchive", + "settle", + "unsettle", + "snooze", + "unsnooze", + "updateMeta", ] as const; export type StockT3McpToolName = (typeof MCP_TOOL_NAMES)[number]; @@ -174,6 +182,16 @@ const receiptSchema = Object.freeze({ ]), additionalProperties: false, }); +const threadMetaFieldsSchema = Object.freeze({ + type: "object", + properties: Object.freeze({ + title: stringSchema, + modelSelection: modelSelectionSchema, + branch: nullableStringSchema, + worktreePath: nullableStringSchema, + }), + additionalProperties: false, +}); function tool( name: StockT3McpToolName, @@ -256,6 +274,36 @@ const TOOLS = Object.freeze([ }), operation: operationSchema, }, ["ref", "response"]), + tool("archive", "Archive a stock thread and confirm the disposition through snapshots.", { + ref: agentRefSchema, + operation: operationSchema, + }, ["ref"]), + tool("unarchive", "Restore an archived stock thread and confirm through snapshots.", { + ref: agentRefSchema, + operation: operationSchema, + }, ["ref"]), + tool("settle", "Settle a stock thread and confirm the disposition through snapshots.", { + ref: agentRefSchema, + operation: operationSchema, + }, ["ref"]), + tool("unsettle", "Reopen a settled stock thread as an explicit user action.", { + ref: agentRefSchema, + operation: operationSchema, + }, ["ref"]), + tool("snooze", "Snooze a stock thread until an ISO timestamp and confirm through snapshots.", { + ref: agentRefSchema, + until: stringSchema, + operation: operationSchema, + }, ["ref", "until"]), + tool("unsnooze", "Wake a snoozed stock thread as an explicit user action.", { + ref: agentRefSchema, + operation: operationSchema, + }, ["ref"]), + tool("updateMeta", "Update canonical stock thread metadata and confirm through snapshots.", { + ref: agentRefSchema, + fields: threadMetaFieldsSchema, + operation: operationSchema, + }, ["ref", "fields"]), ]); function resultContent(payload: unknown) { @@ -480,6 +528,28 @@ export function createStockT3McpFacade(facade: StockT3Facade) { args.response as unknown as UserInputResponse, operation, )); + case "archive": + return success(await facade.archive(args.ref as AgentRef, operation)); + case "unarchive": + return success(await facade.unarchive(args.ref as AgentRef, operation)); + case "settle": + return success(await facade.settle(args.ref as AgentRef, operation)); + case "unsettle": + return success(await facade.unsettle(args.ref as AgentRef, operation)); + case "snooze": + return success(await facade.snooze( + args.ref as AgentRef, + args.until as string, + operation, + )); + case "unsnooze": + return success(await facade.unsnooze(args.ref as AgentRef, operation)); + case "updateMeta": + return success(await facade.updateMeta( + args.ref as AgentRef, + args.fields as ThreadMetaFields, + operation, + )); default: return protocolMismatch("tool.name"); } diff --git a/src/nativeRuntime.ts b/src/nativeRuntime.ts index 6ce10b3..7e477ba 100644 --- a/src/nativeRuntime.ts +++ b/src/nativeRuntime.ts @@ -2,7 +2,9 @@ import type { EnvironmentDescriptor, ShellSnapshot, StockMessage, + StockReadModelSnapshot, StockThreadDetail, + StockThreadIdentity, StockThreadShell, ThreadDetailSnapshot, ConnectionProfile, @@ -157,7 +159,26 @@ export type ControlOperationName = | "interrupt" | "stop" | "respond_to_approval" - | "respond_to_user_input"; + | "respond_to_user_input" + | "archive" + | "unarchive" + | "settle" + | "unsettle" + | "snooze" + | "unsnooze" + | "update_meta"; + +export type LifecycleOperationName = Extract< + ControlOperationName, + "archive" | "unarchive" | "settle" | "unsettle" | "snooze" | "unsnooze" | "update_meta" +>; + +export interface ThreadMetaFields { + readonly title?: string; + readonly modelSelection?: RuntimeModelSelection; + readonly branch?: string | null; + readonly worktreePath?: string | null; +} export const APPROVAL_DECISIONS = [ "accept", @@ -199,11 +220,28 @@ export type ControlOperationResult = } | { readonly kind: "no_op"; - readonly operation: "interrupt" | "stop"; - readonly reason: "turn_terminal" | "session_terminal"; + readonly operation: ControlOperationName; + readonly reason: + | "turn_terminal" + | "session_terminal" + | "already_archived" + | "already_unarchived" + | "already_settled" + | "already_unsettled" + | "already_snoozed_until_target" + | "not_snoozed" + | "metadata_unchanged"; readonly agentRef: AgentRef; readonly observedSequence: number; readonly snapshot: ThreadDetailSnapshot; + } + | { + readonly kind: "pending"; + readonly operation: LifecycleOperationName; + readonly reason: "projection_pending"; + readonly agentRef: AgentRef; + readonly receipt: ControlReceipt; + readonly snapshot: ThreadDetailSnapshot; }; export type SpawnResult = @@ -479,6 +517,9 @@ export function parseProjectCreateIdentity( export interface StockT3RuntimeClient { readonly getDescriptor: (options?: RequestBoundaryOptions) => Promise; readonly getShell: (options?: RequestBoundaryOptions) => Promise; + readonly getSnapshot?: ( + options?: RequestBoundaryOptions, + ) => Promise; readonly getThread: ( threadId: string, options?: RequestBoundaryOptions, @@ -2918,6 +2959,631 @@ export function createStockT3NativeRuntime(options: StockT3NativeRuntimeOptions) ); } + type LifecycleNoOpReason = Extract< + ControlOperationResult, + { readonly kind: "no_op" } + >["reason"]; + + function lifecycleNoOp( + operationName: LifecycleOperationName, + reason: LifecycleNoOpReason, + ref: AgentRef, + preflight: ControlPreflight, + ): ControlOperationResult { + return { + kind: "no_op", + operation: operationName, + reason, + agentRef: ref, + observedSequence: observedControlSequence(preflight), + snapshot: preflight.detail, + }; + } + + async function dispatchLifecycle( + operationName: LifecycleOperationName, + ref: AgentRef, + command: Readonly>, + commandId: string, + preflight: ControlPreflight, + operation: RuntimeOperationOptions & { readonly deadlineMs: number }, + predicate: ControlPredicate, + ): Promise { + try { + return await dispatchControl( + operationName, + ref, + command, + commandId, + observedControlSequence(preflight), + operation, + predicate, + ); + } catch (error) { + const receipt = error instanceof StockRuntimeError + ? error.evidence.receipt as ControlReceipt | undefined + : undefined; + if (error instanceof StockRuntimeError && error.code === "timeout" && receipt !== undefined) { + return { + kind: "pending", + operation: operationName, + reason: "projection_pending", + agentRef: ref, + receipt, + snapshot: preflight.detail, + }; + } + throw error; + } + } + + async function disposition( + operationName: LifecycleOperationName, + commandType: string, + ref: AgentRef, + operation: RuntimeOperationOptions, + noOpReason: LifecycleNoOpReason, + alreadyApplied: ControlPredicate, + payload: Readonly> = {}, + requiredFields: readonly (keyof StockThreadIdentity)[] = [], + validatePreflight?: (preflight: ControlPreflight) => void, + ): Promise { + const bounded = boundedControlOperation(operation); + const preflight = await preflightControl(ref, bounded); + for (const field of requiredFields) { + if (!Object.hasOwn(preflight.shellThread, field) || !Object.hasOwn(preflight.detail.thread, field)) { + throw new StockRuntimeError("protocol_mismatch", { + reason: "lifecycle_projection_unavailable", + field, + }); + } + } + validatePreflight?.(preflight); + if (alreadyApplied(preflight.shellThread, preflight.detail)) { + return lifecycleNoOp(operationName, noOpReason, ref, preflight); + } + const commandId = id(); + const command = Object.freeze({ + type: commandType, + commandId, + threadId: ref.threadId, + ...payload, + }); + return dispatchLifecycle( + operationName, + ref, + command, + commandId, + preflight, + bounded, + alreadyApplied, + ); + } + + function projectedOnBoth( + shellThread: StockThreadShell, + detail: ThreadDetailSnapshot, + predicate: (thread: StockThreadShell | StockThreadDetail) => boolean, + ): boolean { + return predicate(shellThread) && predicate(detail.thread); + } + + interface FullThreadPreflight { + readonly detail: ThreadDetailSnapshot; + } + + function validateFullThreadIdentity(ref: AgentRef, thread: StockThreadDetail): void { + controlIdentifier(thread.projectId, "snapshot.thread.projectId"); + if ( + thread.id !== ref.threadId || + (thread.session !== null && thread.session.threadId !== ref.threadId) + ) { + throw new StockRuntimeError("protocol_mismatch", { + reason: "control_thread_identity_conflict", + }); + } + if (thread.deletedAt != null) { + throw new StockRuntimeError("identity_conflict", { + reason: "thread_not_found", + threadId: ref.threadId, + }); + } + } + + function fullSnapshotClient(): NonNullable { + if (client.getSnapshot === undefined) { + throw new StockRuntimeError("protocol_mismatch", { + reason: "full_snapshot_unavailable", + }); + } + return client.getSnapshot; + } + + async function preflightFullThread( + ref: AgentRef, + operation: RuntimeOperationOptions & { readonly deadlineMs: number }, + ): Promise { + validateControlRef(ref); + const stopped = stopCode(operation); + if (stopped !== null) throw new StockRuntimeError(stopped); + const getSnapshot = fullSnapshotClient(); + const [environment, snapshot] = await Promise.all([ + descriptor({ deadlineMs: operation.deadlineMs, signal: operation.signal }), + getSnapshot({ deadlineMs: operation.deadlineMs, signal: operation.signal }).catch((error) => { + throw mapReceivedError(error); + }), + ] as const); + if (environment.environmentId !== ref.environmentId) { + throw new StockRuntimeError("environment_changed", { + expectedEnvironmentId: ref.environmentId, + actualEnvironmentId: environment.environmentId, + }); + } + const thread = snapshot.threads.find((entry) => entry.id === ref.threadId); + if (thread === undefined) { + throw new StockRuntimeError("identity_conflict", { + reason: "thread_not_found", + threadId: ref.threadId, + }); + } + validateFullThreadIdentity(ref, thread); + return { detail: { snapshotSequence: snapshot.snapshotSequence, thread } }; + } + + async function waitBetweenFullSnapshotReads( + operation: RuntimeOperationOptions & { readonly deadlineMs: number }, + ): Promise { + const remainingMs = operation.deadlineMs - clock(); + if (remainingMs <= 0) return; + await new Promise((resolve) => { + const onAbort = () => { + clearTimeout(timer); + resolve(); + }; + const timer = setTimeout(() => { + operation.signal?.removeEventListener("abort", onAbort); + resolve(); + }, Math.min(25, remainingMs)); + operation.signal?.addEventListener("abort", onAbort, { once: true }); + }); + } + + async function observeFullThread( + ref: AgentRef, + minimumSequence: number, + operation: RuntimeOperationOptions & { readonly deadlineMs: number }, + predicate: (thread: StockThreadDetail) => boolean, + ): Promise<{ readonly matched: boolean; readonly detail: ThreadDetailSnapshot | null }> { + const getSnapshot = fullSnapshotClient(); + let last: ThreadDetailSnapshot | null = null; + const readLimit = Math.max(1, operation.maxReconciliationReads ?? 4); + for (let index = 0; index < readLimit; index += 1) { + const stopped = stopCode(operation); + if (stopped !== null) break; + let snapshot: StockReadModelSnapshot; + try { + snapshot = await getSnapshot({ + deadlineMs: operation.deadlineMs, + signal: operation.signal, + }); + } catch (error) { + if (isAmbiguous(error)) { + if (index + 1 < readLimit) await waitBetweenFullSnapshotReads(operation); + continue; + } + throw mapReceivedError(error); + } + const thread = snapshot.threads.find((entry) => entry.id === ref.threadId); + if (thread === undefined) { + throw new StockRuntimeError("identity_conflict", { + reason: "thread_not_found", + threadId: ref.threadId, + }); + } + validateFullThreadIdentity(ref, thread); + last = { snapshotSequence: snapshot.snapshotSequence, thread }; + if (snapshot.snapshotSequence >= minimumSequence && predicate(thread)) { + return { matched: true, detail: last }; + } + if (index + 1 < readLimit) { + await waitBetweenFullSnapshotReads(operation); + } + } + return { matched: false, detail: last }; + } + + async function dispatchFullLifecycle( + operationName: "archive" | "unarchive", + ref: AgentRef, + commandType: "thread.archive" | "thread.unarchive", + preflight: FullThreadPreflight, + operation: RuntimeOperationOptions & { readonly deadlineMs: number }, + predicate: (thread: StockThreadDetail) => boolean, + ): Promise { + const commandId = id(); + const command = Object.freeze({ type: commandType, commandId, threadId: ref.threadId }); + let acceptedSequence: number | null = null; + let retryState: ControlReceipt["retryState"] = "not_needed"; + try { + acceptedSequence = (await client.dispatch(command, { + deadlineMs: operation.deadlineMs, + signal: operation.signal, + })).sequence; + poller.dispatchObserved(ref.environmentId); + } catch (error) { + const failure = classifyDispatchFailure(error, operation); + if (failure.kind !== "ambiguous") { + if (failure.kind === "cancelled" || failure.kind === "timeout") { + throw new StockRuntimeError(failure.kind, { operation: operationName, commandId }); + } + throw failure.error; + } + const reconciled = await observeFullThread(ref, 0, operation, predicate); + if (reconciled.matched && reconciled.detail !== null) { + return appliedControlResult( + operationName, + ref, + commandId, + null, + "reconciled_before_retry", + reconciled.detail, + ); + } + const stopped = stopCode(operation); + if (stopped !== null) throw new StockRuntimeError(stopped, { operation: operationName, commandId }); + try { + acceptedSequence = (await client.dispatch(command, { + deadlineMs: operation.deadlineMs, + signal: operation.signal, + })).sequence; + retryState = "identical_retry_accepted"; + poller.dispatchObserved(ref.environmentId); + } catch (retryError) { + const retryFailure = classifyDispatchFailure(retryError, operation); + if (retryFailure.kind === "cancelled" || retryFailure.kind === "timeout") { + throw new StockRuntimeError(retryFailure.kind, { operation: operationName, commandId }); + } + if ( + retryFailure.kind === "received" && + (retryFailure.error.code === "protocol_mismatch" || + ![400, 401, 403, 500].includes( + typeof retryFailure.error.evidence.status === "number" + ? retryFailure.error.evidence.status + : 0, + )) + ) { + throw retryFailure.error; + } + retryState = "identical_retry_outcome_unknown"; + } + } + const observed = await observeFullThread(ref, acceptedSequence ?? 0, operation, predicate); + const snapshot = observed.detail ?? preflight.detail; + const receipt: ControlReceipt = { + operation: operationName, + commandId, + acceptedSequence, + observedSequence: snapshot.snapshotSequence, + retryState, + }; + return observed.matched + ? appliedControlResult( + operationName, + ref, + commandId, + acceptedSequence, + retryState, + snapshot, + ) + : { + kind: "pending", + operation: operationName, + reason: "projection_pending", + agentRef: ref, + receipt, + snapshot, + }; + } + + async function archive( + ref: AgentRef, + operation: RuntimeOperationOptions = {}, + ): Promise { + const bounded = boundedControlOperation(operation); + const preflight = await preflightFullThread(ref, bounded); + if (!Object.hasOwn(preflight.detail.thread, "archivedAt")) { + throw new StockRuntimeError("protocol_mismatch", { + reason: "lifecycle_projection_unavailable", + field: "archivedAt", + }); + } + if (preflight.detail.thread.archivedAt != null) { + return { + kind: "no_op", + operation: "archive", + reason: "already_archived", + agentRef: ref, + observedSequence: preflight.detail.snapshotSequence, + snapshot: preflight.detail, + }; + } + return dispatchFullLifecycle( + "archive", + ref, + "thread.archive", + preflight, + bounded, + (thread) => thread.archivedAt != null, + ); + } + + async function unarchive( + ref: AgentRef, + operation: RuntimeOperationOptions = {}, + ): Promise { + const bounded = boundedControlOperation(operation); + const preflight = await preflightFullThread(ref, bounded); + if (!Object.hasOwn(preflight.detail.thread, "archivedAt")) { + throw new StockRuntimeError("protocol_mismatch", { + reason: "lifecycle_projection_unavailable", + field: "archivedAt", + }); + } + if (preflight.detail.thread.archivedAt == null) { + return { + kind: "no_op", + operation: "unarchive", + reason: "already_unarchived", + agentRef: ref, + observedSequence: preflight.detail.snapshotSequence, + snapshot: preflight.detail, + }; + } + return dispatchFullLifecycle( + "unarchive", + ref, + "thread.unarchive", + preflight, + bounded, + (thread) => thread.archivedAt == null, + ); + } + + async function settle( + ref: AgentRef, + operation: RuntimeOperationOptions = {}, + ): Promise { + return disposition( + "settle", + "thread.settle", + ref, + operation, + "already_settled", + (shellThread, detail) => projectedOnBoth( + shellThread, + detail, + (thread) => thread.settledAt != null && thread.settledOverride !== "active", + ), + {}, + ["settledAt", "settledOverride"], + (preflight) => { + const shell = preflight.shellThread; + const sessionActive = [shell.session, preflight.detail.thread.session].some( + (session) => session?.status === "starting" || session?.status === "running", + ); + if ( + sessionActive || + shell.hasPendingApprovals || + shell.hasPendingUserInput || + hasQueuedTurnStart(shell) + ) { + throw new StockRuntimeError("command_rejected", { + reason: "thread_not_settleable", + }); + } + }, + ); + } + + async function unsettle( + ref: AgentRef, + operation: RuntimeOperationOptions = {}, + ): Promise { + return disposition( + "unsettle", + "thread.unsettle", + ref, + operation, + "already_unsettled", + (shellThread, detail) => projectedOnBoth( + shellThread, + detail, + (thread) => thread.settledOverride === "active", + ), + { reason: "user" }, + ["settledAt", "settledOverride"], + ); + } + + function snoozeUntil(value: unknown): string { + if (typeof value !== "string" || value.trim() !== value || !Number.isFinite(Date.parse(value))) { + throw new StockRuntimeError("protocol_mismatch", { field: "snoozedUntil" }); + } + return value; + } + + async function snooze( + ref: AgentRef, + until: string, + operation: RuntimeOperationOptions = {}, + ): Promise { + const parsed = snoozeUntil(until); + if (!(Date.parse(parsed) > Date.parse(now()))) { + throw new StockRuntimeError("command_rejected", { + reason: "snooze_not_in_future", + }); + } + return disposition( + "snooze", + "thread.snooze", + ref, + operation, + "already_snoozed_until_target", + (shellThread, detail) => projectedOnBoth( + shellThread, + detail, + (thread) => thread.snoozedUntil === parsed, + ), + { snoozedUntil: parsed }, + ["snoozedUntil"], + (preflight) => { + const shell = preflight.shellThread; + if ( + shell.hasPendingApprovals || + shell.hasPendingUserInput || + hasQueuedTurnStart(shell) + ) { + throw new StockRuntimeError("command_rejected", { + reason: "thread_not_snoozable", + }); + } + }, + ); + } + + function hasQueuedTurnStart(thread: StockThreadShell): boolean { + if (thread.latestUserMessageAt == null || thread.session?.status === "error") return false; + const messageAt = Date.parse(thread.latestUserMessageAt); + const current = Date.parse(now()); + if (!Number.isFinite(messageAt) || !Number.isFinite(current)) return false; + if (Math.abs(current - messageAt) > 2 * 60_000) return false; + const turn = thread.latestTurn; + if (turn === null) return true; + return [turn.requestedAt, turn.startedAt, turn.completedAt].every( + (candidate) => candidate == null || Date.parse(candidate) < messageAt, + ); + } + + function activelySnoozed(value: string | null | undefined): boolean { + return value != null && Number.isFinite(Date.parse(value)) && Date.parse(value) > Date.parse(now()); + } + + async function unsnooze( + ref: AgentRef, + operation: RuntimeOperationOptions = {}, + ): Promise { + return disposition( + "unsnooze", + "thread.unsnooze", + ref, + operation, + "not_snoozed", + (shellThread, detail) => projectedOnBoth( + shellThread, + detail, + (thread) => !activelySnoozed(thread.snoozedUntil), + ), + { reason: "user" }, + ["snoozedUntil"], + ); + } + + function nullableTrimmed(value: unknown, field: string): string | null { + if (value === null) return null; + if (typeof value !== "string" || value.trim().length === 0 || value.trim() !== value) { + throw new StockRuntimeError("protocol_mismatch", { field }); + } + return value; + } + + function trimmed(value: unknown, field: string): string { + const parsed = nullableTrimmed(value, field); + if (parsed === null) throw new StockRuntimeError("protocol_mismatch", { field }); + return parsed; + } + + function parseThreadMetaFields(value: ThreadMetaFields): ThreadMetaFields { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new StockRuntimeError("protocol_mismatch", { field: "fields" }); + } + const record = value as Readonly>; + const allowed = new Set(["title", "modelSelection", "branch", "worktreePath"]); + for (const key of Object.keys(record)) { + if (!allowed.has(key)) throw new StockRuntimeError("protocol_mismatch", { field: `fields.${key}` }); + } + if (Object.keys(record).length === 0) { + throw new StockRuntimeError("protocol_mismatch", { field: "fields" }); + } + const parsed: { + title?: string; + modelSelection?: RuntimeModelSelection; + branch?: string | null; + worktreePath?: string | null; + } = {}; + if (Object.hasOwn(record, "title")) parsed.title = trimmed(record.title, "fields.title"); + if (Object.hasOwn(record, "branch")) parsed.branch = nullableTrimmed(record.branch, "fields.branch"); + if (Object.hasOwn(record, "worktreePath")) { + parsed.worktreePath = nullableTrimmed(record.worktreePath, "fields.worktreePath"); + } + if (Object.hasOwn(record, "modelSelection")) { + const selection = record.modelSelection; + if (typeof selection !== "object" || selection === null || Array.isArray(selection)) { + throw new StockRuntimeError("protocol_mismatch", { field: "fields.modelSelection" }); + } + const model = selection as Readonly>; + for (const key of Object.keys(model)) { + if (!["instanceId", "model", "options"].includes(key)) { + throw new StockRuntimeError("protocol_mismatch", { field: `fields.modelSelection.${key}` }); + } + } + parsed.modelSelection = { + instanceId: trimmed(model.instanceId, "fields.modelSelection.instanceId"), + model: trimmed(model.model, "fields.modelSelection.model"), + ...(model.options === undefined ? {} : { options: model.options as readonly unknown[] }), + }; + if (parsed.modelSelection.options !== undefined && !Array.isArray(parsed.modelSelection.options)) { + throw new StockRuntimeError("protocol_mismatch", { field: "fields.modelSelection.options" }); + } + } + return Object.freeze(parsed); + } + + function sameModelSelection(left: RuntimeModelSelection, right: RuntimeModelSelection): boolean { + return canonical(left) === canonical(right); + } + + async function updateMeta( + ref: AgentRef, + fields: ThreadMetaFields, + operation: RuntimeOperationOptions = {}, + ): Promise { + const parsed = parseThreadMetaFields(fields); + const bounded = boundedControlOperation(operation); + const initialStop = stopCode(bounded); + if (initialStop !== null) throw new StockRuntimeError(initialStop); + if (parsed.modelSelection !== undefined) { + await validateModelSelectionAvailable(parsed.modelSelection); + const stoppedAfterModelValidation = stopCode(bounded); + if (stoppedAfterModelValidation !== null) { + throw new StockRuntimeError(stoppedAfterModelValidation); + } + } + const matches = (thread: StockThreadShell | StockThreadDetail) => + (parsed.title === undefined || thread.title === parsed.title) && + (parsed.modelSelection === undefined || sameModelSelection(thread.modelSelection, parsed.modelSelection)) && + (!Object.hasOwn(parsed, "branch") || thread.branch === parsed.branch) && + (!Object.hasOwn(parsed, "worktreePath") || thread.worktreePath === parsed.worktreePath); + return disposition( + "update_meta", + "thread.meta.update", + ref, + bounded, + "metadata_unchanged", + (shellThread, detail) => matches(shellThread) && matches(detail.thread), + parsed as Readonly>, + ); + } + async function interrupt( ref: AgentRef, operation: RuntimeOperationOptions = {}, @@ -3520,6 +4186,13 @@ export function createStockT3NativeRuntime(options: StockT3NativeRuntimeOptions) stop, respondToApproval, respondToUserInput, + archive, + unarchive, + settle, + unsettle, + snooze, + unsnooze, + updateMeta, send, wait, async observe(ref: AgentRef, operation: RuntimeOperationOptions = {}) { diff --git a/src/stockT3Contracts.ts b/src/stockT3Contracts.ts index 4c989b3..1b4e336 100644 --- a/src/stockT3Contracts.ts +++ b/src/stockT3Contracts.ts @@ -109,6 +109,12 @@ export interface StockThreadIdentity { readonly latestTurn: StockLatestTurn | null; readonly createdAt: string; readonly updatedAt: string; + readonly archivedAt?: string | null; + readonly settledOverride?: "settled" | "active" | null; + readonly settledAt?: string | null; + readonly snoozedUntil?: string | null; + readonly snoozedAt?: string | null; + readonly deletedAt?: string | null; readonly session: StockSession | null; } @@ -147,6 +153,13 @@ export interface ThreadDetailSnapshot { readonly thread: StockThreadDetail; } +export interface StockReadModelSnapshot { + readonly snapshotSequence: number; + readonly projects: readonly StockProjectShell[]; + readonly threads: readonly StockThreadDetail[]; + readonly updatedAt: string; +} + function literal(value: unknown, allowed: readonly T[], path: string): T { if (typeof value !== "string" || !allowed.includes(value as T)) { throw new ProtocolMismatchError(path); @@ -205,6 +218,48 @@ function decodeThreadIdentity(input: JsonObject, path: string): StockThreadIdent latestTurn: decodeLatestTurn(input.latestTurn, `${path}.latestTurn`), createdAt: iso(input.createdAt, `${path}.createdAt`), updatedAt: iso(input.updatedAt, `${path}.updatedAt`), + ...(input.archivedAt === undefined + ? {} + : { + archivedAt: input.archivedAt === null + ? null + : iso(input.archivedAt, `${path}.archivedAt`), + }), + ...(input.settledOverride === undefined + ? {} + : { + settledOverride: input.settledOverride === null + ? null + : literal(input.settledOverride, ["settled", "active"] as const, `${path}.settledOverride`), + }), + ...(input.settledAt === undefined + ? {} + : { + settledAt: input.settledAt === null + ? null + : iso(input.settledAt, `${path}.settledAt`), + }), + ...(input.snoozedUntil === undefined + ? {} + : { + snoozedUntil: input.snoozedUntil === null + ? null + : iso(input.snoozedUntil, `${path}.snoozedUntil`), + }), + ...(input.snoozedAt === undefined + ? {} + : { + snoozedAt: input.snoozedAt === null + ? null + : iso(input.snoozedAt, `${path}.snoozedAt`), + }), + ...(input.deletedAt === undefined + ? {} + : { + deletedAt: input.deletedAt === null + ? null + : iso(input.deletedAt, `${path}.deletedAt`), + }), session: decodeSession(input.session, `${path}.session`), }; } @@ -306,15 +361,34 @@ export function decodeThreadDetailSnapshot( const input = object(value, "detail"); const snapshotSequence = integer(input.snapshotSequence, "detail.snapshotSequence"); assertSequence(snapshotSequence, options.minimumSequence, "detail.snapshotSequence"); - const threadInput = object(input.thread, "detail.thread"); return { snapshotSequence, - thread: { - ...decodeThreadIdentity(threadInput, "detail.thread"), - messages: array(threadInput.messages, "detail.thread.messages", decodeMessage), - activities: array(threadInput.activities, "detail.thread.activities", (entry) => entry), - checkpoints: array(threadInput.checkpoints, "detail.thread.checkpoints", (entry) => entry), - }, + thread: decodeThreadDetail(input.thread, "detail.thread"), + }; +} + +function decodeThreadDetail(value: unknown, path: string): StockThreadDetail { + const input = object(value, path); + return { + ...decodeThreadIdentity(input, path), + messages: array(input.messages, `${path}.messages`, decodeMessage), + activities: array(input.activities, `${path}.activities`, (entry) => entry), + checkpoints: array(input.checkpoints, `${path}.checkpoints`, (entry) => entry), + }; +} + +export function decodeReadModelSnapshot( + value: unknown, + options: { readonly minimumSequence?: number } = {}, +): StockReadModelSnapshot { + const input = object(value, "snapshot"); + const snapshotSequence = integer(input.snapshotSequence, "snapshot.snapshotSequence"); + assertSequence(snapshotSequence, options.minimumSequence, "snapshot.snapshotSequence"); + return { + snapshotSequence, + projects: array(input.projects, "snapshot.projects", decodeProject), + threads: array(input.threads, "snapshot.threads", decodeThreadDetail), + updatedAt: iso(input.updatedAt, "snapshot.updatedAt"), }; } diff --git a/src/stockT3HttpClient.ts b/src/stockT3HttpClient.ts index 995df2a..04cf7a4 100644 --- a/src/stockT3HttpClient.ts +++ b/src/stockT3HttpClient.ts @@ -4,10 +4,12 @@ import { type EnvironmentDescriptor, type SanitizedDispatchError, type ShellSnapshot, + type StockReadModelSnapshot, type ThreadDetailSnapshot, decodeDescriptor, decodeDispatchError, decodeDispatchResult, + decodeReadModelSnapshot, decodeShellSnapshot, decodeThreadDetailSnapshot, decodeTokenResult, @@ -338,6 +340,20 @@ export function createStockT3HttpClient(options: StockT3HttpClientOptions) { ); }, + async getSnapshot(boundary: RequestBoundaryOptions = {}): Promise { + const { response, body } = await requestJson( + "/api/orchestration/snapshot", + { method: "GET" }, + boundary, + true, + ); + if (!response.ok) throw receivedHttpError(response); + return decodeOrProtocol( + () => decodeReadModelSnapshot(body, { minimumSequence: boundary.minimumSequence }), + response.status, + ); + }, + async getThread( threadId: string, boundary: RequestBoundaryOptions = {}, diff --git a/test/facade-lifecycle.test.ts b/test/facade-lifecycle.test.ts new file mode 100644 index 0000000..f3b4900 --- /dev/null +++ b/test/facade-lifecycle.test.ts @@ -0,0 +1,1082 @@ +import { afterEach, describe, expect, test } from "bun:test"; + +import { + StockRuntimeError, + type AgentRef, + type RuntimeOperationOptions, + type StockT3RuntimeClient, + type T3NativeRuntime, + type ThreadMetaFields, +} from "../src/nativeRuntime"; +import type { + ShellSnapshot, + StockThreadShell, + ThreadDetailSnapshot, +} from "../src/stockT3Contracts"; +import { decodeReadModelSnapshot } from "../src/stockT3Contracts"; +import { createStockT3NativeRuntime as createRuntime } from "./support/modelCache"; +import { createStockT3Facade } from "../src/facade"; +import { createStockT3McpFacade } from "../src/mcp"; +import { StockT3HttpError } from "../src/stockT3HttpClient"; +import { createWorkerOverlay, recordWorkerTerminalState } from "../src/overlay"; + +const ref: AgentRef = Object.freeze({ environmentId: "env-1", threadId: "thread-1" }); +const iso = "2026-08-04T10:00:00.000Z"; +const later = "2026-08-05T10:00:00.000Z"; +const operation = Object.freeze({ timeoutMs: 100, maxReconciliationReads: 1 }); +const openRuntimes = new Set>(); + +function createStockT3NativeRuntime( + options: Parameters[0], +): ReturnType { + const runtime = createRuntime(options); + openRuntimes.add(runtime); + return runtime; +} + +afterEach(() => { + for (const runtime of openRuntimes) runtime.close(); + openRuntimes.clear(); +}); + +type LifecycleMethod = + | "archive" + | "unarchive" + | "settle" + | "unsettle" + | "snooze" + | "unsnooze" + | "updateMeta"; + +type LifecycleState = { + archivedAt: string | null; + settledOverride: "settled" | "active" | null; + settledAt: string | null; + snoozedUntil?: string | null; + snoozedAt?: string | null; + title: string; + modelSelection: { instanceId: string; model: string }; + branch: string | null; + worktreePath: string | null; +}; + +const activeState: LifecycleState = { + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + title: "worker", + modelSelection: { instanceId: "claudeAgent", model: "claude-opus-5" }, + branch: null, + worktreePath: null, +}; + +function thread(state: LifecycleState): StockThreadShell { + return { + id: ref.threadId, + projectId: "project-1", + title: state.title, + modelSelection: state.modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: state.branch, + worktreePath: state.worktreePath, + latestTurn: null, + createdAt: iso, + updatedAt: iso, + archivedAt: state.archivedAt, + settledOverride: state.settledOverride, + settledAt: state.settledAt, + snoozedUntil: state.snoozedUntil, + snoozedAt: state.snoozedAt, + session: null, + latestUserMessageAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + }; +} + +function shell(sequence: number, state: LifecycleState): ShellSnapshot { + return { + snapshotSequence: sequence, + projects: [], + threads: [thread(state)], + updatedAt: iso, + }; +} + +function detail(sequence: number, state: LifecycleState): ThreadDetailSnapshot { + const current = thread(state); + return { + snapshotSequence: sequence, + thread: { + id: current.id, + projectId: current.projectId, + title: current.title, + modelSelection: current.modelSelection, + runtimeMode: current.runtimeMode, + interactionMode: current.interactionMode, + branch: current.branch, + worktreePath: current.worktreePath, + latestTurn: current.latestTurn, + createdAt: current.createdAt, + updatedAt: current.updatedAt, + archivedAt: current.archivedAt, + settledOverride: current.settledOverride, + settledAt: current.settledAt, + snoozedUntil: current.snoozedUntil, + snoozedAt: current.snoozedAt, + session: current.session, + messages: [], + activities: [], + checkpoints: [], + }, + }; +} + +function readModel(sequence: number, state: LifecycleState) { + return { + snapshotSequence: sequence, + projects: [], + threads: [detail(sequence, state).thread], + updatedAt: iso, + }; +} + +function descriptor(environmentId = ref.environmentId) { + return { + environmentId, + label: "local", + platform: { os: "darwin" as const, arch: "arm64" as const }, + serverVersion: "stock", + capabilities: { repositoryIdentity: false }, + }; +} + +function runtimeClient(input: { + state: () => LifecycleState; + sequence?: () => number; + dispatch?: StockT3RuntimeClient["dispatch"]; + calls?: string[]; +}): StockT3RuntimeClient { + const sequence = input.sequence ?? (() => 1); + return { + getDescriptor: async () => { + input.calls?.push("getDescriptor"); + return descriptor(); + }, + getShell: async () => { + input.calls?.push("getShell"); + return shell(sequence(), input.state()); + }, + getSnapshot: async () => { + input.calls?.push("getSnapshot"); + return readModel(sequence(), input.state()); + }, + getThread: async () => { + input.calls?.push("getThread"); + return detail(sequence(), input.state()); + }, + dispatch: input.dispatch ?? (async () => ({ sequence: 2 })), + }; +} + +function invoke( + runtime: ReturnType, + method: LifecycleMethod, + target: AgentRef = ref, + options: RuntimeOperationOptions = operation, +) { + switch (method) { + case "archive": return runtime.archive(target, options); + case "unarchive": return runtime.unarchive(target, options); + case "settle": return runtime.settle(target, options); + case "unsettle": return runtime.unsettle(target, options); + case "snooze": return runtime.snooze(target, later, options); + case "unsnooze": return runtime.unsnooze(target, options); + case "updateMeta": return runtime.updateMeta( + target, + { + title: "renamed", + modelSelection: { instanceId: "claudeAgent", model: "claude-sonnet-4-5" }, + branch: "feat/lifecycle", + worktreePath: "/tmp/lifecycle", + }, + options, + ); + } +} + +const cases: readonly { + method: LifecycleMethod; + command: string; + initial: LifecycleState; + applied: LifecycleState; + noOpReason: string; + operation?: string; + expectedPayload?: Readonly>; +}[] = [ + { + method: "archive", + command: "thread.archive", + initial: activeState, + applied: { ...activeState, archivedAt: later }, + noOpReason: "already_archived", + }, + { + method: "unarchive", + command: "thread.unarchive", + initial: { ...activeState, archivedAt: iso }, + applied: activeState, + noOpReason: "already_unarchived", + }, + { + method: "settle", + command: "thread.settle", + initial: activeState, + applied: { ...activeState, settledOverride: "settled", settledAt: later }, + noOpReason: "already_settled", + }, + { + method: "unsettle", + command: "thread.unsettle", + initial: { ...activeState, settledOverride: "settled", settledAt: iso }, + applied: { ...activeState, settledOverride: "active" }, + noOpReason: "already_unsettled", + expectedPayload: { reason: "user" }, + }, + { + method: "snooze", + command: "thread.snooze", + initial: activeState, + applied: { ...activeState, snoozedUntil: later, snoozedAt: iso }, + noOpReason: "already_snoozed_until_target", + expectedPayload: { snoozedUntil: later }, + }, + { + method: "unsnooze", + command: "thread.unsnooze", + initial: { ...activeState, snoozedUntil: later, snoozedAt: iso }, + applied: activeState, + noOpReason: "not_snoozed", + expectedPayload: { reason: "user" }, + }, + { + method: "updateMeta", + operation: "update_meta", + command: "thread.meta.update", + initial: activeState, + applied: { + ...activeState, + title: "renamed", + modelSelection: { instanceId: "claudeAgent", model: "claude-sonnet-4-5" }, + branch: "feat/lifecycle", + worktreePath: "/tmp/lifecycle", + }, + noOpReason: "metadata_unchanged", + expectedPayload: { + title: "renamed", + modelSelection: { instanceId: "claudeAgent", model: "claude-sonnet-4-5" }, + branch: "feat/lifecycle", + worktreePath: "/tmp/lifecycle", + }, + }, +]; + +describe("stock facade lifecycle commands", () => { + for (const entry of cases) { + test(`${entry.method} refuses an invalid scoped ref before any HTTP`, async () => { + const calls: string[] = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ state: () => entry.initial, calls }), + }); + + await expect(invoke(runtime, entry.method, { environmentId: " ", threadId: ref.threadId })) + .rejects.toMatchObject({ code: "identity_conflict" }); + expect(calls).toEqual([]); + runtime.close(); + }); + + test(`${entry.method} returns a typed no-op when the target state is already projected`, async () => { + let dispatches = 0; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => entry.applied, + dispatch: async () => { + dispatches += 1; + return { sequence: 2 }; + }, + }), + }); + + await expect(invoke(runtime, entry.method)).resolves.toMatchObject({ + kind: "no_op", + operation: entry.operation ?? entry.method, + reason: entry.noOpReason, + agentRef: ref, + }); + expect(dispatches).toBe(0); + runtime.close(); + }); + + test(`${entry.method} returns typed pending after acceptance without projection`, async () => { + const commands: Array>> = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => entry.initial, + dispatch: async (command) => { + commands.push(command); + return { sequence: 2 }; + }, + }), + id: () => `${entry.method}-command`, + }); + + await expect(invoke(runtime, entry.method)).resolves.toMatchObject({ + kind: "pending", + operation: entry.operation ?? entry.method, + reason: "projection_pending", + agentRef: ref, + receipt: { + commandId: `${entry.method}-command`, + acceptedSequence: 2, + }, + }); + expect(commands).toEqual([{ + type: entry.command, + commandId: `${entry.method}-command`, + threadId: ref.threadId, + ...(entry.expectedPayload ?? {}), + }]); + runtime.close(); + }); + } + + test("confirms all seven operations only after aligned projection evidence", async () => { + for (const entry of cases) { + let state = entry.initial; + let sequence = 1; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => state, + sequence: () => sequence, + dispatch: async () => { + state = entry.applied; + sequence = 2; + return { sequence: 2 }; + }, + }), + id: () => `${entry.method}-command`, + }); + + await expect(invoke(runtime, entry.method, ref, { timeoutMs: 1_000 })).resolves.toMatchObject({ + kind: "applied", + operation: entry.operation ?? entry.method, + receipt: { acceptedSequence: 2, observedSequence: 2 }, + }); + runtime.close(); + } + }); + + test("validates snooze and metadata inputs before any HTTP", async () => { + const calls: string[] = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ state: () => activeState, calls }), + }); + + await expect(runtime.snooze(ref, "not-a-date")).rejects.toMatchObject({ + code: "protocol_mismatch", + evidence: { field: "snoozedUntil" }, + }); + await expect(runtime.updateMeta(ref, {})).rejects.toMatchObject({ + code: "protocol_mismatch", + evidence: { field: "fields" }, + }); + await expect(runtime.updateMeta(ref, { regenerateTitle: true } as ThreadMetaFields)) + .rejects.toMatchObject({ code: "protocol_mismatch", evidence: { field: "fields.regenerateTitle" } }); + expect(calls).toEqual([]); + runtime.close(); + }); + + test("never emits thread.delete", async () => { + const commands: Array>> = []; + for (const entry of cases) { + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => entry.initial, + dispatch: async (command) => { + commands.push(command); + return { sequence: 2 }; + }, + }), + }); + await invoke(runtime, entry.method); + runtime.close(); + } + expect(commands).toHaveLength(cases.length); + expect(commands.every((command) => command.type !== "thread.delete")).toBeTrue(); + }); + + test("derives an expired snooze as not snoozed and performs no dispatch", async () => { + let dispatches = 0; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => ({ ...activeState, snoozedUntil: "2026-08-03T10:00:00.000Z", snoozedAt: iso }), + dispatch: async () => { + dispatches += 1; + return { sequence: 2 }; + }, + }), + now: () => iso, + }); + + await expect(runtime.unsnooze(ref)).resolves.toMatchObject({ + kind: "no_op", + operation: "unsnooze", + reason: "not_snoozed", + }); + expect(dispatches).toBe(0); + runtime.close(); + }); + + test("reconciles an ambiguous lifecycle dispatch before one byte-identical retry", async () => { + let state = activeState; + let sequence = 1; + const commands: Array>> = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => state, + sequence: () => sequence, + dispatch: async (command) => { + commands.push(command); + if (commands.length === 1) throw new StockT3HttpError("transport_unavailable", null); + state = { ...activeState, archivedAt: later }; + sequence = 2; + return { sequence: 2 }; + }, + }), + id: () => "archive-command", + }); + + await expect(runtime.archive(ref, { timeoutMs: 1_000 })).resolves.toMatchObject({ + kind: "applied", + operation: "archive", + receipt: { + commandId: "archive-command", + retryState: "identical_retry_accepted", + }, + }); + expect(commands).toHaveLength(2); + expect(commands[1]).toEqual(commands[0]); + runtime.close(); + }); + + test("caps ambiguous lifecycle dispatch at one byte-identical retry and returns pending", async () => { + const commands: Array>> = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => activeState, + dispatch: async (command) => { + commands.push(command); + throw new StockT3HttpError("transport_unavailable", null); + }, + }), + id: () => "archive-command", + }); + + await expect(runtime.archive(ref, { timeoutMs: 1_000, maxReconciliationReads: 1 })) + .resolves.toMatchObject({ + kind: "pending", + operation: "archive", + receipt: { + commandId: "archive-command", + acceptedSequence: null, + retryState: "identical_retry_outcome_unknown", + }, + }); + expect(commands).toHaveLength(2); + expect(commands[1]).toEqual(commands[0]); + runtime.close(); + }); + + test("treats a lagging decoded read model as pending instead of a protocol failure", async () => { + const runtime = createStockT3NativeRuntime({ + client: { + ...runtimeClient({ state: () => activeState }), + getSnapshot: async (boundary = {}) => decodeReadModelSnapshot( + readModel(1, activeState), + { minimumSequence: boundary.minimumSequence }, + ), + dispatch: async () => ({ sequence: 2 }), + }, + }); + + await expect(runtime.archive(ref, { timeoutMs: 1_000, maxReconciliationReads: 1 })) + .resolves.toMatchObject({ + kind: "pending", + operation: "archive", + receipt: { acceptedSequence: 2, observedSequence: 1 }, + }); + runtime.close(); + }); + + test("fails closed when archive cannot access the stock full snapshot", async () => { + const { getSnapshot: _getSnapshot, ...withoutSnapshot } = runtimeClient({ + state: () => activeState, + }); + const runtime = createStockT3NativeRuntime({ client: withoutSnapshot }); + + await expect(runtime.archive(ref)).rejects.toMatchObject({ + code: "protocol_mismatch", + evidence: { reason: "full_snapshot_unavailable" }, + }); + runtime.close(); + }); + + test("refuses archive for a soft-deleted full-snapshot tombstone", async () => { + let dispatches = 0; + const runtime = createStockT3NativeRuntime({ + client: { + ...runtimeClient({ state: () => activeState }), + getSnapshot: async () => ({ + ...readModel(1, activeState), + threads: [{ ...detail(1, activeState).thread, deletedAt: iso }], + }), + dispatch: async () => { + dispatches += 1; + return { sequence: 2 }; + }, + }, + }); + + await expect(runtime.archive(ref)).rejects.toMatchObject({ + code: "identity_conflict", + evidence: { reason: "thread_not_found", threadId: ref.threadId }, + }); + expect(dispatches).toBe(0); + runtime.close(); + }); + + test("confirms archive from stock's full snapshot after active shell/detail disappearance", async () => { + let archived = false; + let sequence = 1; + const client: StockT3RuntimeClient = { + getDescriptor: async () => descriptor(), + getSnapshot: async () => readModel( + sequence, + archived ? { ...activeState, archivedAt: later } : activeState, + ), + getShell: async () => archived + ? { ...shell(sequence, activeState), threads: [] } + : shell(sequence, activeState), + getThread: async () => archived ? undefined : detail(sequence, activeState), + dispatch: async () => { + archived = true; + sequence = 2; + return { sequence: 2 }; + }, + }; + const runtime = createStockT3NativeRuntime({ + client, + id: () => "archive-command", + }); + await expect(runtime.archive(ref, { timeoutMs: 1_000 })).resolves.toMatchObject({ + kind: "applied", + operation: "archive", + receipt: { acceptedSequence: 2, observedSequence: 2 }, + snapshot: { thread: { archivedAt: later } }, + }); + runtime.close(); + }); + + test("preflights and confirms unarchive through the full snapshot", async () => { + let archived = true; + let sequence = 1; + const client: StockT3RuntimeClient = { + getDescriptor: async () => descriptor(), + getSnapshot: async () => readModel( + sequence, + archived ? { ...activeState, archivedAt: iso } : activeState, + ), + getShell: async () => archived + ? { ...shell(sequence, activeState), threads: [] } + : shell(sequence, activeState), + getThread: async () => archived ? undefined : detail(sequence, activeState), + dispatch: async () => { + archived = false; + sequence = 2; + return { sequence: 2 }; + }, + }; + const runtime = createStockT3NativeRuntime({ + client, + id: () => "unarchive-command", + }); + + await expect(runtime.unarchive(ref, { timeoutMs: 1_000 })).resolves.toMatchObject({ + kind: "applied", + operation: "unarchive", + receipt: { acceptedSequence: 2, observedSequence: 2 }, + snapshot: { thread: { archivedAt: null } }, + }); + runtime.close(); + }); + + test("dispatches unsettle for an auto-settled thread without an explicit settled override", async () => { + let state = activeState; + let sequence = 1; + let dispatches = 0; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => state, + sequence: () => sequence, + dispatch: async () => { + dispatches += 1; + state = { ...activeState, settledOverride: "active" }; + sequence = 2; + return { sequence: 2 }; + }, + }), + }); + + await expect(runtime.unsettle(ref, { timeoutMs: 1_000 })).resolves.toMatchObject({ + kind: "applied", + operation: "unsettle", + }); + expect(dispatches).toBe(1); + runtime.close(); + }); + + test("refuses settle for a running thread and snooze for pending user work", async () => { + let dispatches = 0; + const baseClient = runtimeClient({ + state: () => activeState, + dispatch: async () => { + dispatches += 1; + return { sequence: 2 }; + }, + }); + const running = createStockT3NativeRuntime({ + client: { + ...baseClient, + getShell: async () => ({ + ...shell(1, activeState), + threads: [{ + ...thread(activeState), + session: { + threadId: ref.threadId, + status: "running", + providerName: "claudeAgent", + activeTurnId: "turn-1", + lastError: null, + updatedAt: iso, + }, + }], + }), + }, + }); + await expect(running.settle(ref)).rejects.toMatchObject({ + code: "command_rejected", + evidence: { reason: "thread_not_settleable" }, + }); + running.close(); + + const pending = createStockT3NativeRuntime({ + client: { + ...baseClient, + getShell: async () => ({ + ...shell(1, activeState), + threads: [{ ...thread(activeState), hasPendingUserInput: true }], + }), + }, + now: () => iso, + }); + await expect(pending.snooze(ref, later)).rejects.toMatchObject({ + code: "command_rejected", + evidence: { reason: "thread_not_snoozable" }, + }); + pending.close(); + expect(dispatches).toBe(0); + }); + + test("refuses settle when only the detail projection reports a running session", async () => { + let dispatches = 0; + const terminalDetail = detail(1, activeState); + const runtime = createStockT3NativeRuntime({ + client: { + ...runtimeClient({ state: () => activeState }), + getThread: async () => ({ + ...terminalDetail, + thread: { + ...terminalDetail.thread, + session: { + threadId: ref.threadId, + status: "running", + providerName: "claudeAgent", + activeTurnId: "turn-1", + lastError: null, + updatedAt: iso, + }, + }, + }), + dispatch: async () => { + dispatches += 1; + return { sequence: 2 }; + }, + }, + }); + + await expect(runtime.settle(ref, operation)).rejects.toMatchObject({ + code: "command_rejected", + evidence: { reason: "thread_not_settleable" }, + }); + expect(dispatches).toBe(0); + runtime.close(); + }); + + test("fails closed when required lifecycle projection fields are absent", async () => { + const projected = thread(activeState); + const { settledAt: _settledAt, ...withoutSettledAt } = projected; + let dispatches = 0; + const runtime = createStockT3NativeRuntime({ + client: { + ...runtimeClient({ state: () => activeState }), + getShell: async () => ({ + ...shell(1, activeState), + threads: [withoutSettledAt as StockThreadShell], + }), + dispatch: async () => { + dispatches += 1; + return { sequence: 2 }; + }, + }, + }); + + await expect(runtime.settle(ref)).rejects.toMatchObject({ + code: "protocol_mismatch", + evidence: { reason: "lifecycle_projection_unavailable", field: "settledAt" }, + }); + expect(dispatches).toBe(0); + runtime.close(); + }); + + test("rejects null metadata strings and past snoozes locally", async () => { + const calls: string[] = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ state: () => activeState, calls }), + now: () => iso, + }); + + for (const fields of [ + { title: null }, + { modelSelection: { instanceId: null, model: "claude-opus-5" } }, + { modelSelection: { instanceId: "claudeAgent", model: null } }, + ]) { + await expect(runtime.updateMeta( + ref, + fields as unknown as ThreadMetaFields, + { timeoutMs: 100, maxReconciliationReads: 1 }, + )) + .rejects.toMatchObject({ code: "protocol_mismatch" }); + } + await expect(runtime.snooze(ref, iso)).rejects.toMatchObject({ + code: "command_rejected", + evidence: { reason: "snooze_not_in_future" }, + }); + expect(calls).toEqual([]); + runtime.close(); + }); + + test("rejects an unavailable metadata model before any HTTP", async () => { + const calls: string[] = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ state: () => activeState, calls }), + }); + + await expect(runtime.updateMeta(ref, { + modelSelection: { instanceId: "claudeAgent", model: "missing-model" }, + }, operation)).rejects.toMatchObject({ + code: "model_unavailable", + evidence: { reason: "model_slug_unavailable" }, + }); + expect(calls).toEqual([]); + runtime.close(); + }); + + test("confirms equivalent model options independent of object key order", async () => { + const requested = { + instanceId: "claudeAgent", + model: "claude-opus-5", + options: [{ id: "thinking", value: { enabled: true, budget: 8_000 } }], + }; + const projected = { + instanceId: "claudeAgent", + model: "claude-opus-5", + options: [{ value: { budget: 8_000, enabled: true }, id: "thinking" }], + }; + let state: LifecycleState = activeState; + let sequence = 1; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ + state: () => state, + sequence: () => sequence, + dispatch: async () => { + state = { ...activeState, modelSelection: projected }; + sequence = 2; + return { sequence: 2 }; + }, + }), + }); + + await expect(runtime.updateMeta(ref, { modelSelection: requested }, { timeoutMs: 1_000 })) + .resolves.toMatchObject({ kind: "applied", operation: "update_meta" }); + runtime.close(); + }); + + test("paces transient full-snapshot failures before the next reconciliation read", async () => { + let reads = 0; + let ambiguousAt = 0; + let retryAt = 0; + const client: StockT3RuntimeClient = { + ...runtimeClient({ state: () => activeState }), + getSnapshot: async () => { + reads += 1; + if (reads === 1) return readModel(1, activeState); + if (reads === 2) { + ambiguousAt = performance.now(); + throw new StockT3HttpError("transport_unavailable", null); + } + retryAt = performance.now(); + return readModel(1, activeState); + }, + dispatch: async () => ({ sequence: 2 }), + }; + const runtime = createStockT3NativeRuntime({ client }); + + await expect(runtime.archive(ref, { timeoutMs: 1_000, maxReconciliationReads: 2 })) + .resolves.toMatchObject({ kind: "pending", operation: "archive" }); + expect(reads).toBe(3); + expect(retryAt - ambiguousAt).toBeGreaterThanOrEqual(20); + runtime.close(); + }); + + test("restores terminal overlay state when a lifecycle result remains pending", async () => { + const overlay = createWorkerOverlay({ maxWorkers: 1 }); + overlay.attach(ref, { role: "worker", parentRef: null }); + recordWorkerTerminalState(overlay, ref, true); + const pending = { + kind: "pending" as const, + operation: "archive" as const, + reason: "projection_pending" as const, + agentRef: ref, + receipt: { + operation: "archive" as const, + commandId: "archive-command", + acceptedSequence: 2, + observedSequence: 1, + retryState: "not_needed" as const, + }, + snapshot: detail(1, activeState), + }; + const facade = createStockT3Facade({ + archive: async () => pending, + } as unknown as T3NativeRuntime, { overlay }); + + await expect(facade.archive(ref)).resolves.toEqual(pending); + const replacement = { environmentId: ref.environmentId, threadId: "thread-2" }; + expect(() => overlay.attach(replacement, { role: "worker", parentRef: null })).not.toThrow(); + expect(overlay.getWorker(replacement)?.ref).toEqual(replacement); + }); + + test("keeps overlay capacity occupied until concurrent pending lifecycle mutations finish", async () => { + const overlay = createWorkerOverlay({ maxWorkers: 1 }); + overlay.attach(ref, { role: "worker", parentRef: null }); + recordWorkerTerminalState(overlay, ref, true); + const pending = { + kind: "pending" as const, + operation: "archive" as const, + reason: "projection_pending" as const, + agentRef: ref, + receipt: { + operation: "archive" as const, + commandId: "archive-command", + acceptedSequence: 2, + observedSequence: 1, + retryState: "not_needed" as const, + }, + snapshot: detail(1, activeState), + }; + const releases: Array<(result: typeof pending) => void> = []; + const facade = createStockT3Facade({ + archive: async () => new Promise((resolve) => releases.push(resolve)), + } as unknown as T3NativeRuntime, { overlay }); + const first = facade.archive(ref); + const second = facade.archive(ref); + const replacement = { environmentId: ref.environmentId, threadId: "thread-2" }; + + releases[0]!(pending); + await first; + let capacityError: unknown; + try { + overlay.attach(replacement, { role: "worker", parentRef: null }); + } catch (error) { + capacityError = error; + } + expect(capacityError).toMatchObject({ code: "overlay_capacity_exceeded" }); + + releases[1]!(pending); + await second; + expect(() => overlay.attach(replacement, { role: "worker", parentRef: null })).not.toThrow(); + }); + + test("defers a concurrent wait terminal release until the lifecycle group finishes", async () => { + const overlay = createWorkerOverlay({ maxWorkers: 1 }); + overlay.attach(ref, { role: "worker", parentRef: null }); + const pending = { + kind: "pending" as const, + operation: "archive" as const, + reason: "projection_pending" as const, + agentRef: ref, + receipt: { + operation: "archive" as const, + commandId: "archive-command", + acceptedSequence: 2, + observedSequence: 1, + retryState: "not_needed" as const, + }, + snapshot: detail(1, activeState), + }; + let releaseArchive!: (result: typeof pending) => void; + const terminalBase = detail(3, activeState); + const terminalSnapshot: ThreadDetailSnapshot = { + ...terminalBase, + thread: { + ...terminalBase.thread, + latestTurn: { + turnId: "turn-1", + state: "completed", + requestedAt: iso, + startedAt: iso, + completedAt: iso, + assistantMessageId: "assistant-1", + }, + }, + }; + const receipt = { + agentRef: ref, + leaseId: "lease-1", + commandId: "turn-command", + messageId: "message-1", + acceptedSequence: 2, + observedSequence: 2, + leaseExpiresAt: Date.now() + 1_000, + leaseState: "active" as const, + }; + const facade = createStockT3Facade({ + archive: async () => new Promise((resolve) => { + releaseArchive = resolve; + }), + wait: async () => terminalSnapshot, + } as unknown as T3NativeRuntime, { overlay }); + const archive = facade.archive(ref); + await facade.wait(receipt); + const replacement = { environmentId: ref.environmentId, threadId: "thread-2" }; + + let capacityError: unknown; + try { + overlay.attach(replacement, { role: "worker", parentRef: null }); + } catch (error) { + capacityError = error; + } + expect(capacityError).toMatchObject({ code: "overlay_capacity_exceeded" }); + + releaseArchive(pending); + await expect(archive).resolves.toEqual(pending); + expect(() => overlay.attach(replacement, { role: "worker", parentRef: null })).not.toThrow(); + }); +}); + +describe("stock facade lifecycle MCP parity", () => { + test("publishes exact schemas and routes all seven methods without drift", async () => { + const calls: Array<{ method: string; args: readonly unknown[] }> = []; + const snapshot = detail(2, activeState); + const fake = { + archive: async (...args: readonly unknown[]) => (calls.push({ method: "archive", args }), snapshot), + unarchive: async (...args: readonly unknown[]) => (calls.push({ method: "unarchive", args }), snapshot), + settle: async (...args: readonly unknown[]) => (calls.push({ method: "settle", args }), snapshot), + unsettle: async (...args: readonly unknown[]) => (calls.push({ method: "unsettle", args }), snapshot), + snooze: async (...args: readonly unknown[]) => (calls.push({ method: "snooze", args }), snapshot), + unsnooze: async (...args: readonly unknown[]) => (calls.push({ method: "unsnooze", args }), snapshot), + updateMeta: async (...args: readonly unknown[]) => (calls.push({ method: "updateMeta", args }), snapshot), + } as unknown as ReturnType; + const mcp = createStockT3McpFacade(fake); + const names = mcp.listTools().map((tool) => tool.name); + expect(names).toEqual(expect.arrayContaining([ + "archive", "unarchive", "settle", "unsettle", "snooze", "unsnooze", "updateMeta", + ])); + + for (const name of ["archive", "unarchive", "settle", "unsettle", "unsnooze"] as const) { + const result = await mcp.callTool(name, { ref, operation }); + expect(result.isError).toBeFalse(); + } + expect((await mcp.callTool("snooze", { ref, until: later, operation })).isError).toBeFalse(); + const fields: ThreadMetaFields = { title: "renamed", branch: null }; + expect((await mcp.callTool("updateMeta", { ref, fields, operation })).isError).toBeFalse(); + expect(calls).toEqual([ + { method: "archive", args: [ref, operation] }, + { method: "unarchive", args: [ref, operation] }, + { method: "settle", args: [ref, operation] }, + { method: "unsettle", args: [ref, operation] }, + { method: "unsnooze", args: [ref, operation] }, + { method: "snooze", args: [ref, later, operation] }, + { method: "updateMeta", args: [ref, fields, operation] }, + ]); + + const updateMeta = mcp.listTools().find((tool) => tool.name === "updateMeta")!; + const fieldsSchema = updateMeta.inputSchema.properties.fields as { + additionalProperties: boolean; + properties: Readonly>; + }; + expect(fieldsSchema.additionalProperties).toBeFalse(); + expect(Object.keys(fieldsSchema.properties)).toEqual([ + "title", "modelSelection", "branch", "worktreePath", + ]); + }); + + test("keeps lifecycle validation evidence identical across direct and MCP with zero HTTP", async () => { + const calls: string[] = []; + const runtime = createStockT3NativeRuntime({ + client: runtimeClient({ state: () => activeState, calls }), + }); + const facade = createStockT3Facade(runtime); + const mcp = createStockT3McpFacade(facade); + + const direct = await runtime.snooze(ref, "invalid").catch((error) => error); + expect(direct).toBeInstanceOf(StockRuntimeError); + const throughMcp = await mcp.callTool("snooze", { ref, until: "invalid" }); + expect(throughMcp).toMatchObject({ + isError: true, + structuredContent: { + error: { type: "stock_runtime", code: direct.code, evidence: direct.evidence }, + }, + }); + + const forbidden = await mcp.callTool("updateMeta", { + ref, + fields: { regenerateTitle: true }, + }); + expect(forbidden).toMatchObject({ + isError: true, + structuredContent: { + error: { + type: "stock_runtime", + code: "protocol_mismatch", + evidence: { field: "fields.regenerateTitle" }, + }, + }, + }); + expect(calls).toEqual([]); + runtime.close(); + }); +}); diff --git a/test/stock-t3-contracts.test.ts b/test/stock-t3-contracts.test.ts index b45c6f8..8c627e5 100644 --- a/test/stock-t3-contracts.test.ts +++ b/test/stock-t3-contracts.test.ts @@ -5,6 +5,7 @@ import { decodeDescriptor, decodeDispatchError, decodeDispatchResult, + decodeReadModelSnapshot, decodeShellSnapshot, decodeThreadDetailSnapshot, } from "../src/stockT3Contracts"; @@ -140,6 +141,75 @@ describe("stock T3 narrow contracts", () => { expect(detail.thread.messages[1]?.turnId).toBe("turn-1"); }); + test("decodes archived threads from the full read model", () => { + const snapshot = decodeReadModelSnapshot({ + snapshotSequence: 12, + projects: [], + threads: [{ + ...threadBase, + archivedAt: iso, + settledOverride: "settled", + settledAt: iso, + snoozedUntil: iso, + snoozedAt: iso, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + }], + updatedAt: iso, + }); + + expect(snapshot.snapshotSequence).toBe(12); + expect(snapshot.threads).toHaveLength(1); + expect(snapshot.threads[0]).toMatchObject({ + id: "thread-1", + archivedAt: iso, + settledOverride: "settled", + settledAt: iso, + snoozedUntil: iso, + snoozedAt: iso, + }); + }); + + test("preserves absence of lifecycle projections instead of inventing null state", () => { + const { + archivedAt: _archivedAt, + settledOverride: _settledOverride, + settledAt: _settledAt, + ...withoutLifecycle + } = threadBase; + const snapshot = decodeThreadDetailSnapshot({ + snapshotSequence: 11, + thread: { + ...withoutLifecycle, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + }, + }); + + expect(Object.hasOwn(snapshot.thread, "archivedAt")).toBeFalse(); + expect(Object.hasOwn(snapshot.thread, "settledOverride")).toBeFalse(); + expect(Object.hasOwn(snapshot.thread, "settledAt")).toBeFalse(); + const readModel = decodeReadModelSnapshot({ + snapshotSequence: 11, + projects: [], + threads: [{ + ...withoutLifecycle, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + }], + updatedAt: iso, + }); + expect(Object.hasOwn(readModel.threads[0]!, "archivedAt")).toBeFalse(); + expect(Object.hasOwn(readModel.threads[0]!, "settledOverride")).toBeFalse(); + expect(Object.hasOwn(readModel.threads[0]!, "settledAt")).toBeFalse(); + }); + test("rejects negative or regressing sequence anchors", () => { expect(() => decodeDispatchResult({ sequence: -1 })).toThrow(ProtocolMismatchError); expect(() => diff --git a/test/stock-t3-http-client.test.ts b/test/stock-t3-http-client.test.ts index 920375a..a56f23f 100644 --- a/test/stock-t3-http-client.test.ts +++ b/test/stock-t3-http-client.test.ts @@ -36,7 +36,7 @@ describe("stock T3 HTTP client", () => { }); expect(JSON.stringify(client.observations())).not.toContain("must-not-appear"); }); - test("uses only descriptor, shell, detail, dispatch, and token HTTP routes", async () => { + test("uses only descriptor, shell, full snapshot, detail, dispatch, and token HTTP routes", async () => { const requests: Request[] = []; const fetch = async (input: string | URL | Request, init?: RequestInit) => { const request = new Request(input, init); @@ -52,6 +52,14 @@ describe("stock T3 HTTP client", () => { updatedAt: "2026-07-31T18:00:00.000Z", }); } + if (request.url.endsWith("/api/orchestration/snapshot")) { + return Response.json({ + snapshotSequence: 7, + projects: [], + threads: [], + updatedAt: "2026-07-31T18:00:00.000Z", + }); + } if (request.url.endsWith("/api/orchestration/threads/thread%2Fone")) { return new Response( JSON.stringify({ code: "not_found", reason: "thread_not_found", traceId: "x" }), @@ -74,6 +82,7 @@ describe("stock T3 HTTP client", () => { await client.getDescriptor(); await client.getShell(); + expect(await client.getSnapshot()).toMatchObject({ snapshotSequence: 7, threads: [] }); expect(await client.getThread("thread/one")).toBeUndefined(); expect(await client.dispatch({ type: "thread.delete", commandId: "c", threadId: "t" })).toEqual({ sequence: 9 }); expect(await client.exchangeToken({ grantType: "pairing", credential: "bootstrap-secret" })).toEqual({ @@ -85,12 +94,14 @@ describe("stock T3 HTTP client", () => { expect(requests.map((request) => new URL(request.url).pathname)).toEqual([ "/.well-known/t3/environment", "/api/orchestration/shell", + "/api/orchestration/snapshot", "/api/orchestration/threads/thread%2Fone", "/api/orchestration/dispatch", "/oauth/token", ]); expect(requests[0]?.headers.get("authorization")).toBeNull(); expect(requests[1]?.headers.get("authorization")).toBe("Bearer bearer-secret"); + expect(requests[2]?.headers.get("authorization")).toBe("Bearer bearer-secret"); }); test("preserves a reverse-proxy path prefix on every stock endpoint", async () => {