feat(rpc): pending/failed terminals and message-correlated append joins (S05, part of #4272)

changes/unreleased/4374.md

@@ -0,0 +1,59 @@
+<!-- changes-log: packages/coding-agent/src/core/changes.md -->
+## 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.
+
+<!-- changes-log: packages/coding-agent/src/modes/rpc/changes.md -->
+## 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.
+
+<!-- changes-log: packages/coding-agent/CHANGELOG.md -->
+### 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)).

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";
 

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 });
 	}
 

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<Map<string, string>> = {
@@ -47,6 +48,7 @@ export function consumeLifecycleDelivery(
 			message.content,
 			message.display,
 			message.details,
+			{ recoveryOwner: recoveryCarrier(message) },
 		);
 	controller.obligations.consumed(ticketId, { runId, transcriptEntryId });
 	return transcriptEntryId;

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<T> = {
 			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;
 	}
 

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<object, RecoveryCarrier>();
+const producers = new WeakMap<object, SessionRecoveryIdentity>();
+
+/** 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<object, string>();
+	private byMessage = new WeakMap<object, string>();
+	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);
 	}
 }

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 }

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") {