fix: preserve Goal continuation through compaction recovery

packages/agent/src/agent-loop.ts

@@ -107,17 +107,24 @@ export async function runAgentLoop(
 	signal: AbortSignal | undefined,
 	streamFn: StreamFn,
 ): Promise<AgentMessage[]> {
-	const newMessages: AgentMessage[] = [...prompts];
+	const newMessages: AgentMessage[] = [];
 	const currentContext: AgentContext = {
 		...context,
-		messages: [...context.messages, ...prompts],
+		messages: [...context.messages],
 	};
 
 	await emit({ type: "agent_start" });
 	await emit({ type: "turn_start" });
 	for (const prompt of prompts) {
+		if (config.shouldDeliverMessage?.(prompt) === false) continue;
 		await emit({ type: "message_start", message: prompt });
 		await emit({ type: "message_end", message: prompt });
+		currentContext.messages.push(prompt);
+		newMessages.push(prompt);
+	}
+	if (newMessages.length === 0) {
+		await emit({ type: "agent_end", messages: newMessages });
+		return newMessages;
 	}
 
 	await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn());
@@ -188,6 +195,7 @@ async function runLoop(
 	let config = initialConfig;
 	let firstTurn = true;
 	let firstProviderRequest = true;
+	let continuingFromTerminatingQueue = false;
 	// Check for steering messages at start (user may have typed while waiting)
 	let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || [];
 	let drainedTerminatingQueue: "steering" | "followUp" | undefined;
@@ -204,7 +212,8 @@ async function runLoop(
 
 	// Outer loop: continues when queued follow-up messages arrive after agent would stop
 	while (true) {
-		let hasMoreToolCalls = true;
+		let hasMoreToolCalls = !continuingFromTerminatingQueue;
+		continuingFromTerminatingQueue = false;
 
 		// Inner loop: process tool calls and steering messages
 		while (hasMoreToolCalls || pendingMessages.length > 0) {
@@ -224,13 +233,18 @@ async function runLoop(
 
 			// Process pending messages (inject before next assistant response)
 			if (pendingMessages.length > 0) {
+				let deliveredPendingMessage = false;
 				for (const message of pendingMessages) {
+					if (config.shouldDeliverMessage?.(message) === false) continue;
+					deliveredPendingMessage = true;
 					await emit({ type: "message_start", message });
 					await emit({ type: "message_end", message });
 					currentContext.messages.push(message);
 					newMessages.push(message);
 				}
 				pendingMessages = [];
+				if (!deliveredPendingMessage && !hasMoreToolCalls) break;
+				if (deliveredPendingMessage) hasMoreToolCalls = true;
 			}
 
 			// Stream assistant response. Continuation-scoped overrides apply to one
@@ -381,6 +395,7 @@ async function runLoop(
 		if (followUpMessages.length > 0) {
 			// Set as pending so inner loop processes them
 			pendingMessages = followUpMessages;
+			continuingFromTerminatingQueue = true;
 			continue;
 		}
 

packages/agent/src/agent.ts

@@ -173,6 +173,13 @@ class PendingMessageQueue {
 		this.messages = [...messages, ...this.messages];
 	}
 
+	remove(message: AgentMessage): boolean {
+		const index = this.messages.indexOf(message);
+		if (index === -1) return false;
+		this.messages.splice(index, 1);
+		return true;
+	}
+
 	clear(): void {
 		this.messages = [];
 		this.clearGeneration++;
@@ -219,6 +226,7 @@ export class Agent {
 		context: PrepareNextTurnContext,
 		signal?: AbortSignal,
 	) => Promise<AgentLoopTurnUpdate | undefined> | AgentLoopTurnUpdate | undefined;
+	private messageFilter?: (message: AgentMessage) => boolean;
 	private activeRun?: ActiveRun;
 	/** Session identifier forwarded to providers for cache-aware backends. */
 	public sessionId?: string;
@@ -333,6 +341,16 @@ export class Agent {
 		this.clearFollowUpQueue();
 	}
 
+	/** Remove one exact queued message without disturbing identical siblings. */
+	removeQueuedMessage(message: AgentMessage): boolean {
+		return this.steeringQueue.remove(message) || this.followUpQueue.remove(message);
+	}
+
+	/** Install a last-moment admission check for prompt and drained queue messages. */
+	setMessageFilter(filter: ((message: AgentMessage) => boolean) | undefined): void {
+		this.messageFilter = filter;
+	}
+
 	/** Returns true when either queue still contains pending messages. */
 	hasQueuedMessages(): boolean {
 		return this.steeringQueue.hasItems() || this.followUpQueue.hasItems();
@@ -546,6 +564,7 @@ export class Agent {
 		let followUpQueueGeneration = this.followUpQueue.getClearGeneration();
 		return {
 			model: this._state.model,
+			shouldDeliverMessage: this.messageFilter,
 			reasoning: this._state.thinkingLevel === "off" ? undefined : this._state.thinkingLevel,
 			sessionId: this.sessionId,
 			onPayload: this.onPayload,

packages/agent/src/changes.md

@@ -1,5 +1,12 @@
 # Changes
 
+## 2026-07-31 - Exact queued-message cancellation
+
+### What changed and why
+
+- `Agent.removeQueuedMessage()` removes one exact steering or follow-up message object without clearing identical siblings.
+- Coding-agent extension delivery receipts use this identity-safe primitive to revoke superseded hidden work while preserving unrelated queued user and extension messages.
+
 ## 2026-07-30 - Bound empty Kimi assistant responses
 
 ### What changed and why

packages/agent/src/types.ts

@@ -145,6 +145,8 @@ export interface PrepareNextTurnContext extends ShouldStopAfterTurnContext {}
 
 export interface AgentLoopConfig extends SimpleStreamOptions {
 	model: Model<any>;
+	/** Last-moment admission check before a queued message enters loop-owned context. */
+	shouldDeliverMessage?: (message: AgentMessage) => boolean;
 
 	/**
 	 * Maximum time in milliseconds to wait for the FIRST provider stream event.

packages/coding-agent/docs/extensions.md

@@ -1488,7 +1488,7 @@ extension-declared MCP servers.
 Inject a custom message into the session. Custom messages participate in LLM context. For durable TUI-only content that should not be sent to the LLM, use [`pi.appendEntry()`](#piappendentrycustomtype-data) with [`pi.registerEntryRenderer()`](#piregisterentryrenderercustomtype-renderer).
 
 ```typescript
-pi.sendMessage({
+const delivery = pi.sendMessage({
   customType: "my-extension",
   content: "Message text",
   display: true,
@@ -1497,6 +1497,13 @@ pi.sendMessage({
   triggerTurn: true,
   deliverAs: "steer",
 });
+
+delivery.onStarted(() => {
+  // This exact custom message began a model turn.
+});
+
+// Removes only this pending delivery; returns false after it started or was already cancelled.
+delivery.cancel();
 ```
 
 **Options:**
@@ -1506,6 +1513,10 @@ pi.sendMessage({
   - `"nextTurn"` - Queued for next user prompt. Does not interrupt or trigger anything.
 - `triggerTurn: true` - If agent is idle, trigger an LLM response immediately. Only applies to `"steer"` and `"followUp"` modes (ignored for `"nextTurn"`).
 
+The returned `MessageDelivery` has an opaque `id`, `cancel()`, `onStarted()`, and `onCancelled()`. Cancellation is
+identity-based, so equal-content sibling messages remain queued. Clearing the session queue or disposing the session
+also cancels pending receipts.
+
 ### pi.sendUserMessage(content, options?)
 
 Send a user message to the agent. Unlike `sendMessage()` which sends custom messages, this sends an actual user message that appears as if typed by the user. Always triggers a turn.

packages/coding-agent/docs/sdk.md

@@ -73,8 +73,8 @@ interface AgentSession {
   prompt(text: string, options?: PromptOptions): Promise<void>;
 
   // Queue messages during streaming
-  steer(text: string): Promise<void>;
-  followUp(text: string): Promise<void>;
+  steer(text: string, images?: ImageContent[], options?: QueuedInputOptions): Promise<void>;
+  followUp(text: string, images?: ImageContent[], options?: QueuedInputOptions): Promise<void>;
 
   // Subscribe to events (returns unsubscribe function)
   subscribe(listener: (event: AgentSessionEvent) => void): () => void;
@@ -232,6 +232,8 @@ await session.followUp("After you're done, also do this");
 ```
 
 Both `steer()` and `followUp()` expand file-based prompt templates but error on extension commands (extension commands cannot be queued).
+They also emit the same correlated extension input/disposition lifecycle as queued `prompt()` calls. SDK calls default to
+`source: "interactive"`; transport adapters pass `source: "rpc"` when the input came from RPC.
 
 ### Agent and AgentState
 

packages/coding-agent/src/changes.md

@@ -1,3 +1,14 @@
+## Automatic compaction timeout recovery preserves active work (2026-08-01)
+
+- First-pass automatic compaction timeouts no longer fall back to a context-free marker: the deterministic checkpoint includes bounded recent user intent and the latest active todo state.
+- Compaction todo recovery now snapshots only the latest state and correctly restores it when all surviving todo records predate the newest compaction boundary.
+- Regression coverage drives both the required automatic fallback handler and the post-compaction todo bridge.
+
+## Direct input admission and cancellable extension deliveries (2026-07-31)
+
+- Public `AgentSession.steer()` and `followUp()` now run the same correlated extension input/disposition admission as queued `prompt()` calls. Classic RPC and app-server steering preserve `source: "rpc"`, while compaction queue transfer reuses the same path without duplicate events.
+- `pi.sendMessage()` returns an identity-safe delivery receipt with cancellation and started/cancelled subscriptions. Queue clear, disposal, and dispatch failure revoke pending receipts without removing identical sibling messages.
+- Direct reload retires the old extension runner only after removed-extension notifications finish, so captured pre-reload APIs are usable during teardown and stale afterward.
 ## Backfill: injected app-server turns (2026-08-01)
 
 ### What changed

packages/coding-agent/src/config.ts

@@ -304,11 +304,13 @@ function getEntrypointPackageDir(): string | undefined {
 	return undefined;
 }
 
-function isSelfUpdatePathWritable(): boolean {
-	const packageDir = getPackageDir();
+export function isSelfUpdatePathWritable(
+	packageDir = getPackageDir(),
+	access: (path: string, mode: number) => void = accessSync,
+): boolean {
 	try {
-		accessSync(packageDir, constants.W_OK);
-		accessSync(dirname(packageDir), constants.W_OK);
+		access(packageDir, constants.W_OK);
+		access(dirname(packageDir), constants.W_OK);
 		return true;
 	} catch {
 		return false;