fix(paseo-omp): correlate terminal events by request

paseo-omp/SUPPORT.md

@@ -26,7 +26,7 @@ Do not put credentials, private repository paths, session transcripts, or unreda
 - `omp` and `omp-plugin` are independent provider identities. Agents, provider settings, and persisted handles do not migrate automatically between them.
 - Do not open the same underlying OMP session concurrently through both providers. Reservation tracking is provider-local and cannot coordinate ownership with Paseo's bundled adapter.
 - OMP 18.1.15 does not advertise typed tool approvals. The plugin uses its bounded generic permission fallback until both peers negotiate `typedToolApprovals: 1`.
-- OMP 18.2 does not positively correlate ordinary `prompt` requests with their eventual `agent_end`: the immediate response omits `agentInvoked: true`, `prompt_result` reports only local-only `false` results, and live user messages may omit their persisted `entryId`. Branch history can correlate user timeline entries, but even a new exact-text entry cannot authenticate a terminal event or its success/error outcome. Ownership-sensitive later turns require a request-matched positive prompt result observed before the terminal event; without that evidence, the plugin rejects the event and fails the turn closed after the ownership deadline. An unavailable state or history lookup cannot bypass that guard. Ordinary later turns on OMP 18.2 therefore remain blocked pending an upstream request-correlated completion contract.
+- OMP releases that do not correlate ordinary `prompt` requests with their terminal `agent_end` remain limited on ownership-sensitive later turns: the immediate response may omit `agentInvoked: true`, legacy `prompt_result` reports only local-only `false` results, and live user messages may omit their persisted `entryId`. Branch history can correlate user timeline entries, but even a new exact-text entry cannot authenticate a terminal event or its success/error outcome. The plugin accepts a later-turn terminal carrying the matching RPC `requestId`; without that identity, it rejects the event and fails the turn closed after the ownership deadline. An unavailable state or history lookup cannot bypass that guard.
 - Timeline correlation rebuilds an invalid watermark from a complete pre-prompt `get_branch_messages` snapshot, not replayed model context or an evicting identity cache. Snapshots are limited to 1,024 entries and 4 MiB; duplicate IDs, surplus exact-text matches, unavailable history, and exceeded bounds leave users uncorrelated rather than claiming an old entry. Repeated accepted prompts within a turn consume matching branch occurrences in order. User entries never grant terminal ownership to a later turn.
 - Configured MCP servers are supported and bridged into OMP. Paseo's own orchestration tools appear under their native names when the daemon's **Enable Paseo tools** / `daemon.mcp.injectIntoAgents` setting is enabled; other MCP servers remain namespaced. Exact Paseo `toolPolicy` preapproval cannot be represented by OMP `set_host_tools` and therefore fails session startup closed. `disallowedTools` applies only to recognized native OMP built-ins; unknown names are rejected and MCP tools are not silently filtered through it.
 - `qwen2.5:0.5b` is provided only for free exploratory inference. It may ignore exact-output instructions and is not a deterministic protocol or tool-use oracle; use `canary-mock/Deterministic Canary` for assertions.

paseo-omp/TESTING.md

@@ -32,7 +32,7 @@ Classifications:
 | Image prompts | **Equivalent** | Valid native image models receive image blocks; text-only models receive private content-addressed local files with an aggregate cap and turn/session/failure cleanup. The file path is valid because the direct provider and OMP child share the daemon host. |
 | Structured attachments | **Equivalent** | Forge change requests/issues, legacy GitHub forms, text, reviews, and uploaded files render to bounded OMP prompt text; regression in `tests/provider.test.ts`. |
 | Optimistic message correlation | **Equivalent** | Native entry lookup, repeated-text occurrence correlation, steering correlation, replay-boundary dedupe, and exactly-one `session.prompt_result` regressions. Bounded branch snapshots rebuild incomplete or evicted replay watermarks; duplicate IDs, surplus matching entries, unavailable snapshots, and count/byte overflow retain local user-message fallback instead of claiming an old entry. |
-| Terminal ownership on OMP 18.2 later turns | **Blocked** | A new exact-text branch entry proves user-message persistence, not ownership of an unkeyed `agent_end`. Later turns fail closed without a request-matched positive prompt result observed before the terminal frame. `tests/provider.test.ts` covers stale ends before and after new entries, with and without live echoes, repeated prompts, evidence arriving during an older terminal check, and unavailable/hung history. Ordinary OMP 18.2 prompts need an upstream request-correlated completion contract; no branch-based terminal compatibility is claimed. |
+| Terminal ownership on later turns | **Version-gated** | A new exact-text branch entry proves user-message persistence, not ownership of an unkeyed `agent_end`. The plugin accepts a later-turn terminal only when it carries the originating prompt's RPC `requestId`. Older OMP releases remain fail-closed after the ownership deadline. `tests/provider.test.ts` covers matching and mismatched request IDs, stale ends before and after new entries, repeated prompts, evidence arriving during an older terminal check, and unavailable or hung history. |
 | Streaming assistant text | **Equivalent** | `OmpTimelineProjector` publishes stable complete snapshots with frame coalescing and bounded retained bytes. |
 | Streaming reasoning | **Equivalent** | Indexed thinking blocks map to stable `reasoning` items and share stream bounds. |
 | `contentIndex` ordering | **Equivalent** | Stable 0→1→0 updates, sparse-index rejection, and 64-block bounds are tested in `tests/provider.test.ts`. |

