diff --git a/changes/unreleased/4374.md b/changes/unreleased/4374.md new file mode 100644 index 0000000000..f309dcabdd --- /dev/null +++ b/changes/unreleased/4374.md @@ -0,0 +1,59 @@ + +## 2026-10-26 - Retain accepted terminals until their owned transcript append (part of #4272) + +### What changed + +- `packages/coding-agent/src/core/agent-session.ts`: carry producer identity through transcript refusal, attach recovery owners to message/custom appends, and resolve a committed lifecycle hint before terminal publication. +- `packages/coding-agent/src/core/session-manager.ts`: accept an internal recovery owner on owned-message and custom-message appends, recording its identity only in the committed transcript journal. +- `packages/coding-agent/src/core/session-recovery-identity.ts`: bind the existing S04a mint to a host-issued random epoch nonce, reset the producer at bootstrap, and preserve message identity for append-before-end lifecycle delivery. +- Re-setting the same recovery epoch preserves open assistant and message identities; only a new epoch clears them. +- `packages/coding-agent/src/core/lifecycle-delivery.ts` and `packages/coding-agent/src/core/session-transcript-journal.ts`: carry the owner on the committed append, without adding it to persisted entries. + +### Why + +- `packages/coding-agent/src/core/agent-session.ts` publishes terminal content before persistence; a failed append must not erase that content from recovery. +- `packages/coding-agent/src/core/session-manager.ts` is the commit boundary. Joining by its explicit owner works in either publication order and never relies on adjacency. +- `packages/coding-agent/src/core/session-recovery-identity.ts`, `packages/coding-agent/src/core/lifecycle-delivery.ts`, and `packages/coding-agent/src/core/session-transcript-journal.ts` preserve ownership across asynchronous publication and reject carriers from retired epochs. + +### Why an extension could not handle it + +- `packages/coding-agent/src/core/agent-session.ts` and `packages/coding-agent/src/core/session-manager.ts` own publication and persistence ordering below extension delivery. +- `packages/coding-agent/src/core/session-recovery-identity.ts`, `packages/coding-agent/src/core/lifecycle-delivery.ts`, and `packages/coding-agent/src/core/session-transcript-journal.ts` implement an internal, non-persisted identity channel that extensions must not serialize. + +### Expected merge conflict zones + +- `packages/coding-agent/src/core/agent-session.ts`: message terminal persistence and idle custom-message publication. +- `packages/coding-agent/src/core/session-manager.ts`: `_appendEntry`, `appendOwnedMessage`, and `appendCustomMessageEntry`. +- `packages/coding-agent/src/core/session-recovery-identity.ts`, `packages/coding-agent/src/core/lifecycle-delivery.ts`, and `packages/coding-agent/src/core/session-transcript-journal.ts`: internal carrier and append-owner fields. + + +## 2026-10-26 - Epoch-bound pending and failed terminal recovery (part of #4272) + +### What changed + +- `packages/coding-agent/src/modes/rpc/compact-settled-messages.ts` retains pending/failed terminal payloads and both directions of the durable-message join in the S03 spillable collection. +- `packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts` captures terminal identity and payload synchronously, performs append joins in the history-acceptance step, and retires recovery storage on session replacement or close. +- `packages/coding-agent/src/modes/rpc/accepted-transcript.ts` exposes availability without allocating a frozen transcript root on every accepted append. +- `packages/coding-agent/src/modes/rpc/session-binding.ts` and `packages/coding-agent/src/modes/rpc/session-registry-open.ts` hand the host nonce to the in-process producer. +- `packages/coding-agent/src/modes/rpc/session-worker-client.ts`, `packages/coding-agent/src/modes/rpc/session-worker-protocol.ts`, and `packages/coding-agent/src/modes/rpc/session-worker.ts` carry the nonce in the private initialization/replacement handshake, before the producer resumes. + +### Why + +- `packages/coding-agent/src/modes/rpc/compact-settled-messages.ts` and `packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts` must retain a terminal even when neither the transcript nor its optional recovery spool can write it. +- `packages/coding-agent/src/modes/rpc/session-binding.ts` and `packages/coding-agent/src/modes/rpc/session-registry-open.ts` establish the same epoch authority for the in-process producer as the worker path. +- `packages/coding-agent/src/modes/rpc/session-worker-client.ts`, `packages/coding-agent/src/modes/rpc/session-worker-protocol.ts`, and `packages/coding-agent/src/modes/rpc/session-worker.ts` implement the approved RULING-24 epoch-bound carrier clarification without changing public JSONL. + +### Why an extension could not handle it + +- `packages/coding-agent/src/modes/rpc/compact-settled-messages.ts`, `packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts`, `packages/coding-agent/src/modes/rpc/session-binding.ts`, and `packages/coding-agent/src/modes/rpc/session-registry-open.ts` own host acceptance and recovery roots below extensions. +- `packages/coding-agent/src/modes/rpc/session-worker-client.ts`, `packages/coding-agent/src/modes/rpc/session-worker-protocol.ts`, and `packages/coding-agent/src/modes/rpc/session-worker.ts` own the private worker bootstrap and output-credit boundary. + +### Expected merge conflict zones + +- `packages/coding-agent/src/modes/rpc/compact-settled-messages.ts`: new fork-only reducer. +- `packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts`, `packages/coding-agent/src/modes/rpc/session-binding.ts`, and `packages/coding-agent/src/modes/rpc/session-registry-open.ts`: epoch bootstrap and transcript acceptance. +- `packages/coding-agent/src/modes/rpc/session-worker-client.ts`, `packages/coding-agent/src/modes/rpc/session-worker-protocol.ts`, and `packages/coding-agent/src/modes/rpc/session-worker.ts`: private commit and replacement frames. + + +### Fixed +- The experimental, OFF-by-default compact recovery path retains terminal content when transcript persistence fails and rejects terminal carriers from retired session epochs (part of [#4272](https://github.com/code-yeongyu/senpi/issues/4272)). diff --git a/packages/agent/src/internal-events.ts b/packages/agent/src/internal-events.ts index dd67290784..caf8787618 100644 --- a/packages/agent/src/internal-events.ts +++ b/packages/agent/src/internal-events.ts @@ -1,3 +1,4 @@ +/** Internal to senpi packages; not a public API; may change without notice. */ import { type AssistantProgress, subscribeAssistantCapture } from "@earendil-works/pi-ai"; import type { AgentEvent, AgentLoopConfig } from "./types.ts"; diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 394a223ab0..2e6ad37202 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -112,7 +112,7 @@ import type { CredentialReplacement } from "./auth-storage.ts"; import { type BashResult, executeBashWithOperations } from "./bash-executor.ts"; import { envValue } from "./brand.ts"; import { compactUpdatesEnabled } from "./engine-socket-flag.ts"; -import { carryRecovery, SessionRecoveryIdentity } from "./session-recovery-identity.ts"; +import { carryRecovery, recoveryCarrier, SessionRecoveryIdentity } from "./session-recovery-identity.ts"; import type { BrowserEngine } from "./browser-engine.ts"; import { type ClientMessageIdentity, @@ -1532,7 +1532,7 @@ export class AgentSession { // prompt-owned run's prompt throws the first one once the queue settles; any other run reports it // as a continuation error when it happens. private readonly _transcriptWriteFailures = new TranscriptWriteFailures(); - private readonly _recoveryIdentity = compactUpdatesEnabled() ? new SessionRecoveryIdentity() : undefined; + private readonly _recoveryIdentity = compactUpdatesEnabled() ? new SessionRecoveryIdentity(undefined, this) : undefined; private _promptOwnsRun = false; // Preserve provenance across agent-core's conversion of our admission error // into an assistant error message. Matching provider text alone is not proof @@ -3956,6 +3956,15 @@ export class AgentSession { : event.type === "message_update" ? withResolvedToolName(event, (name) => this.resolveToolCallName(name)) : event; + let consumedEntry: string | undefined; + if (event.type === "message_end") { + try { + consumedEntry = this._lifecycleBinding.consumedEntry(event.message); + } catch (error) { + this._transcriptWriteFailures.record(event.message, error); + } + } + this._recoveryIdentity?.hint(event, consumedEntry); this._emit(carryRecovery(event, sessionEvent), event); if (event.type === "agent_end") { if (this._abortProvenance.takeLateUserJoin()) await this._emitSessionAbort(); @@ -3976,12 +3985,13 @@ export class AgentSession { if (event.message.role === "custom") { // Persist as CustomMessageEntry entryId = - this._lifecycleBinding.consumedEntry(event.message) ?? + consumedEntry ?? this.sessionManager.appendCustomMessageEntry( event.message.customType, event.message.content, event.message.display, event.message.details, + { recoveryOwner: recoveryCarrier(event) }, ); this._incrementMessageRevision(); this.externalAdmission.observePersisted(event.message); @@ -3995,6 +4005,7 @@ export class AgentSession { // Regular LLM message - persist as SessionMessageEntry entryId = this.sessionManager.appendOwnedMessage(event.message, { inputSource: this._userInputSources.get(event.message), + recoveryOwner: recoveryCarrier(event), }); this._emitEntryAppended(entryId); this._incrementMessageRevision(); @@ -4008,7 +4019,7 @@ export class AgentSession { this._takePersistWatcher(event.message)?.failed(error); this._sessionLogger.warn("transcript_write_failed", { role: event.message.role, error: errorMessage }); this._transcriptWriteFailures.record(event.message, error); - this._emit({ type: "transcript_write_failed", role: event.message.role, errorMessage }); + this._emit(carryRecovery(event, { type: "transcript_write_failed", role: event.message.role, errorMessage })); if (!this._promptOwnsRun) this._reportContinuationTranscriptFailure(); } if (entryId) this._entryIdsByMessage.set(event.message, entryId); @@ -8055,17 +8066,20 @@ export class AgentSession { private _appendCustomMessage(appMessage: CustomMessage): void { this.agent.state.messages.push(appMessage); + const start = { type: "message_start" as const, message: appMessage }; + this._recoveryIdentity?.observe(start); const entryId = this.sessionManager.appendCustomMessageEntry( appMessage.customType, appMessage.content, appMessage.display, appMessage.details, + { recoveryOwner: recoveryCarrier(start) }, ); this._emitEntryAppended(entryId); this._recordWrittenTicket(appMessage, entryId); this._takePersistWatcher(appMessage)?.persisted(entryId); this._incrementMessageRevision(); - this._emit({ type: "message_start", message: appMessage }); + this._emit(start); this._emit({ type: "message_end", message: appMessage }); } diff --git a/packages/coding-agent/src/core/lifecycle-delivery.ts b/packages/coding-agent/src/core/lifecycle-delivery.ts index c6a184e662..89fc3668da 100644 --- a/packages/coding-agent/src/core/lifecycle-delivery.ts +++ b/packages/coding-agent/src/core/lifecycle-delivery.ts @@ -6,6 +6,7 @@ import type { HistoryReducer } from "./session-history-reduce.ts"; import type { SessionLifecycleController } from "./session-lifecycle.ts"; import type { DeliveryTicketId } from "./session-lifecycle-types.ts"; import { isRecord } from "./work-custody-record.ts"; +import { recoveryCarrier } from "./session-recovery-identity.ts"; /** Keep delivery identities, not materialized conversation history, between engine turns. */ export const lifecycleDeliveryEntries: HistoryReducer> = { @@ -47,6 +48,7 @@ export function consumeLifecycleDelivery( message.content, message.display, message.details, + { recoveryOwner: recoveryCarrier(message) }, ); controller.obligations.consumed(ticketId, { runId, transcriptEntryId }); return transcriptEntryId; diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index 0da7d65342..9da7883351 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -33,6 +33,7 @@ import { EventLoopSlicer, mapInSlices } from "./event-loop-slicer.ts"; import { projectRetainedMessages } from "./extensions/builtin/compaction/retained-message-projection.ts"; import { resolveMovedPath } from "./extensions/builtin/moved-path-guard/resolve.ts"; import type { InputSource } from "./extensions/types.ts"; +import type { RecoveryCarrier } from "./session-recovery-identity.ts"; import type { ModelChangeOrigin, ModelChangeSource } from "./model-change-origin.ts"; import { sessionCwdMatcher } from "./moved-session-cwd.ts"; import type { RepositoryIdentity } from "./repository-identity.ts"; @@ -1923,7 +1924,7 @@ export class SessionManager { return bytes; } - private _appendEntry(entry: SessionEntry, shareable?: object): void { + private _appendEntry(entry: SessionEntry, shareable?: object, recoveryOwner?: RecoveryCarrier): void { const residentEntry = this.residentStore.externalize(entry); const bytes = this._persist(residentEntry); const committed = this._shareMessage(residentEntry, shareable); @@ -1937,6 +1938,7 @@ export class SessionManager { id: residentEntry.id, entry: serialized, persistence: this._transcriptPersistence(), + ...(recoveryOwner ? { owningMessageId: recoveryOwner.messageId, recoveryEpoch: recoveryOwner.epoch } : {}), }); } } @@ -2059,15 +2061,16 @@ export class SessionManager { */ appendOwnedMessage( message: Message | CustomMessage | BashExecutionMessage, - options?: { readonly inputSource?: InputSource | undefined }, + options?: { readonly inputSource?: InputSource | undefined; readonly recoveryOwner?: RecoveryCarrier }, ): string { - return this._appendMessage(message, true, options?.inputSource); + return this._appendMessage(message, true, options?.inputSource, options?.recoveryOwner); } private _appendMessage( message: Message | CustomMessage | BashExecutionMessage, owned: boolean, inputSource?: InputSource, + recoveryOwner?: RecoveryCarrier, ): string { const entry: SessionMessageEntry = { type: "message", @@ -2077,7 +2080,7 @@ export class SessionManager { message, ...(inputSource === undefined ? {} : { inputSource }), }; - this._appendEntry(entry, owned ? message : undefined); + this._appendEntry(entry, owned ? message : undefined, recoveryOwner); const order = this.entryOrdersById.get(entry.id); if (order !== undefined) { this.messageEntryPositions.set(message, { entryId: entry.id, order }); @@ -2341,6 +2344,7 @@ export class SessionManager { content: string | (TextContent | ImageContent)[], display: boolean, details?: T, + options?: { readonly recoveryOwner?: RecoveryCarrier }, ): string { const entry: CustomMessageEntry = { type: "custom_message", @@ -2352,7 +2356,7 @@ export class SessionManager { parentId: this.leafId, timestamp: new Date().toISOString(), }; - this._appendEntry(entry); + this._appendEntry(entry, undefined, options?.recoveryOwner); return entry.id; } diff --git a/packages/coding-agent/src/core/session-recovery-identity.ts b/packages/coding-agent/src/core/session-recovery-identity.ts index 40d8dbd201..96f1d05529 100644 --- a/packages/coding-agent/src/core/session-recovery-identity.ts +++ b/packages/coding-agent/src/core/session-recovery-identity.ts @@ -3,9 +3,18 @@ import { randomUUID } from "node:crypto"; /** The recovery identity of one session record; never a property on the record, so no byte changes (R3.C1.2). */ export interface RecoveryCarrier { readonly messageId: string; + /** Host-issued nonce, internal to the producer/lane handshake; never serialized with a public record. */ + readonly epoch: string; + readonly durableEntryId?: string; } const carriers = new WeakMap(); +const producers = new WeakMap(); + +/** Internal bootstrap, kept off AgentSession's public SDK surface. */ +export function setSessionRecoveryEpoch(session: object, epoch: string): void { + producers.get(session)?.reset(epoch); +} export function recoveryCarrier(record: object): RecoveryCarrier | undefined { return carriers.get(record); @@ -27,7 +36,26 @@ type MessageEvent = { readonly type: string; readonly message?: { readonly role? */ export class SessionRecoveryIdentity { private openAssistant: string | undefined; - private readonly byMessage = new WeakMap(); + private byMessage = new WeakMap(); + private epoch: string; + + constructor(epoch: string = randomUUID(), owner?: object) { + this.epoch = epoch; + if (owner) producers.set(owner, this); + } + + reset(epoch: string): void { + if (epoch === this.epoch) return; + this.epoch = epoch; + this.openAssistant = undefined; + this.byMessage = new WeakMap(); + } + + /** A committed lifecycle delivery can predate its terminal record. */ + hint(event: object, durableEntryId: string | undefined): void { + const carrier = carriers.get(event); + if (carrier && durableEntryId) carriers.set(event, { ...carrier, durableEntryId }); + } observe(event: MessageEvent): void { const message = event.message; @@ -41,7 +69,9 @@ export class SessionRecoveryIdentity { : (known ?? (event.type === "message_end" ? randomUUID() : undefined)); if (messageId === undefined) return; if (assistant) this.openAssistant = event.type === "message_end" ? undefined : messageId; - else this.byMessage.set(message, messageId); - carriers.set(event, { messageId }); + this.byMessage.set(message, messageId); + const carrier = { messageId, epoch: this.epoch }; + carriers.set(event, carrier); + carriers.set(message, carrier); } } diff --git a/packages/coding-agent/src/core/session-transcript-journal.ts b/packages/coding-agent/src/core/session-transcript-journal.ts index 578e361c1e..028b6fd9b0 100644 --- a/packages/coding-agent/src/core/session-transcript-journal.ts +++ b/packages/coding-agent/src/core/session-transcript-journal.ts @@ -39,6 +39,8 @@ export type TranscriptTransition = readonly id: string; readonly entry?: string; readonly persistence?: TranscriptPersistence; + readonly owningMessageId?: string; + readonly recoveryEpoch?: string; } | { readonly kind: "persisted"; readonly persistence: TranscriptPersistence } | { readonly kind: "leaf"; readonly leafId: string | null } diff --git a/packages/coding-agent/src/modes/rpc/accepted-transcript.ts b/packages/coding-agent/src/modes/rpc/accepted-transcript.ts index 207d8beb91..1d08844237 100644 --- a/packages/coding-agent/src/modes/rpc/accepted-transcript.ts +++ b/packages/coding-agent/src/modes/rpc/accepted-transcript.ts @@ -49,6 +49,10 @@ export class AcceptedTranscript { return this.version; } + get available(): boolean { + return this.unavailable === null; + } + apply(transition: TranscriptTransition): void { this.version += 1; if (transition.kind === "replace") { diff --git a/packages/coding-agent/src/modes/rpc/compact-settled-messages.ts b/packages/coding-agent/src/modes/rpc/compact-settled-messages.ts new file mode 100644 index 0000000000..45dbd74eed --- /dev/null +++ b/packages/coding-agent/src/modes/rpc/compact-settled-messages.ts @@ -0,0 +1,153 @@ +import type { RecoveryCarrier } from "../../core/session-recovery-identity.ts"; +import type { TranscriptTransition } from "../../core/session-transcript-journal.ts"; +import { CompactContentIndex, type FrozenIndex } from "./compact-content-index.ts"; +import { + COMPACT_LEAF_CODE_UNITS, + CompactContentStore, + type CompactContentStoreOptions, + type LeafId, +} from "./compact-content-store.ts"; + +export type SettledDiagnostic = + | "unknown_message" + | "duplicate_end" + | "retired_owner" + | "conflicting_owner" + | "unsupported_role"; + +export interface TerminalRow { + readonly kind: "terminal"; + readonly messageId: string; + readonly turnId: string | null; + readonly terminalSequence: number; + readonly role: string; + readonly runtimeItemKey: null; + readonly entryId: null; + readonly status: "pending" | "failed"; + readonly error: { readonly errorMessage: string } | null; + readonly payload: readonly LeafId[]; +} + +export interface OwnerRow { + readonly kind: "owner"; + readonly messageId: string; + readonly entryId: string; +} + +type Row = TerminalRow | OwnerRow | { readonly kind: "ended" }; +export type SettledRoot = FrozenIndex; + +/** One spillable collection owns payloads, end markers and both directions of the durable join. */ +export class CompactSettledMessages { + private readonly store: CompactContentStore; + private readonly rows: CompactContentIndex; + private readonly epoch: string; + private readonly diagnose: (reason: SettledDiagnostic) => void; + + constructor(epoch: string, options: CompactContentStoreOptions, diagnose: (reason: SettledDiagnostic) => void) { + this.epoch = epoch; + this.diagnose = diagnose; + this.store = new CompactContentStore(options); + this.rows = new CompactContentIndex(this.store, (row) => (row.kind === "terminal" ? row.payload : []), 128); + } + + valid(carrier: RecoveryCarrier | undefined): carrier is RecoveryCarrier { + return carrier !== undefined && carrier.epoch === this.epoch; + } + + end(record: object, carrier: RecoveryCarrier | undefined, sequence: number, turnId: string | null): void { + if (!this.valid(carrier)) return this.diagnose("unknown_message"); + const id = carrier.messageId; + if (this.rows.has(`end:${id}`)) return this.diagnose("duplicate_end"); + if (carrier.durableEntryId && !this.rows.has(`owner:${id}`)) { + if (this.rows.has(`entry:${carrier.durableEntryId}`)) return this.diagnose("conflicting_owner"); + this.join(id, carrier.durableEntryId); + } + this.rows.set(`end:${id}`, { kind: "ended" }); + if (this.rows.has(`owner:${id}`)) return; + const message: unknown = Reflect.get(record, "message"); + const role = String(Reflect.get(Object(message), "role")); + const payload: LeafId[] = []; + const text = JSON.stringify({ + terminal: message, + correlation: { + clientMessageId: Reflect.get(record, "clientMessageId"), + clientTurnId: Reflect.get(record, "clientTurnId"), + clientMessages: Reflect.get(record, "clientMessages"), + }, + }); + for (let offset = 0; offset < text.length; offset += COMPACT_LEAF_CODE_UNITS) + payload.push(this.store.createLeaf(text.slice(offset, offset + COMPACT_LEAF_CODE_UNITS))); + this.rows.set(`terminal:${id}`, { + kind: "terminal", + messageId: id, + turnId, + terminalSequence: sequence, + role, + runtimeItemKey: null, + entryId: null, + status: "pending", + error: null, + payload, + }); + for (const leaf of payload) this.store.adjust(leaf, -1, 0); + if (!["user", "assistant", "toolResult", "custom"].includes(role)) this.diagnose("unsupported_role"); + } + + fail(carrier: RecoveryCarrier | undefined, errorMessage: string): void { + if (!this.valid(carrier) || !this.rows.has(`terminal:${carrier.messageId}`)) + return this.diagnose("unknown_message"); + this.rows.patch(`terminal:${carrier.messageId}`, { status: "failed", error: { errorMessage } }); + } + + append(transition: Extract): void { + const id = transition.owningMessageId; + if (id === undefined) return; + if (transition.recoveryEpoch !== this.epoch) return this.diagnose("retired_owner"); + if (this.rows.has(`owner:${id}`) || this.rows.has(`entry:${transition.id}`)) + return this.diagnose("conflicting_owner"); + this.join(id, transition.id); + } + + private join(messageId: string, entryId: string): void { + const owner: OwnerRow = { kind: "owner", messageId, entryId }; + this.rows.set(`owner:${messageId}`, owner); + this.rows.set(`entry:${entryId}`, owner); + this.rows.delete(`terminal:${messageId}`); + } + + freeze(): SettledRoot { + return this.rows.freeze(); + } + + release(root: SettledRoot): void { + this.rows.release(root); + } + + async read(root: SettledRoot) { + const settledMessages: (Omit & { terminal: unknown; correlation: unknown })[] = + []; + const owners: OwnerRow[] = []; + for (const [key, row] of await this.rows.entries(root)) { + if (row.kind === "owner" && key.startsWith("owner:")) owners.push(row); + if (row.kind !== "terminal") continue; + const { payload, kind: _kind, ...fields } = row; + const bytes = (await Promise.all(payload.map((leaf) => this.store.readLeaf(leaf)))).join(""); + const value: { terminal: unknown; correlation: unknown } = JSON.parse(bytes); + settledMessages.push({ ...fields, ...value }); + } + return { settledMessages, owners }; + } + + storageStatus() { + return { ...this.store.status(), ...this.rows.status() }; + } + + idle(): Promise { + return this.store.idle(); + } + + retire(): Promise { + return this.store.retire(); + } +} diff --git a/packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts b/packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts index a73eaf0a35..674980c5b3 100644 --- a/packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts +++ b/packages/coding-agent/src/modes/rpc/session-acceptance-lane.ts @@ -1,8 +1,12 @@ import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import type { RecoveryCarrier } from "../../core/session-recovery-identity.ts"; import type { TranscriptTransition } from "../../core/session-transcript-journal.ts"; import { AcceptedTranscript, type TranscriptRoot } from "./accepted-transcript.ts"; import { correlatedRecord } from "./client-message-events.ts"; +import { CompactSettledMessages, type SettledDiagnostic, type SettledRoot } from "./compact-settled-messages.ts"; +import type { CompactContentStoreOptions } from "./compact-content-store.ts"; /** Where one accepted recovery record sits in its runtime incarnation (R3.C1.1). */ interface AcceptedRef { @@ -27,11 +31,14 @@ export interface FrozenAcceptedRoot { readonly turnId: string | null; readonly correlation: AcceptedCorrelation | null; readonly transcript: TranscriptRoot; + readonly settled: SettledRoot; } /** The incarnation's transcript as its manager journaled it up to bootstrap: base, unpersisted entries, leaf. */ export interface SessionBootstrap { readonly transcript: readonly TranscriptTransition[]; + /** Replacement handshake: the host fills this nonce before returning the worker's output credit. */ + readonly epochSignal?: SharedArrayBuffer; } /** Recovery kinds take a sequence; responses and every other record keep FIFO order with none. */ @@ -48,6 +55,7 @@ const SEQUENCED = new Set([ "tool_execution_end", "entry_appended", "session_abort", + "transcript_write_failed", ]); function field(record: object, key: string): unknown { @@ -76,6 +84,25 @@ export class SessionAcceptanceLane { private correlation: AcceptedCorrelation | null = null; private readonly live = new Set(); private retiredAs: "close" | "worker_failed" | undefined; + private nonce: string = randomUUID(); + private settled!: CompactSettledMessages; + private readonly rootOwners = new WeakMap(); + private cleanup: Promise = Promise.resolve(); + private readonly storage: CompactContentStoreOptions | undefined; + private readonly diagnose: (reason: SettledDiagnostic) => void; + + constructor( + storage?: CompactContentStoreOptions, + diagnose: (reason: SettledDiagnostic) => void = (reason) => + process.stderr.write(`senpi rpc acceptance: settled transition: ${reason}\n`), + ) { + this.storage = storage; + this.diagnose = diagnose; + } + + get recoveryEpoch(): string { + return this.nonce; + } get retired(): boolean { return this.retiredAs !== undefined; @@ -85,8 +112,19 @@ export class SessionAcceptanceLane { return this.live.size; } - bootstrap(input: SessionBootstrap): void { + bootstrap(input: SessionBootstrap, nonce: string = randomUUID()): void { if (this.retired) return; + if (this.settled) this.retireSettled(); + this.nonce = nonce; + if (input.epochSignal) Buffer.from(nonce, "ascii").copy(Buffer.from(input.epochSignal)); + this.settled = new CompactSettledMessages( + nonce, + { + ...this.storage, + dir: join(this.storage?.dir ?? tmpdir(), `senpi-settled-${randomUUID()}`), + }, + this.diagnose, + ); this.epoch = randomUUID(); this.sequence = 0; this.transcript = new AcceptedTranscript(); @@ -102,6 +140,7 @@ export class SessionAcceptanceLane { for (const transition of transitions) { this.sequence += 1; this.transcript.apply(transition); + if (transition.kind === "append" && this.transcript.available) this.settled.append(transition); } } @@ -113,8 +152,10 @@ export class SessionAcceptanceLane { if (type === "turn_start" || (this.turnId === null && turnScoped(type, record))) this.turnId = randomUUID(); const turnId = this.turnId; if (type === "turn_end" || type === "agent_end") this.turnId = null; - const messageId = type.startsWith("message_") ? (carrier?.messageId ?? null) : null; + const messageId = type.startsWith("message_") && this.settled.valid(carrier) ? carrier.messageId : null; if (type === "message_start") this.activeMessageId = messageId; + if (type === "message_end") this.settled.end(record, carrier, this.sequence, turnId); + if (type === "transcript_write_failed") this.settled.fail(carrier, String(field(record, "errorMessage"))); if (type === "message_end" && this.activeMessageId === messageId) this.activeMessageId = null; return { sessionEpoch: this.epoch, @@ -133,19 +174,42 @@ export class SessionAcceptanceLane { turnId: this.turnId, correlation: this.correlation, transcript: this.transcript.freeze(), + settled: this.settled.freeze(), }); + this.rootOwners.set(root, this.settled); this.live.add(root); return root; } release(root: FrozenAcceptedRoot): void { - this.live.delete(root); + if (this.live.delete(root)) this.rootOwners.get(root)?.release(root.settled); + } + + readSettled(root: FrozenAcceptedRoot) { + const owner = this.rootOwners.get(root); + if (this.retired || !owner || !this.live.has(root)) throw new Error("accepted root released"); + return owner.read(root.settled); + } + + storageStatus() { + return this.settled.storageStatus(); + } + + async idle(): Promise { + await this.cleanup; + await this.settled.idle(); + } + + private retireSettled(): void { + for (const root of this.live) this.release(root); + this.cleanup = Promise.all([this.cleanup, this.settled.retire()]).then(() => {}); + void this.cleanup.catch(() => this.diagnose("retired_owner")); } /** No record is accepted after retirement, and every root this lane published is released. */ retire(reason: "close" | "worker_failed"): void { this.retiredAs ??= reason; - this.live.clear(); + if (this.settled) this.retireSettled(); } /** A change of the outer client correlation is its own transition, sequenced just before its record (C1.1). */ diff --git a/packages/coding-agent/src/modes/rpc/session-binding.ts b/packages/coding-agent/src/modes/rpc/session-binding.ts index fbee04f577..5ee14f065a 100644 --- a/packages/coding-agent/src/modes/rpc/session-binding.ts +++ b/packages/coding-agent/src/modes/rpc/session-binding.ts @@ -1,6 +1,6 @@ import { bindToProviderScope, runWithProviderScope } from "@earendil-works/pi-ai/node/provider-scope"; import type { AgentSessionRuntime } from "../../core/agent-session-runtime.ts"; -import { recoveryCarrier } from "../../core/session-recovery-identity.ts"; +import { recoveryCarrier, setSessionRecoveryEpoch } from "../../core/session-recovery-identity.ts"; import { createRpcConnectionHandler, type RpcConnectionHandler, @@ -41,6 +41,7 @@ export async function createRpcSessionBinding( // stream already carries that transition; no extra subscription is needed. const toolSpans = createToolAttributionSpans(sessionId); const lane = entry.acceptance; + if (lane) setSessionRecoveryEpoch(entry.runtime.session, lane.recoveryEpoch); if (lane) writer.registerAcceptance(sessionId, lane); // The manager whose journal the lane follows; a replacement's bootstrap moves the lane to the new one. let journaled = lane ? entry.runtime.session.sessionManager : undefined; @@ -50,6 +51,7 @@ export async function createRpcSessionBinding( if (lane && "type" in record && record.type === "session_replaced") { journaled = entry.runtime?.session.sessionManager; lane.bootstrap({ transcript: journaled?.beginTranscriptJournal() ?? [] }); + if (entry.runtime) setSessionRecoveryEpoch(entry.runtime.session, lane.recoveryEpoch); } else if (lane) { lane.applyTranscript(journaled?.drainTranscript() ?? []); lane.accept(record, recoveryCarrier(record)); diff --git a/packages/coding-agent/src/modes/rpc/session-registry-open.ts b/packages/coding-agent/src/modes/rpc/session-registry-open.ts index cc588deee6..714066a5d1 100644 --- a/packages/coding-agent/src/modes/rpc/session-registry-open.ts +++ b/packages/coding-agent/src/modes/rpc/session-registry-open.ts @@ -9,6 +9,7 @@ import { assertRequestedModelResolved } from "./open-session-model.ts"; import { canonicalSessionPath } from "./session-path-key.ts"; import { SessionAcceptanceLane } from "./session-acceptance-lane.ts"; +import { setSessionRecoveryEpoch } from "../../core/session-recovery-identity.ts"; import { attachToOpenSession } from "./session-registry-attach.ts"; import { type PendingClaims, @@ -173,6 +174,7 @@ export async function openRegistrySession( if (compactUpdatesEnabled() && host.options.workerIsolate !== true) { entry.acceptance = new SessionAcceptanceLane(); entry.acceptance.bootstrap({ transcript: manager.beginTranscriptJournal() }); + setSessionRecoveryEpoch(entry.runtime.session, entry.acceptance.recoveryEpoch); } entry.state = "open"; settleOpen.resolve(); diff --git a/packages/coding-agent/src/modes/rpc/session-worker-client.ts b/packages/coding-agent/src/modes/rpc/session-worker-client.ts index 792fa783cd..5b31e21a06 100644 --- a/packages/coding-agent/src/modes/rpc/session-worker-client.ts +++ b/packages/coding-agent/src/modes/rpc/session-worker-client.ts @@ -157,12 +157,16 @@ export class SessionWorkerClient { } async commit(): Promise { - const result = await this.request({ type: "commit", ...(this.acceptance ? { acceptance: true } : {}) }); + const nonce = randomUUID(); + const result = await this.request({ + type: "commit", + ...(this.acceptance ? { acceptance: true, recoveryEpoch: nonce } : {}), + }); if (result.type !== "ready") throw new Error("Invalid worker commit response"); if (this.acceptance) { // Accepted before the open resolves, so the entry is admitted only after its baseline (R3.C1.1). if (!result.bootstrap) throw new Error("Invalid worker commit response"); - this.acceptance.bootstrap(result.bootstrap); + this.acceptance.bootstrap(result.bootstrap, nonce); } this.snapshot = result.snapshot; return result.snapshot; diff --git a/packages/coding-agent/src/modes/rpc/session-worker-protocol.ts b/packages/coding-agent/src/modes/rpc/session-worker-protocol.ts index 28532a35f9..ee8505bb6b 100644 --- a/packages/coding-agent/src/modes/rpc/session-worker-protocol.ts +++ b/packages/coding-agent/src/modes/rpc/session-worker-protocol.ts @@ -62,7 +62,7 @@ export type HostToSessionWorker = subscriptionLimits: readonly MirroredLimits[]; } /** `acceptance`: the host owns an acceptance lane, so the isolate sends a bootstrap and ordered output frames. */ - | { type: "commit"; request: number; acceptance?: true } + | { type: "commit"; request: number; acceptance?: true; recoveryEpoch?: string } | { type: "bind"; request: number; diff --git a/packages/coding-agent/src/modes/rpc/session-worker.ts b/packages/coding-agent/src/modes/rpc/session-worker.ts index 8cffeede61..5f0fdb15ea 100644 --- a/packages/coding-agent/src/modes/rpc/session-worker.ts +++ b/packages/coding-agent/src/modes/rpc/session-worker.ts @@ -8,7 +8,7 @@ import type { ExtensionUIContext } from "../../core/extensions/types.ts"; import { attachMcpGrantPresentation } from "../../core/mcp-grant-presentations.ts"; import { takeOverStdout } from "../../core/output-guard.ts"; import type { SecurityNoticeSnapshot } from "../../core/security-notices.ts"; -import { recoveryCarrier } from "../../core/session-recovery-identity.ts"; +import { recoveryCarrier, setSessionRecoveryEpoch } from "../../core/session-recovery-identity.ts"; import { getDefaultSessionDir, type SessionManager } from "../../core/session-manager.ts"; import { subscriptionLimits } from "../../core/subscription-limits/limit-store.ts"; import { liveSessionWritePaths } from "../../core/session-write-reservation.ts"; @@ -117,12 +117,18 @@ class WorkerEventWriter extends SessionEventWriter { const replaced = "type" in record && record.type === "session_replaced"; if (ordered && replaced) journaled = session.sessionManager; const transitions = ordered && !replaced ? (journaled?.drainTranscript() ?? []) : []; + const epochSignal = ordered && replaced ? new SharedArrayBuffer(36) : undefined; const order = ordered ? { ordinal: outputOrdinal++, recovery: recoveryCarrier(record), ...(replaced - ? { bootstrap: { transcript: withinFrameLimit(session.sessionManager.beginTranscriptJournal()) } } + ? { + bootstrap: { + transcript: withinFrameLimit(session.sessionManager.beginTranscriptJournal()), + epochSignal, + }, + } : {}), ...(transitions.length > 0 ? { transitions: withinFrameLimit(transitions) } : {}), } @@ -138,6 +144,7 @@ class WorkerEventWriter extends SessionEventWriter { activity, snapshot: replacement, })); + if (epochSignal) setSessionRecoveryEpoch(session, Buffer.from(epochSignal).toString("ascii")); return true; } } @@ -223,6 +230,8 @@ async function handle(message: HostToSessionWorker): Promise { const opened = await registry.openSession(prepared.profile); entry = registry.getForCommand(opened.sessionId, "open_session"); ordered = message.acceptance === true; + if (message.recoveryEpoch && entry.runtime) + setSessionRecoveryEpoch(entry.runtime.session, message.recoveryEpoch); journaled = ordered ? entry.runtime?.session.sessionManager : undefined; send({ type: "ready", diff --git a/packages/coding-agent/test/suite/regressions/4272-epoch-carrier.test.ts b/packages/coding-agent/test/suite/regressions/4272-epoch-carrier.test.ts new file mode 100644 index 0000000000..f6f303be93 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-epoch-carrier.test.ts @@ -0,0 +1,47 @@ +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { Worker } from "node:worker_threads"; +import { expect, it } from "vitest"; +import { recoveryCarrier, SessionRecoveryIdentity } from "../../../src/core/session-recovery-identity.ts"; +import { SessionAcceptanceLane } from "../../../src/modes/rpc/session-acceptance-lane.ts"; +import { within } from "./4272-worker-acceptance-support.ts"; + +const terminal = () => ({ type: "message_end", message: { role: "user", content: "terminal", timestamp: 1 } }); + +// Part of #4272: the six RULING-24 boundary-proof REDs, at the production acceptance seam. +for (const transport of ["in-process", "worker frame"] as const) { + for (const kind of ["previous epoch", "foreign producer", "unminted id"] as const) { + it(`R24.3 Q7 rejects ${kind} across ${transport}`, async () => { + const lane = new SessionAcceptanceLane(); + lane.bootstrap({ transcript: [] }); + const record = terminal(); + const producer = new SessionRecoveryIdentity(kind === "previous epoch" ? lane.recoveryEpoch : undefined); + producer.observe(record); + let frame = { + record, + recovery: + kind === "unminted id" ? { messageId: randomUUID(), epoch: randomUUID() } : recoveryCarrier(record), + }; + if (kind === "previous epoch") lane.bootstrap({ transcript: [] }); + if (transport === "worker frame") { + const worker = new Worker( + `const { parentPort, workerData } = require("node:worker_threads"); parentPort.postMessage(workerData);`, + { eval: true, workerData: frame }, + ); + const received = once(worker, "message"); + const exited = once(worker, "exit"); + try { + [frame] = await within("carrier postMessage", received); + expect((await within("carrier worker exit", exited))[0]).toBe(0); + } finally { + await worker.terminate(); + } + } + try { + expect(lane.accept(frame.record, frame.recovery)?.messageId).toBeNull(); + } finally { + lane.retire("close"); + } + }); + } +} diff --git a/packages/coding-agent/test/suite/regressions/4272-epoch-nonce-not-persisted.test.ts b/packages/coding-agent/test/suite/regressions/4272-epoch-nonce-not-persisted.test.ts new file mode 100644 index 0000000000..dbef34096e --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-epoch-nonce-not-persisted.test.ts @@ -0,0 +1,128 @@ +import { readFile } from "node:fs/promises"; +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, expect, it, vi } from "vitest"; +import { recoveryCarrier } from "../../../src/core/session-recovery-identity.ts"; +import { SessionEventWriter } from "../../../src/modes/rpc/session-event-writer.ts"; +import { RpcSessionRegistry, type RpcSessionEntry } from "../../../src/modes/rpc/session-registry.ts"; +import { dataOf, statusHost } from "../rpc-session-status-support.ts"; +import { workerAcceptanceHost } from "./4272-worker-acceptance-support.ts"; + +afterEach(() => { + vi.unstubAllEnvs(); + vi.restoreAllMocks(); +}); + +const promptText = "nonce isolation prompt"; +const answerText = "an answer streamed in a few deltas"; + +/** Observe the actual legacy sink bytes, preserving its writes and backpressure. */ +function captureLegacyBytes(): string[] { + const lines: string[] = []; + const register = SessionEventWriter.prototype.registerConnection; + vi.spyOn(SessionEventWriter.prototype, "registerConnection").mockImplementation(function ( + this: SessionEventWriter, + id, + connection, + options, + ) { + register.call( + this, + id, + { + ...connection, + writeRaw: (line) => { + lines.push(line); + connection.writeRaw(line); + }, + }, + options, + ); + }); + return lines; +} + +function expectNoEpoch(bytes: string, nonce: string): void { + expect(bytes).not.toContain("recoveryEpoch"); + expect(bytes).not.toContain(nonce); +} + +async function expectIsolatedBytes(file: string, lines: readonly string[], nonce: string): Promise { + const persisted = await readFile(file, "utf8"); + const legacy = lines.join(""); + // Non-vacuity: both raw surfaces must contain the completed conversation. + for (const bytes of [persisted, legacy]) { + expect(bytes).toContain(promptText); + expect(bytes).toContain(answerText); + expect(bytes).toContain('"role":"user"'); + expect(bytes).toContain('"role":"assistant"'); + expectNoEpoch(bytes, nonce); + } + expect(legacy).toContain('"type":"message_end"'); +} + +// Part of #4272, RULING-24: the journal's epoch must never become a persisted entry field. +it("keeps the real in-process epoch out of persisted JSONL and public legacy bytes", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const lines = captureLegacyBytes(); + await using host = await statusHost(); + const admissions = vi.spyOn(RpcSessionRegistry.prototype, "getForCommand"); + const client = host.client("legacy"); + const opened = dataOf(await client.request({ type: "open_session", cwd: host.cwd })); + const sessionId = String(opened.sessionId); + const entry: RpcSessionEntry | undefined = admissions.mock.results.findLast( + (result) => result.type === "return", + )?.value; + const lane = entry?.acceptance; + const session = entry?.runtime?.session; + if (!lane || !session) throw new Error("no real in-process lane"); + const file = session.sessionManager.getSessionFile(); + if (!file) throw new Error("no persisted session file"); + const nonce = lane.recoveryEpoch; + expect(nonce).not.toBe(""); + const terminalEpochs: (string | undefined)[] = []; + const stop = session.subscribe((event) => { + if (event.type === "message_end" && (event.message.role === "user" || event.message.role === "assistant")) + terminalEpochs.push(recoveryCarrier(event)?.epoch); + }); + try { + const done = client.waitFor((record) => record.type === "agent_idle"); + host.faux.setResponses([fauxAssistantMessage(answerText, { timestamp: 1 })]); + dataOf(await client.request({ type: "prompt", sessionId, message: promptText })); + await done; + await host.writer.flush(); + // Verify this was the producer's actual epoch, not an unrelated random string. + expect(terminalEpochs).toEqual([nonce, nonce]); + await expectIsolatedBytes(file, lines, nonce); + } finally { + stop(); + } +}, 90000); + +it("keeps the real worker epoch out of persisted JSONL and public legacy bytes", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const lines = captureLegacyBytes(); + const host = await workerAcceptanceHost(); + try { + const { sessionId, worker } = await host.open(); + const lane = worker.acceptance; + const file = host.registry.peek(sessionId)?.sessionPath; + if (!lane || !file) throw new Error("no real worker lane or persisted session file"); + const nonce = lane.recoveryEpoch; + expect(nonce).not.toBe(""); + const done = host.next("nonce isolation turn idle", (record) => record.type === "agent_idle"); + expect((await host.request({ type: "prompt", sessionId, message: promptText })).success).toBe(true); + await done; + await host.writer.flush(); + const terminals = host.frames.filter(({ frame }) => { + if (!("type" in frame.record) || frame.record.type !== "message_end") return false; + const role = Reflect.get(Object("message" in frame.record ? frame.record.message : undefined), "role"); + return role === "user" || role === "assistant"; + }); + // Internal worker carriers must contain the nonce that the public sink excludes. + expect(terminals.map(({ frame }) => frame.recovery?.epoch)).toEqual([nonce, nonce]); + await expectIsolatedBytes(file, lines, nonce); + expect(host.failures).toEqual([]); + } finally { + await host.dispose(); + } +}, 90000); diff --git a/packages/coding-agent/test/suite/regressions/4272-epoch-producer.test.ts b/packages/coding-agent/test/suite/regressions/4272-epoch-producer.test.ts new file mode 100644 index 0000000000..9a5db4a82d --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-epoch-producer.test.ts @@ -0,0 +1,126 @@ +import { once } from "node:events"; +import { Worker } from "node:worker_threads"; +import { expect, it } from "vitest"; +import { + type RecoveryCarrier, + recoveryCarrier, + SessionRecoveryIdentity, + setSessionRecoveryEpoch, +} from "../../../src/core/session-recovery-identity.ts"; +import { SessionAcceptanceLane } from "../../../src/modes/rpc/session-acceptance-lane.ts"; +import { within } from "./4272-worker-acceptance-support.ts"; + +type Frame = { record: { type: string; message: { role: string; content: string } }; recovery: RecoveryCarrier }; + +it("same-epoch rebind preserves open messages, while a new epoch clears them", () => { + const session = {}; + const identity = new SessionRecoveryIdentity("epoch", session); + const assistant = { type: "message_start", message: { role: "assistant" } }; + const user = { type: "message_start", message: { role: "user" } }; + identity.observe(assistant); + identity.observe(user); + setSessionRecoveryEpoch(session, "epoch"); + const update = { type: "message_update", message: { role: "assistant" } }; + const userEnd = { type: "message_end", message: user.message }; + identity.observe(update); + identity.observe(userEnd); + expect(recoveryCarrier(update)).toEqual(recoveryCarrier(assistant)); + expect(recoveryCarrier(userEnd)).toEqual(recoveryCarrier(user)); + setSessionRecoveryEpoch(session, "new-epoch"); + const orphan = { type: "message_update", message: { role: "assistant" } }; + const newUserEnd = { type: "message_end", message: user.message }; + identity.observe(orphan); + identity.observe(newUserEnd); + expect(recoveryCarrier(orphan)).toBeUndefined(); + expect(recoveryCarrier(newUserEnd)?.epoch).toBe("new-epoch"); + expect(recoveryCarrier(newUserEnd)?.messageId).not.toBe(recoveryCarrier(user)?.messageId); +}); + +async function mint(epoch: string, workerFrame: boolean): Promise { + const record = { type: "message_end", message: { role: "user", content: "a direct terminal" } }; + if (!workerFrame) { + new SessionRecoveryIdentity(epoch).observe(record); + const recovery = recoveryCarrier(record); + if (!recovery) throw new Error("no mint"); + return { record, recovery }; + } + const worker = new Worker( + ` + const { parentPort, workerData } = require("node:worker_threads"); + import(workerData.url).then(({ SessionRecoveryIdentity, recoveryCarrier }) => { + const record = workerData.record; + new SessionRecoveryIdentity(workerData.epoch).observe(record); + parentPort.postMessage({ record, recovery: recoveryCarrier(record) }); + }); + `, + { + eval: true, + workerData: { + epoch, + record, + url: new URL("../../../src/core/session-recovery-identity.ts", import.meta.url).href, + }, + }, + ); + const received = once(worker, "message"); + const exited = once(worker, "exit"); + try { + const [frame] = await within("producer worker mint", received); + expect((await within("producer worker exit", exited))[0]).toBe(0); + return frame; + } finally { + await worker.terminate(); + } +} + +// Part of #4272, RULING-24 conditions 1-4: the host nonce, not object identity, crosses the boundary. +for (const transport of ["in-process", "worker frame"] as const) { + it(`R24.1 R24.2 R24.3 Q1 Q7 restart rejects old terminals and accepts a new start-less mint via ${transport}`, async () => { + const lane = new SessionAcceptanceLane(); + lane.bootstrap({ transcript: [] }); + const oldNonce = lane.recoveryEpoch; + const old = await mint(oldNonce, transport === "worker frame"); + lane.bootstrap({ transcript: [] }); + const current = await mint(lane.recoveryEpoch, transport === "worker frame"); + try { + expect(lane.recoveryEpoch).not.toBe(oldNonce); + expect(lane.accept(old.record, old.recovery)?.messageId).toBeNull(); + const accepted = lane.accept(current.record, current.recovery); + expect(accepted?.messageId).toBe(current.recovery.messageId); + expect(accepted?.sessionEpoch).not.toBe(current.recovery.epoch); + const root = lane.freeze(); + expect(root.settled).toBeDefined(); + const before = await lane.readSettled(root); + expect(before.settledMessages).toHaveLength(1); + lane.accept(current.record, current.recovery); + lane.accept(current.record, undefined); + const after = lane.freeze(); + expect(await lane.readSettled(after)).toEqual(before); + expect(JSON.stringify(current.record)).not.toContain(current.recovery.epoch); + expect(JSON.stringify(current.record)).not.toContain(current.recovery.messageId); + } finally { + lane.retire("close"); + await lane.idle?.(); + } + }); +} + +it("R24.4 S04a producer binds start/update/end and direct-terminal ids to the host bootstrap nonce", () => { + const identity = new SessionRecoveryIdentity("host-nonce"); + const start = { type: "message_start", message: { role: "assistant" } }; + const update = { type: "message_update", message: { role: "assistant" } }; + const end = { type: "message_end", message: { role: "assistant" } }; + for (const event of [start, update, end]) identity.observe(event); + expect(recoveryCarrier(start)).toMatchObject({ epoch: "host-nonce", messageId: expect.any(String) }); + expect(recoveryCarrier(update)).toEqual(recoveryCarrier(start)); + expect(recoveryCarrier(end)).toEqual(recoveryCarrier(start)); + identity.reset("next-host-nonce"); + const orphan = { type: "message_update", message: { role: "assistant" } }; + identity.observe(orphan); + expect(recoveryCarrier(orphan)).toBeUndefined(); + const direct = { type: "message_end", message: { role: "assistant" } }; + identity.observe(direct); + expect(recoveryCarrier(direct)?.epoch).toBe("next-host-nonce"); + expect(recoveryCarrier(direct)?.messageId).not.toBe(recoveryCarrier(start)?.messageId); + expect(Object.keys(direct)).toEqual(["type", "message"]); +}); diff --git a/packages/coding-agent/test/suite/regressions/4272-settled-support.ts b/packages/coding-agent/test/suite/regressions/4272-settled-support.ts new file mode 100644 index 0000000000..e028389e79 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-settled-support.ts @@ -0,0 +1,45 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect } from "vitest"; +import { SessionManager } from "../../../src/core/session-manager.ts"; +import { recoveryCarrier, SessionRecoveryIdentity } from "../../../src/core/session-recovery-identity.ts"; +import type { CompactContentStoreOptions } from "../../../src/modes/rpc/compact-content-store.ts"; +import { type SettledDiagnostic } from "../../../src/modes/rpc/compact-settled-messages.ts"; +import { SessionAcceptanceLane } from "../../../src/modes/rpc/session-acceptance-lane.ts"; + +export async function settledRig(options: Partial = {}) { + const dir = await mkdtemp(join(tmpdir(), "s05-settled-")); + const diagnostics: SettledDiagnostic[] = []; + const lane = new SessionAcceptanceLane({ dir, ...options }, (reason) => diagnostics.push(reason)); + const manager = SessionManager.inMemory(dir); + lane.bootstrap({ transcript: manager.beginTranscriptJournal() }); + const producer = new SessionRecoveryIdentity(lane.recoveryEpoch); + return { + dir, + lane, + manager, + producer, + diagnostics, + end(role = "user", content = "terminal") { + const record = { type: "message_end", message: { role, content, timestamp: 1 } }; + producer.observe(record); + const carrier = recoveryCarrier(record); + if (!carrier) throw new Error("producer failed to mint"); + lane.accept(record, carrier); + return { record, carrier }; + }, + async cut() { + const root = lane.freeze(); + // On the preceding implementation the value assertion is RED, not a missing-module exception. + expect(root.settled, "C2.1 cut owns its terminal state").toBeDefined(); + return { root, ...(await lane.readSettled(root)) }; + }, + async close() { + lane.retire("close"); + // The pre-S05 RED baseline has no spool cleanup method. + await lane.idle?.(); + await rm(dir, { recursive: true, force: true }); + }, + }; +} diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-bytes.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-bytes.test.ts new file mode 100644 index 0000000000..2562be0295 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-bytes.test.ts @@ -0,0 +1,56 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it, vi } from "vitest"; +import { SessionManager } from "../../../src/core/session-manager.ts"; +import { + carryRecovery, + recoveryCarrier, + SessionRecoveryIdentity, +} from "../../../src/core/session-recovery-identity.ts"; + +afterEach(() => vi.unstubAllEnvs()); + +// Part of #4272, R24.1: exact pre-carrier JSON grammar, not just absence of a named property. +for (const enabled of [false, true]) { + it(`R24.1 C2.4 legacy event and transcript bytes retain their grammar with flag ${enabled ? "ON" : "OFF"}`, async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", enabled ? "1" : "0"); + const dir = await mkdtemp(join(tmpdir(), "s05-bytes-")); + const manager = SessionManager.create(dir, dir); + try { + await manager.persistHeaderNow(); + manager.beginTranscriptJournal(); + const message = { role: "user" as const, content: "payload", timestamp: 1 }; + const event = { type: "message_end", message }; + if (enabled) new SessionRecoveryIdentity().observe(event); + const carrier = recoveryCarrier(event); + const copy = carryRecovery(event, { ...event }); + expect(JSON.stringify(copy)).toBe( + '{"type":"message_end","message":{"role":"user","content":"payload","timestamp":1}}', + ); + expect(Object.getOwnPropertySymbols(copy)).toEqual([]); + const id = manager.appendOwnedMessage(message, { recoveryOwner: carrier }); + const entry = manager.getEntry(id); + if (!entry) throw new Error("missing committed entry"); + const file = manager.getSessionFile(); + if (!file) throw new Error("missing transcript"); + const bytes = await readFile(file, "utf8"); + expect(bytes.split("\n").at(-2)).toBe( + JSON.stringify({ + type: "message", + id, + parentId: entry.parentId, + timestamp: entry.timestamp, + message, + }), + ); + const transitions = manager.drainTranscript(); + const append = transitions.find((transition) => transition.kind === "append"); + expect(append?.kind === "append" ? append.owningMessageId : undefined).toBe(carrier?.messageId); + if (carrier) expect(bytes).not.toContain(carrier.epoch); + } finally { + manager.dispose(); + await rm(dir, { recursive: true, force: true }); + } + }); +} diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-hint-failure.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-hint-failure.test.ts new file mode 100644 index 0000000000..f1c86900db --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-hint-failure.test.ts @@ -0,0 +1,47 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, expect, it, vi } from "vitest"; +import { lifecycleDeliveryEntries } from "../../../src/core/lifecycle-delivery.ts"; +import { recoveryCarrier } from "../../../src/core/session-recovery-identity.ts"; +import { createBoundLifecycleHarness } from "../lifecycle-test-harness.ts"; + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); +}); + +// Part of #4272: resolving the hint before publication must preserve the old persistence catch boundary. +it("C2.3 a lifecycle hint read failure still publishes its terminal and correlated refusal", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const h = await createBoundLifecycleHarness({ persistSession: true }); + try { + const ticket = h.session.workObligations.open("task:hint", "children").transfer("omo-wake-queue:hint").ticket(); + const reduce = h.sessionManager.reduceHistory.bind(h.sessionManager); + vi.spyOn(h.sessionManager, "reduceHistory").mockImplementation((reducer) => { + if (reducer === lifecycleDeliveryEntries) throw new Error("delivery history read refused"); + return reduce(reducer); + }); + h.setResponses([fauxAssistantMessage("done")]); + await Promise.allSettled([ + h.session.sendCustomMessage( + { customType: "senpi-task.completion", content: "preserve this result", display: false }, + { triggerTurn: true, ticket }, + ), + ]); + const end = h.events.find( + (event) => + event.type === "message_end" && + event.message.role === "custom" && + event.message.content === "preserve this result", + ); + expect(end, "the terminal precedes any persistence refusal").toBeDefined(); + const failed = h.events.find( + (event) => event.type === "transcript_write_failed" && event.errorMessage === "delivery history read refused", + ); + expect(failed).toBeDefined(); + if (!end || !failed) throw new Error("missing terminal or refusal"); + expect(recoveryCarrier(failed)).toEqual(recoveryCarrier(end)); + expect(h.events.indexOf(end)).toBeLessThan(h.events.indexOf(failed)); + } finally { + h.cleanup(); + } +}, 30000); diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-inprocess.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-inprocess.test.ts new file mode 100644 index 0000000000..c4610fd915 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-inprocess.test.ts @@ -0,0 +1,126 @@ +import { chmodSync, readFileSync } from "node:fs"; +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, expect, it, vi } from "vitest"; +import { recoveryCarrier } from "../../../src/core/session-recovery-identity.ts"; +import { type FrozenAcceptedRoot } from "../../../src/modes/rpc/session-acceptance-lane.ts"; +import { RpcSessionRegistry, type RpcSessionEntry } from "../../../src/modes/rpc/session-registry.ts"; +import { dataOf, statusHost } from "../rpc-session-status-support.ts"; + +afterEach(() => { + vi.unstubAllEnvs(); + vi.restoreAllMocks(); +}); + +// Part of #4272: actual host acceptance before persistence; no synthetic append or reducer-only substitute. +it("T1 C2.1-C2.4 in-process cuts preserve pending and failed payloads; new_session explicitly clears them", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + await using host = await statusHost(); + const admissions = vi.spyOn(RpcSessionRegistry.prototype, "getForCommand"); + const client = host.client("terminal"); + const opened = dataOf(await client.request({ type: "open_session", cwd: host.cwd })); + const sessionId = String(opened.sessionId); + const entry = admissions.mock.results.findLast((result) => result.type === "return")?.value as + | RpcSessionEntry + | undefined; + const lane = entry?.acceptance; + const session = entry?.runtime?.session; + if (!lane || !session) throw new Error("no real lane"); + const file = session.sessionManager.getSessionFile(); + if (!file) throw new Error("no file"); + const cuts: { role: string; id: string; root: FrozenAcceptedRoot; wire: string }[] = []; + const stop = session.subscribe((event) => { + if (event.type !== "message_end") return; + if (event.message.role !== "user" && event.message.role !== "assistant") return; + const carrier = recoveryCarrier(event); + if (!carrier) throw new Error("no terminal carrier"); + cuts.push({ + role: event.message.role, + id: carrier.messageId, + root: lane.freeze(), + wire: JSON.stringify(event.message), + }); + }); + const prompt = async (text: string) => { + const since = client.inbox.length; + const done = client.waitFor((record) => client.inbox.indexOf(record) >= since && record.type === "agent_idle"); + host.faux.setResponses([fauxAssistantMessage(`answer ${text}`, { timestamp: 1 })]); + dataOf(await client.request({ type: "prompt", sessionId, message: text })); + await done; + await host.writer.flush(); + }; + try { + await prompt("first"); + expect(cuts.map((cut) => cut.role)).toEqual(["user", "assistant"]); + for (const cut of cuts) { + expect(cut.root.settled).toBeDefined(); + const pending = (await lane.readSettled(cut.root)).settledMessages; + expect(pending.find((row) => row.messageId === cut.id)).toMatchObject({ status: "pending", entryId: null }); + expect(JSON.stringify(pending.find((row) => row.messageId === cut.id)?.terminal)).toBe(cut.wire); + } + const joinedRoot = lane.freeze(); + const joined = await lane.readSettled(joinedRoot); + expect(joined.settledMessages).toEqual([]); + expect(joined.owners.map((row) => row.messageId)).toEqual(expect.arrayContaining(cuts.map((cut) => cut.id))); + chmodSync(file, 0o444); + await prompt("refused"); + const failedRoot = lane.freeze(); + const failed = await lane.readSettled(failedRoot); + expect(failed.settledMessages).toHaveLength(2); + for (const row of failed.settledMessages) { + expect(row.status).toBe("failed"); + expect(row.error?.errorMessage).toContain("EACCES"); + } + chmodSync(file, 0o644); + await prompt("later"); + const before = { bytes: readFileSync(file, "utf8"), records: client.inbox.length }; + const fresh = lane.freeze(); + expect((await lane.readSettled(fresh)).settledMessages).toEqual(failed.settledMessages); + expect(await lane.readSettled(failedRoot)).toEqual(failed); + expect(client.inbox).toHaveLength(before.records); + expect(readFileSync(file, "utf8")).toBe(before.bytes); + const nonce = lane.recoveryEpoch; + expect(before.bytes).not.toContain(nonce); + expect(JSON.stringify(client.inbox)).not.toContain(nonce); + expect(JSON.stringify(client.inbox)).not.toContain('"recoveryEpoch"'); + dataOf(await client.request({ type: "new_session", sessionId })); + expect(lane.recoveryEpoch).not.toBe(nonce); + expect((await lane.readSettled(lane.freeze())).settledMessages).toEqual([]); + expect(lane.liveRoots).toBe(1); + } finally { + stop(); + chmodSync(file, 0o644); + for (const cut of cuts) lane.release(cut.root); + } +}, 90000); + +it("C2.2 Q3 idle custom append already owns its message before start and end", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + await using host = await statusHost(); + const admissions = vi.spyOn(RpcSessionRegistry.prototype, "getForCommand"); + const client = host.client("custom"); + dataOf(await client.request({ type: "open_session", cwd: host.cwd })); + const entry = admissions.mock.results.findLast((result) => result.type === "return")?.value as + | RpcSessionEntry + | undefined; + const lane = entry?.acceptance; + const session = entry?.runtime?.session; + if (!lane || !session) throw new Error("no real lane"); + const cuts: FrozenAcceptedRoot[] = []; + const stop = session.subscribe((event) => { + if ((event.type === "message_start" || event.type === "message_end") && event.message.role === "custom") + cuts.push(lane.freeze()); + }); + try { + await session.sendCustomMessage({ customType: "test", content: "idle notice", display: true }); + expect(cuts).toHaveLength(2); + for (const root of cuts) { + expect(root.settled).toBeDefined(); + const cut = await lane.readSettled(root); + expect(cut.settledMessages).toEqual([]); + expect(cut.owners).toHaveLength(1); + } + } finally { + stop(); + for (const root of cuts) lane.release(root); + } +}, 60000); diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-joins.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-joins.test.ts new file mode 100644 index 0000000000..ef3b4a235a --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-joins.test.ts @@ -0,0 +1,179 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { recoveryCarrier } from "../../../src/core/session-recovery-identity.ts"; +import { readTranscriptRoot } from "../../../src/modes/rpc/accepted-transcript.ts"; +import { settledRig } from "./4272-settled-support.ts"; + +afterEach(() => vi.unstubAllEnvs()); + +// Part of #4272: C2.1-C2.4, ruling 12 Q3/Q4/Q7. +for (const order of ["end-append", "append-end"] as const) { + it(`T1 C2.1 C2.2 Q3 ${order}: every cut has exactly one terminal representation`, async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const rig = await settledRig(); + try { + const message = { role: "user" as const, content: "frozen payload", timestamp: 1 }; + const start = { type: "message_start", message }; + rig.producer.observe(start); + const owner = recoveryCarrier(start); + if (!owner) throw new Error("missing owner"); + rig.lane.accept(start, owner); + const end = () => { + const record = { type: "message_end", message }; + rig.producer.observe(record); + rig.lane.accept(record, recoveryCarrier(record)); + }; + const append = () => { + rig.manager.appendOwnedMessage(message, { recoveryOwner: owner }); + rig.lane.applyTranscript(rig.manager.drainTranscript()); + }; + const operations = order === "end-append" ? [end, append] : [append, end]; + const held = []; + for (const operation of operations) { + operation(); + const cut = await rig.cut(); + held.push(cut); + const history = readTranscriptRoot(cut.root.transcript); + expect(history.ok).toBe(true); + const entries = history.ok ? history.entries.filter((entry) => entry.type === "message") : []; + expect(cut.settledMessages.length + entries.length).toBe(1); + if (cut.settledMessages.length) + expect(cut.settledMessages[0]).toMatchObject({ + messageId: owner.messageId, + entryId: null, + status: "pending", + terminal: message, + runtimeItemKey: null, + }); + else expect(cut.owners[0]).toMatchObject({ messageId: owner.messageId, entryId: entries[0]?.id }); + } + message.content = "mutated after acceptance"; + expect(await rig.lane.readSettled(held[0].root)).toEqual({ + settledMessages: held[0].settledMessages, + owners: held[0].owners, + }); + expect(held[1].settledMessages).toEqual([]); + } finally { + await rig.close(); + } + }); +} + +it("T1 C2.3 C2.4 failure preserves the held pending cut and fresh reads replay no effects", async () => { + const rig = await settledRig(); + try { + const { carrier } = rig.end("assistant"); + const pending = await rig.cut(); + rig.lane.accept({ type: "transcript_write_failed", errorMessage: "EACCES denied" }, carrier); + const failed = await rig.cut(); + expect(failed.settledMessages[0]).toMatchObject({ + status: "failed", + error: { errorMessage: "EACCES denied" }, + terminalSequence: pending.settledMessages[0]?.terminalSequence, + }); + for (const type of ["turn_end", "agent_end", "compaction_end", "context_edited"]) + rig.lane.accept({ type }, undefined); + rig.lane.applyTranscript([{ kind: "leaf", leafId: null }]); + rig.end("user", "later message"); + const beforeRead = rig.lane.freeze(); + const persist = vi.spyOn(rig.manager, "_persist"); + for (let i = 0; i < 3; i++) { + expect((await rig.lane.readSettled(pending.root)).settledMessages[0]?.status).toBe("pending"); + expect((await rig.cut()).settledMessages.find((row) => row.messageId === carrier.messageId)).toEqual( + failed.settledMessages[0], + ); + } + expect(rig.lane.freeze().sequence).toBe(beforeRead.sequence); + expect(persist).not.toHaveBeenCalled(); + } finally { + await rig.close(); + } +}); + +for (const role of ["user", "assistant", "toolResult", "custom", "futureRole"]) { + it(`C2.1 Q4 retains the full ${role} terminal without inventing durability`, async () => { + const rig = await settledRig(); + try { + const { carrier } = rig.end(role); + const cut = await rig.cut(); + expect(cut.settledMessages[0]).toMatchObject({ + messageId: carrier.messageId, + role, + status: "pending", + entryId: null, + }); + expect(cut.owners).toEqual([]); + expect(rig.diagnostics).toEqual(role === "futureRole" ? ["unsupported_role"] : []); + } finally { + await rig.close(); + } + }); +} + +it("C2.2 Q3 durableEntryId joins an existing base entry before the terminal", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const rig = await settledRig(); + try { + const id = rig.manager.appendCustomMessageEntry("notice", "existing", true); + rig.lane.applyTranscript(rig.manager.drainTranscript()); + const event = { type: "message_end", message: { role: "custom", content: "existing" } }; + rig.producer.observe(event); + rig.producer.hint?.(event, id); + rig.lane.accept(event, recoveryCarrier(event)); + const cut = await rig.cut(); + expect(cut.settledMessages).toEqual([]); + expect(cut.owners[0]?.entryId).toBe(id); + } finally { + await rig.close(); + } +}); + +for (const invalid of ["duplicate end", "unknown end", "retired append"] as const) { + it(`Q7 C2.3 ${invalid} is diagnostic-only and leaves settled state unchanged`, async () => { + const rig = await settledRig(); + try { + const { record, carrier } = rig.end(); + const before = await rig.cut(); + if (invalid === "retired append") + rig.lane.applyTranscript([ + { + kind: "append", + id: "foreign-entry", + entry: "{}", + owningMessageId: carrier.messageId, + recoveryEpoch: "retired", + }, + ]); + else rig.lane.accept(record, invalid === "duplicate end" ? carrier : undefined); + const after = await rig.cut(); + expect(after.settledMessages).toEqual(before.settledMessages); + expect(after.owners).toEqual(before.owners); + expect(rig.diagnostics).toEqual([ + invalid === "duplicate end" + ? "duplicate_end" + : invalid === "unknown end" + ? "unknown_message" + : "retired_owner", + ]); + expect(rig.lane.retired).toBe(false); + } finally { + await rig.close(); + } + }); +} + +it("C2.2 Q6 a later owned append joins a failed row and releases its payload", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const rig = await settledRig(); + try { + const { carrier } = rig.end(); + rig.lane.accept({ type: "transcript_write_failed", errorMessage: "denied" }, carrier); + rig.manager.appendOwnedMessage({ role: "user", content: "terminal", timestamp: 1 }, { recoveryOwner: carrier }); + rig.lane.applyTranscript(rig.manager.drainTranscript()); + const cut = await rig.cut(); + expect(cut.settledMessages).toEqual([]); + expect(cut.owners).toHaveLength(1); + expect(rig.lane.storageStatus().residentPayloadBytes).toBe(0); + } finally { + await rig.close(); + } +}); diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-lifetime.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-lifetime.test.ts new file mode 100644 index 0000000000..17beff3119 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-lifetime.test.ts @@ -0,0 +1,72 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { settledRig } from "./4272-settled-support.ts"; + +afterEach(() => vi.unstubAllEnvs()); + +// Part of #4272, ruling 12 Q5: transcript navigation is not explicit session clear. +it("C2.3 Q5 leaf moves, branches, edits and compaction preserve failed overlays", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const rig = await settledRig(); + try { + const id = rig.manager.appendMessage({ role: "user", content: "durable context", timestamp: 1 }); + rig.lane.applyTranscript(rig.manager.drainTranscript()); + const { carrier } = rig.end("assistant", "refused content"); + rig.lane.accept({ type: "transcript_write_failed", errorMessage: "denied" }, carrier); + const failed = (await rig.cut()).settledMessages; + for (const action of [ + () => rig.manager.branch(id), + () => rig.manager.resetLeaf(), + () => rig.manager.branch(id), + () => rig.manager.appendContextEdit(id, { content: "edited" }), + () => rig.manager.appendCompaction("summary", id, 100), + ]) { + action(); + rig.lane.applyTranscript(rig.manager.drainTranscript()); + expect((await rig.cut()).settledMessages).toEqual(failed); + } + } finally { + await rig.close(); + } +}); + +for (const reason of ["close", "worker_failed"] as const) { + it(`C2.3 Q5 ${reason} retires failed content and every held root`, async () => { + const rig = await settledRig(); + try { + const { carrier } = rig.end(); + rig.lane.accept({ type: "transcript_write_failed", errorMessage: "denied" }, carrier); + const cut = await rig.cut(); + expect(cut.settledMessages[0]?.status).toBe("failed"); + rig.lane.retire(reason); + await rig.lane.idle(); + expect(rig.lane.liveRoots).toBe(0); + expect(() => rig.lane.readSettled(cut.root)).toThrow("accepted root released"); + expect(rig.lane.accept({ type: "message_end" }, carrier)).toBeUndefined(); + } finally { + await rig.close(); + } + }); +} + +it("C2.2 C2.4 one thousand joined messages leave zero payload residency", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const rig = await settledRig(); + try { + for (let index = 0; index < 1000; index++) { + const { carrier } = rig.end(); + rig.manager.appendOwnedMessage( + { role: "user", content: "terminal", timestamp: 1 }, + { recoveryOwner: carrier }, + ); + rig.lane.applyTranscript(rig.manager.drainTranscript()); + } + await rig.lane.idle?.(); + const cut = await rig.cut(); + expect(cut.settledMessages).toEqual([]); + expect(cut.owners).toHaveLength(1000); + expect(rig.lane.storageStatus().residentPayloadBytes).toBe(0); + expect(rig.lane.storageStatus().overrideSlots).toBeLessThan(128); + } finally { + await rig.close(); + } +}, 30000); diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-producer-paths.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-producer-paths.test.ts new file mode 100644 index 0000000000..a0bb24f2f9 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-producer-paths.test.ts @@ -0,0 +1,105 @@ +import type { AgentTool } from "@earendil-works/pi-agent-core"; +import { fauxAssistantMessage, fauxToolCall } from "@earendil-works/pi-ai"; +import { Type } from "typebox"; +import { afterEach, expect, it, vi } from "vitest"; +import { recoveryCarrier, setSessionRecoveryEpoch } from "../../../src/core/session-recovery-identity.ts"; +import { SessionAcceptanceLane, type FrozenAcceptedRoot } from "../../../src/modes/rpc/session-acceptance-lane.ts"; +import { createBoundLifecycleHarness, runToAgentIdle } from "../lifecycle-test-harness.ts"; + +afterEach(() => vi.unstubAllEnvs()); + +// Part of #4272: the two producer append paths that differ from ordinary user/assistant persistence. +it("C2.1 T1 toolResult freezes the full result before its owned durable append", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const tool: AgentTool = { + name: "result", + label: "Result", + description: "Return a deterministic result", + parameters: Type.Object({}), + execute: async () => ({ content: [{ type: "text", text: "full tool output" }], details: { complete: true } }), + }; + const h = await createBoundLifecycleHarness({ tools: [tool], persistSession: true }); + const lane = new SessionAcceptanceLane(); + lane.bootstrap({ transcript: h.sessionManager.beginTranscriptJournal() }); + setSessionRecoveryEpoch?.(h.session, lane.recoveryEpoch); + let cut: FrozenAcceptedRoot | undefined; + let messageId: string | undefined; + const stop = h.session.subscribe((event) => { + lane.applyTranscript(h.sessionManager.drainTranscript()); + lane.accept(event, recoveryCarrier(event)); + if (event.type === "message_end" && event.message.role === "toolResult") { + cut = lane.freeze(); + messageId = recoveryCarrier(event)?.messageId; + } + }); + try { + h.setResponses([ + fauxAssistantMessage([fauxToolCall("result", {})], { stopReason: "toolUse" }), + fauxAssistantMessage("done"), + ]); + await h.session.prompt("use the tool"); + expect(cut?.settled).toBeDefined(); + if (!cut) throw new Error("no tool result cut"); + expect((await lane.readSettled(cut)).settledMessages.find((row) => row.messageId === messageId)).toMatchObject({ + role: "toolResult", + status: "pending", + entryId: null, + terminal: { content: [{ type: "text", text: "full tool output" }], details: { complete: true } }, + }); + const final = await lane.readSettled(lane.freeze()); + expect(final.settledMessages).toEqual([]); + expect(final.owners.find((row) => row.messageId === messageId)?.entryId).toBeTypeOf("string"); + } finally { + stop(); + lane.retire("close"); + await lane.idle?.(); + h.cleanup(); + } +}, 60000); + +it("C2.2 Q3 lifecycle consume binds the entry before its custom terminal and carries a committed hint", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const h = await createBoundLifecycleHarness({ persistSession: true }); + const lane = new SessionAcceptanceLane(); + lane.bootstrap({ transcript: h.sessionManager.beginTranscriptJournal() }); + setSessionRecoveryEpoch?.(h.session, lane.recoveryEpoch); + const cuts: { root: FrozenAcceptedRoot; id: string; hint?: string }[] = []; + const stop = h.session.subscribe((event) => { + lane.applyTranscript(h.sessionManager.drainTranscript()); + lane.accept(event, recoveryCarrier(event)); + if ( + event.type === "message_end" && + event.message.role === "custom" && + event.message.customType === "senpi-task.completion" + ) { + const carrier = recoveryCarrier(event); + if (!carrier) throw new Error("no custom owner"); + cuts.push({ root: lane.freeze(), id: carrier.messageId, hint: carrier.durableEntryId }); + } + }); + try { + const ticket = h.session.workObligations + .open("task:review", "children") + .transfer("omo-wake-queue:review") + .ticket(); + h.setResponses([fauxAssistantMessage("incorporated")]); + await runToAgentIdle(h, () => + h.session.sendCustomMessage( + { customType: "senpi-task.completion", content: "review result", display: false }, + { triggerTurn: true, ticket }, + ), + ); + expect(cuts).toHaveLength(1); + const cut = cuts[0]; + expect(cut.root.settled).toBeDefined(); + expect(cut.hint).toBeTypeOf("string"); + const state = await lane.readSettled(cut.root); + expect(state.settledMessages.some((row) => row.messageId === cut.id)).toBe(false); + expect(state.owners.find((row) => row.messageId === cut.id)?.entryId).toBe(cut.hint); + } finally { + stop(); + lane.retire("close"); + await lane.idle?.(); + h.cleanup(); + } +}, 60000); diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-storage.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-storage.test.ts new file mode 100644 index 0000000000..8e932c71ee --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-storage.test.ts @@ -0,0 +1,98 @@ +import { readdir, unlink } from "node:fs/promises"; +import { join } from "node:path"; +import { expect, it } from "vitest"; +import { CompactStorageReadError, nodeCompactStorageIo } from "../../../src/modes/rpc/compact-storage-io.ts"; +import { settledRig } from "./4272-settled-support.ts"; + +// Part of #4272, X03: both transcript refusal and spool refusal preserve all distinct terminals. +for (const fault of [false, true]) { + it(`X03 C2.4 retains 260 failed 260000-byte terminals when spool ${fault ? "fails" : "spills"}`, async () => { + const rig = await settledRig({ + spillAboveBytes: 128 * 1024, + spillTargetBytes: 64 * 1024, + ...(fault + ? { + io: { + ...nodeCompactStorageIo, + writeFile: async () => { + throw new Error("spool denied"); + }, + }, + } + : {}), + }); + try { + const body = "x".repeat(260000); + const ids = []; + for (let i = 0; i < 260; i++) { + const { carrier } = rig.end("toolResult", body); + ids.push(carrier.messageId); + rig.lane.accept({ type: "transcript_write_failed", errorMessage: "EACCES transcript" }, carrier); + } + await rig.lane.idle?.(); + const cut = await rig.cut(); + expect(cut.settledMessages.map((row) => row.messageId).sort()).toEqual(ids.sort()); + for (const row of cut.settledMessages) { + expect(row).toMatchObject({ + status: "failed", + error: { errorMessage: "EACCES transcript" }, + entryId: null, + }); + expect(row.terminal).toMatchObject({ content: body }); + } + const status = rig.lane.storageStatus(); + if (fault) { + expect(status.spillHalted).toBe(true); + expect(status.spillFailureRetainedBytes).toBeGreaterThan(260 * 260000); + expect(status.failures.some((failure) => failure.cause.includes("spool denied"))).toBe(true); + } else { + expect(status.residentPayloadBytes).toBeLessThanOrEqual(64 * 1024); + expect(status.basePages).toBeGreaterThan(0); + } + } finally { + await rig.close(); + } + }, 60000); +} + +it("C2.4 Q7 spool failure never marks a pending row failed", async () => { + const rig = await settledRig({ + spillAboveBytes: 0, + spillTargetBytes: 0, + io: { + ...nodeCompactStorageIo, + writeFile: async () => { + throw new Error("spool only"); + }, + }, + }); + try { + rig.end(); + await rig.lane.idle?.(); + expect((await rig.cut()).settledMessages[0]).toMatchObject({ status: "pending", error: null }); + expect(rig.lane.storageStatus().spillHalted).toBe(true); + } finally { + await rig.close(); + } +}); + +it("C2.4 a missing committed leaf returns a typed read error without changing failed status", async () => { + const rig = await settledRig({ spillAboveBytes: 0, spillTargetBytes: 0 }); + try { + const { carrier } = rig.end(); + rig.lane.accept({ type: "transcript_write_failed", errorMessage: "transcript denied" }, carrier); + await rig.lane.idle?.(); + const root = rig.lane.freeze(); + expect(root.settled).toBeDefined(); + const [spool] = await readdir(rig.dir); + const files = await readdir(join(rig.dir, spool)); + const leaf = files.find((name) => name.startsWith("leaf-")); + if (!leaf) throw new Error("no leaf was spilled"); + await unlink(join(rig.dir, spool, leaf)); + await expect(rig.lane.readSettled(root)).rejects.toBeInstanceOf(CompactStorageReadError); + expect(rig.lane.storageStatus().failures.at(-1)?.operation).toBe("leaf_read"); + expect(rig.lane.freeze().sequence).toBe(root.sequence); + } finally { + await rig.close(); + } +}); diff --git a/packages/coding-agent/test/suite/regressions/4272-terminal-worker.test.ts b/packages/coding-agent/test/suite/regressions/4272-terminal-worker.test.ts new file mode 100644 index 0000000000..28640338df --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/4272-terminal-worker.test.ts @@ -0,0 +1,109 @@ +import { chmodSync } from "node:fs"; +import { afterEach, expect, it, vi } from "vitest"; +import { recoveryCarrier, SessionRecoveryIdentity } from "../../../src/core/session-recovery-identity.ts"; +import type { FrozenAcceptedRoot } from "../../../src/modes/rpc/session-acceptance-lane.ts"; +import { creditOf, type OutputFrame, workerAcceptanceHost } from "./4272-worker-acceptance-support.ts"; + +afterEach(() => { + vi.unstubAllEnvs(); + vi.restoreAllMocks(); +}); + +// Part of #4272: real WorkerSessionRegistry + SessionWorkerClient, not only structuredClone. +it("T1 R24.1 R24.2 Q7 worker terminals join, failures survive, and replacement fences delayed old output", async () => { + vi.stubEnv("SENPI_ENGINE_SOCKET_COMPACT_UPDATES", "1"); + const host = await workerAcceptanceHost(); + let file: string | undefined; + let releaseSnapshot: (() => void) | undefined; + try { + const { sessionId, worker } = await host.open(); + const lane = worker.acceptance; + if (!lane) throw new Error("no worker lane"); + file = host.registry.peek(sessionId)?.sessionPath; + if (!file) throw new Error("no worker file"); + const cuts: FrozenAcceptedRoot[] = []; + const enqueue = host.writer.enqueue.bind(host.writer); + let stateResponse = false; + vi.spyOn(host.writer, "enqueue").mockImplementation((id, record) => { + if ("type" in record && record.type === "response" && "command" in record && record.command === "get_state") + stateResponse = true; + if ("type" in record && record.type === "message_end" && "message" in record) { + const role = Reflect.get(Object(record.message), "role"); + if (role === "user" || role === "assistant") cuts.push(lane.freeze()); + } + return enqueue(id, record); + }); + const prompt = async (message: string) => { + const done = host.next("worker terminal", (record) => record.type === "agent_idle"); + expect((await host.request({ type: "prompt", sessionId, message })).success).toBe(true); + await done; + await host.writer.flush(); + }; + await prompt("first"); + expect(cuts).toHaveLength(2); + for (const cut of cuts) { + expect(cut.settled).toBeDefined(); + expect((await lane.readSettled(cut)).settledMessages.at(-1)?.status).toBe("pending"); + } + const joined = await lane.readSettled(lane.freeze()); + expect(joined.settledMessages).toEqual([]); + expect(joined.owners.length).toBeGreaterThanOrEqual(2); + const old = host.frames.find(({ frame }) => "type" in frame.record && frame.record.type === "message_end")?.frame; + if (!old?.recovery) throw new Error("worker did not carry the minted owner"); + expect(old.recovery.epoch).toBe(lane.recoveryEpoch); + expect(JSON.stringify(host.records)).not.toContain(lane.recoveryEpoch); + chmodSync(file, 0o444); + await prompt("refused"); + const failed = await lane.readSettled(lane.freeze()); + expect(failed.settledMessages).toHaveLength(2); + expect( + failed.settledMessages.every((row) => row.status === "failed" && row.error?.errorMessage.includes("EACCES")), + ).toBe(true); + chmodSync(file, 0o644); + expect((await host.request({ type: "new_session", sessionId })).success).toBe(true); + expect(lane.recoveryEpoch).not.toBe(old.recovery.epoch); + const replacementStart = host.frames.length; + await prompt("after replacement"); + const newEnds = host.frames + .slice(replacementStart) + .filter(({ frame }) => "type" in frame.record && frame.record.type === "message_end"); + expect(newEnds.length).toBeGreaterThan(0); + expect(newEnds.every(({ frame }) => frame.recovery?.epoch === lane.recoveryEpoch)).toBe(true); + // Hold the snapshot after get_state's response: exchange() blocks the actual worker on this + // unreturned credit, so it cannot race synthetic output. Terminate it before releasing the hold. + const held = host.holdReconcile((message) => stateResponse && message?.type === "snapshot"); + releaseSnapshot = held.release; + expect((await host.request({ type: "get_state", sessionId })).success).toBe(true); + await held.reached; + const deliver = async (record: object, recovery: OutputFrame["recovery"]) => { + const ordinal: unknown = Reflect.get(worker, "expectedOrdinal"); + if (typeof ordinal !== "number") throw new Error("missing client ordinal seam"); + const frame: OutputFrame = { + type: "output", + ordinal, + record, + recovery, + signal: new SharedArrayBuffer(4), + activity: { busy: false, streaming: false }, + }; + worker.worker.emit("message", frame); + expect(await creditOf(frame.signal)).toBe(1); + }; + await deliver(old.record, old.recovery); + expect((await lane.readSettled(lane.freeze())).settledMessages).toEqual([]); + const terminal = { type: "message_end", message: { role: "user", content: "new direct terminal" } }; + new SessionRecoveryIdentity(lane.recoveryEpoch).observe(terminal); + await deliver(terminal, recoveryCarrier(terminal)); + const accepted = await lane.readSettled(lane.freeze()); + expect(accepted.settledMessages).toHaveLength(1); + await deliver(terminal, recoveryCarrier(terminal)); + await deliver(terminal, undefined); + expect(await lane.readSettled(lane.freeze())).toEqual(accepted); + expect(host.failures).toEqual([]); + expect(lane.retired).toBe(false); + } finally { + if (file) chmodSync(file, 0o644); + await host.dispose(); + releaseSnapshot?.(); + } +}, 90000);