ask_user posts the question and ends the turn; delete the message_response delta (CL-7443)

apps/sidecar/src/conversation-state.test.ts

@@ -9,7 +9,7 @@ import type {
   RepoId,
   RepoStore,
 } from "@intx/hub-sessions/substrate";
-import type { ConversationTurn } from "@intx/types/runtime";
+import type { ConversationTurn, PendingOperation } from "@intx/types/runtime";
 
 import {
   createDurableConversationStore,
@@ -169,6 +169,60 @@ test("two rooms of one agent keep isolated turns; a new room starts empty", asyn
   expect(await store.bindOriginatingWorkbench("chan_b")).toBe(false);
 });
 
+const EMPTY_TOKEN_USAGE = {
+  input: 0,
+  output: 0,
+  cacheRead: 0,
+  cacheWrite: 0,
+  thinking: 0,
+};
+
+test("a retired signal-kind pending op is dropped on restore, not thrown on", async () => {
+  const root = tmpDir("conv-retired-kind-");
+  const agentStateDir = path.join(root, "agent-state", "default", "chan_a");
+  await fs.promises.mkdir(agentStateDir, { recursive: true });
+
+  const survivingOp: PendingOperation = {
+    correlationId: "corr-approval",
+    kind: "approval",
+    registeredAt: 0,
+    gateId: "gate-approval",
+  };
+  const retiredOp = {
+    correlationId: "corr-retired",
+    kind: "message_response",
+    registeredAt: 0,
+    gateId: "gate-retired",
+  };
+
+  await fs.promises.writeFile(
+    path.join(agentStateDir, "checkpoint.json"),
+    JSON.stringify({
+      turns: [],
+      pendingOperations: [survivingOp, retiredOp],
+      tokenUsage: EMPTY_TOKEN_USAGE,
+      connectorState: null,
+    }),
+  );
+  await fs.promises.writeFile(
+    path.join(agentStateDir, "checkpoint.meta.json"),
+    JSON.stringify({
+      checkpointSeq: 0,
+      turnCount: 0,
+      pendingOperations: [survivingOp, retiredOp],
+      tokenUsage: EMPTY_TOKEN_USAGE,
+      connectorState: null,
+    }),
+  );
+
+  const reconstructed = await reconstructDurableConversation(
+    agentStateDir,
+    "default/chan_a",
+  );
+
+  expect(reconstructed?.pendingOperations).toEqual([survivingOp]);
+});
+
 test("two queued mails bind and mirror under each message's From not a later origin", async () => {
   const root = tmpDir("conv-queued-");
   const { store } = await makeStore(root);

apps/sidecar/src/conversation-state.ts

@@ -131,6 +131,7 @@ import path from "node:path";
 import { type } from "arktype";
 
 import { getLogger } from "@intx/log";
+import { reportError } from "@corbits/error-sink";
 import { createConnectorRouter } from "@intx/harness";
 import type { ConnectorReplyParts, RouteDecision } from "@intx/harness";
 import {
@@ -143,6 +144,7 @@ import type {
   RepoStore,
 } from "@intx/hub-sessions/substrate";
 import { WORKFLOW_RUN_AGENT_STATE_PREFIX } from "@intx/hub-sessions/substrate";
+import { SignalKind } from "@intx/types";
 
 import { UNSCOPED_ORIGINATING_WORKBENCH_ID } from "./originating-workbench";
 import {
@@ -995,6 +997,51 @@ function castValidatedEnvelope<T>(items: unknown[]): T[] {
   return items as T[];
 }
 
+/**
+ * A pending operation's `kind` envelope: just enough structure to read
+ * `kind` back off an otherwise-unvalidated `PendingOperation` (the element
+ * validator itself lives in the reactor, per `castValidatedEnvelope`'s
+ * doc comment above). Used only to detect a `kind` the running build's
+ * `SignalKind` no longer recognizes -- e.g. CL-7443 retired
+ * `message_response` -- before the reactor's `rehydrateGates` reaches it,
+ * since an unclassified kind there throws and blocks every future respawn.
+ */
+const PendingOperationKindEnvelope = type({ kind: "unknown" });
+
+/**
+ * Drop any restored pending operation whose `kind` is not a `SignalKind`
+ * this build still classifies, reporting each drop through `reportError`
+ * rather than letting the reactor's `rehydrateGates` throw on it. A retired
+ * signal kind (CL-7443's `message_response`) can still be sitting in a warm
+ * agent's on-disk checkpoint/WAL from before the retirement; without this
+ * filter, every respawn of that agent would throw
+ * "Unclassified signal kind" and never start. The person's in-flight
+ * question is simply lost here -- their eventual reply just becomes the
+ * next ordinary turn (docs/CHAT.md).
+ */
+function dropUnclassifiedPendingOperations(
+  items: unknown[],
+  agentKey: string,
+): unknown[] {
+  return items.filter((item) => {
+    const envelope = PendingOperationKindEnvelope(item);
+    const kind = envelope instanceof type.errors ? undefined : envelope.kind;
+    if (SignalKind(kind) instanceof type.errors) {
+      reportError(
+        new Error(
+          `pending operation with unclassified signal kind ${JSON.stringify(kind)} dropped on restore for agent ${agentKey}`,
+        ),
+        {
+          operation: "pending_op_dropped_unknown_kind",
+          agentId: agentKey,
+        },
+      );
+      return false;
+    }
+    return true;
+  });
+}
+
 /**
  * Reconstruct the warm agent's conversation from the two-tier on-disk
  * layout under `agentStateDir`
@@ -1050,7 +1097,7 @@ export async function reconstructDurableConversation(
   return {
     turns: castValidatedEnvelope<ConversationTurn>(turns),
     pendingOperations: castValidatedEnvelope<PendingOperation>(
-      metadata.pendingOperations,
+      dropUnclassifiedPendingOperations(metadata.pendingOperations, agentKey),
     ),
     tokenUsage: metadata.tokenUsage,
     connectorState: metadata.connectorState,