paseo-omp/server/provider/omp-rpc.ts

@@ -570,6 +570,7 @@ const OmpToolApprovalResponseSchema = z.union([
 ]);
 const OmpAgentEndEnvelopeSchema = z.object({
   type: z.literal("agent_end"),
+  requestId: IDENTIFIER.optional(),
   messageCount: z.number().int().nonnegative().optional(),
   isTerminal: z.boolean().optional(),
 });
@@ -590,6 +591,7 @@ const OmpAgentSessionEventSchema = z.discriminatedUnion("type", [
   z.object({ type: z.literal("agent_start") }),
   z.object({
     type: z.literal("agent_end"),
+    requestId: IDENTIFIER.optional(),
     messages: z.array(OmpMessageSchema).max(MAX_ARRAY_ITEMS).optional(),
     messageCount: z.number().int().nonnegative().optional(),
     isTerminal: z.boolean().optional(),
@@ -1011,6 +1013,7 @@ export interface OmpRuntimeSession {
     message: string,
     images?: readonly OmpImage[],
     onAccepted?: () => void,
+    onRequested?: (requestId: string) => void,
   ): Promise<{ requestId: string; agentInvoked?: boolean }>;
   compact(customInstructions?: string): Promise<OmpCompactionResult>;
   setAutoCompaction(enabled: boolean): Promise<void>;
@@ -2639,6 +2642,7 @@ class OmpRpcSession implements OmpRuntimeSession {
     message: string,
     images: readonly OmpImage[] = [],
     onAccepted?: () => void,
+    onRequested?: (requestId: string) => void,
   ): Promise<{ requestId: string; agentInvoked?: boolean }> {
     const safeMessage = validateBoundedText(message, "prompt", MAX_TEXT_LENGTH);
     let acknowledgement: z.infer<typeof OmpPromptAckSchema> | undefined;
@@ -2650,6 +2654,7 @@ class OmpRpcSession implements OmpRuntimeSession {
         onAccepted?.();
       },
     );
+    onRequested?.(request.id);
     await request.promise;
     return { requestId: request.id, ...acknowledgement };
   }

paseo-omp/server/provider/session.ts

@@ -1912,9 +1912,16 @@ export class OmpProviderSession {
           return;
         }
       }
-      const acknowledgement = await runtime.prompt(payload.text, payload.images, () => {
-        turn.promptAcceptedEventIndex ??= turn.bufferedEvents.length;
-      });
+      const acknowledgement = await runtime.prompt(
+        payload.text,
+        payload.images,
+        () => {
+          turn.promptAcceptedEventIndex ??= turn.bufferedEvents.length;
+        },
+        (requestId) => {
+          turn.nativeRequestId = requestId;
+        },
+      );
       if (this.closed || turn.terminal) return;
       turn.nativeRequestId = acknowledgement.requestId;
       this.publishPromptResult(turn, { type: "turn", turnId: turn.turnId });
@@ -3109,6 +3116,13 @@ export class OmpProviderSession {
       }
       return;
     }
+    if (
+      event.type === "agent_end" &&
+      event.requestId !== undefined &&
+      event.requestId !== turn.nativeRequestId
+    ) {
+      return;
+    }
     if (turn.starting) {
       if (
         turn.bufferedEvents.length >= MAX_BUFFERED_TURN_EVENTS ||
@@ -3140,6 +3154,13 @@ export class OmpProviderSession {
     ) {
       return;
     }
+    if (
+      event.type === "agent_end" &&
+      event.requestId !== undefined &&
+      event.requestId !== turn.nativeRequestId
+    ) {
+      return;
+    }
     if (event.type === "message_end") {
       const entryId = nativeEntryId(event.message);
       if (!entryId || turn.streamedMessageEntryIds.length >= MAX_AGENT_END_CORRELATION_MESSAGES) {
@@ -3251,7 +3272,15 @@ export class OmpProviderSession {
       turn.awaitingPermissionEvidence = false;
     }
     if (event.type === "agent_end") {
-      if (event.isTerminal === false) return;
+      if (event.isTerminal === false) {
+        turn.completedMessageCount = 0;
+        turn.streamedMessageEntryIds.length = 0;
+        turn.streamedMessageIdentityComplete = true;
+        turn.lastCompletedAssistantOutcome = undefined;
+        turn.lastCompletedAssistantEntryId = undefined;
+        return;
+      }
+      if (event.requestId !== undefined) this.markTerminalOwnershipEvidence(turn);
       if (
         !turn.interrupted &&
         turn.terminalOwnershipRequired &&

paseo-omp/tests/omp-rpc.test.ts

@@ -223,6 +223,20 @@ describe("OMP RPC transport", () => {
       agentInvoked: false,
     });
 
+    const correlatedEnd = nextEvent((listener) => session.onEvent(listener));
+    child.write({
+      type: "agent_end",
+      requestId: "prompt-2",
+      messages: [],
+      isTerminal: true,
+    });
+    await expect(correlatedEnd).resolves.toEqual({
+      type: "agent_end",
+      requestId: "prompt-2",
+      messages: [],
+      isTerminal: true,
+    });
+
     await session.steer("focus");
     await session.followUp("verify");
     await session.setAutoCompaction(false);