fix(orchestrator): fail closed on cross-conversation run identity

src/agent/spawn.ts

@@ -91,7 +91,7 @@ import { clearNativeStartFailure, nativeStartFailure, recordNativeStartFailure }
 import { asCliEventRecord, discriminate, fieldString, type CliEventRecord } from '../types/cli-events.js';
 import { isRemoteTarget, type RemoteTarget } from '../messaging/types.js';
 import { buildRemoteBindingKey } from '../messaging/session-key.js';
-import { runPinFields } from '../messaging/run-pin.js';
+import { runPinFields, sameRunConversation } from '../messaging/run-pin.js';
 import { isRetiredCliSelection, retiredRuntimeDiagnostic } from '../types/cli-engine.js';
 import { runBeforeSpawnChecks, type PolicyVerdict } from '../core/policy-hooks.js';
 import { appendTraceEvent, createTraceId, finalizeTraceRun, stampTraceTool, startTraceRun, updateTraceToolRow } from '../trace/store.js';
@@ -798,6 +798,12 @@ export function canSteerAgent(scopeKey: string): boolean {
     return typeof run?.steerTurnInBand === 'function' || typeof run?.replaceTurn === 'function';
 }
 
+/** Native replacement owns a stricter mismatch contract than queue fallback:
+ * malformed cross-conversation metadata must fail before it is persisted. */
+export function hasActiveMainReplacement(scopeKey: string): boolean {
+    return typeof activeMainProcesses.get(scopeKey)?.replaceTurn === 'function';
+}
+
 export type SteerOutcome = 'steered' | 'fallback-queue' | 'new-run' | 'cancelled' | 'retired';
 
 export async function steerAgent(
@@ -866,6 +872,19 @@ export async function steerAgent(
             sessionId: chatSessionId, reason: outcome.reason, requestId: capturedMeta.requestId }));
         return 'fallback-queue';
     }
+    if (run && !sameRunConversation(
+        { origin: run.meta.origin, remoteKey: run.meta.remoteKey
+            ?? (isRemoteTarget(run.meta.target) ? buildRemoteBindingKey(run.meta.target) : undefined) },
+        { origin: source, remoteKey: meta?.remoteKey
+            ?? (isRemoteTarget(meta?.target) ? buildRemoteBindingKey(meta.target) : undefined) },
+    )) {
+        // In-band and kill-steer mutate the turn already owned by `run`.
+        // Different remote keys are different conversations, even when a legacy
+        // scope collapse put them in the same process slot. Let the caller queue
+        // a separate follow-up instead of giving this run another user's input
+        // and delivery address (#743).
+        return 'fallback-queue';
+    }
     if (typeof run?.steerTurnInBand === 'function') {
         // codex-app same-turn steer. The user row is written only AFTER the
         // server accepts — a fallback must not leave a duplicate insert for the

src/agent/spawn/queue.ts

@@ -9,6 +9,7 @@ import { sessionLanes, type SessionLanes } from '../../orchestrator/session-lane
 import { scopeForChatSession } from '../../orchestrator/scope.js';
 import { settleOnce } from '../../orchestrator/request-registry.js';
 import { readSlackWorkflowMetadata, type SlackWorkflowMetadata } from '../../slack/workflow.js';
+import { runPinFields } from '../../messaging/run-pin.js';
 
 type QueueItem = {
     slackWorkflow?: SlackWorkflowMetadata;
@@ -452,7 +453,14 @@ export function createQueueController(
         console.log(`[queue] +1 (${messageQueue.length} pending)`);
         deps.broadcast('queue_update', {
             ...queueUpdatePayload(item.scope),
-            ...(item.requestId ? { requestId: item.requestId, origin: item.source || 'web' } : {}),
+            ...(item.requestId ? runPinFields({
+                requestId: item.requestId,
+                origin: item.source || 'web',
+                scope: item.scope,
+                sessionId: item.chatSessionId,
+                remoteKey: item.remoteKey,
+                target: item.target,
+            }) : {}),
         });
         void processQueue(item.scope);
         return item.id;
@@ -519,7 +527,10 @@ export function createQueueController(
                 return;
             }
             const collectedItems = multiSessionEnabled && item!.collect
-                ? messageQueue.filter(candidate => normalizeScope(candidate.scope) === itemScope && candidate.collect === true && !scheduledItemIds.has(candidate.id))
+                ? messageQueue.filter(candidate => normalizeScope(candidate.scope) === itemScope
+                    && candidate.collect === true
+                    && candidate.remoteKey === item!.remoteKey
+                    && !scheduledItemIds.has(candidate.id))
                 : [];
             const runItems = [item!, ...collectedItems];
             const runIds = new Set(runItems.map(candidate => candidate.id));
@@ -585,6 +596,7 @@ export function createQueueController(
             // turns with the same duplicate to prevent.
             if (requestId) deps.broadcast('queued_run_started', stripUndefined({
                 requestId, origin, scope: item.scope, target, sessionId: effectiveSessionId,
+                remoteKey: item.remoteKey,
                 slackWorkflow: item.slackWorkflow,
             }));
             deps.broadcast('queue_update', queueUpdatePayload(item.scope));
@@ -614,7 +626,11 @@ export function createQueueController(
                 } catch (err: unknown) {
                     const msg = (err as Error).message;
                     console.error('[queue:orchestrate]', msg);
-                    deps.broadcast('orchestrate_done', { text: `[error] ${msg}`, error: true, origin, chatId, target, requestId, replyViaTarget, fromQueue: true, ...(eventScope || {}) });
+                    deps.broadcast('orchestrate_done', {
+                        ...runPinFields({ requestId, origin, scope: item.scope, sessionId: effectiveSessionId,
+                            remoteKey: item.remoteKey, target }),
+                        text: `[error] ${msg}`, error: true, chatId, replyViaTarget, fromQueue: true,
+                    });
                     // The pipeline threw before reaching its own settle site, so
                     // this is the last place that can answer the caller.
                     settleOnce(requestId, 'failed', { error: msg });
@@ -625,8 +641,12 @@ export function createQueueController(
             if (!inserted) {
                 messageQueue.unshift(...runItems);
             } else {
-                deps.broadcast('orchestrate_done', { text: `[error] setup failed: ${(setupErr as Error).message}`, error: true, origin, chatId, target, requestId, replyViaTarget, fromQueue: true,
-                    ...(multiSessionEnabled ? { scope: item.scope, sessionId: effectiveSessionId } : {}) });
+                deps.broadcast('orchestrate_done', {
+                    ...runPinFields({ requestId, origin, scope: item.scope, sessionId: effectiveSessionId,
+                        remoteKey: item.remoteKey, target }),
+                    text: `[error] setup failed: ${(setupErr as Error).message}`, error: true,
+                    chatId, replyViaTarget, fromQueue: true,
+                });
                 // Re-queued items settle on their eventual run; these do not get
                 // another chance, so answer the caller here.
                 settleOnce(requestId, 'failed', { error: `setup failed: ${(setupErr as Error).message}` });

src/messaging/run-pin.ts

@@ -15,7 +15,7 @@
 // Both are the same mistake in different clothes: inferring a run's context
 // from process-global state instead of carrying it. A pin is carried.
 
-import type { RemoteTarget } from './types.js';
+import { isRemoteTarget, type RemoteTarget } from './types.js';
 
 export interface RunPin {
     origin?: string | undefined;
@@ -41,7 +41,7 @@ export function runPinFields(pin: RunPin): Record<string, unknown> {
     if (pin.scope) fields['scope'] = pin.scope;
     if (pin.sessionId) fields['sessionId'] = pin.sessionId;
     if (pin.remoteKey) fields['remoteKey'] = pin.remoteKey;
-    if (pin.target && typeof pin.target.targetId === 'string' && pin.target.targetId) {
+    if (isRemoteTarget(pin.target)) {
         fields['target'] = { ...pin.target };
     }
     return fields;
@@ -63,3 +63,35 @@ export function sameRunConversation(a: RunPin, b: RunPin): boolean {
     if (a.remoteKey || b.remoteKey) return a.remoteKey === b.remoteKey;
     return (a.origin ?? '') === (b.origin ?? '');
 }
+
+function sameTarget(expected: RemoteTarget, actual: unknown): boolean {
+    return isRemoteTarget(actual)
+        && actual.channel === expected.channel
+        && actual.targetKind === expected.targetKind
+        && actual.peerKind === expected.peerKind
+        && actual.targetId === expected.targetId
+        && actual.threadId === expected.threadId
+        && actual.guildId === expected.guildId
+        && actual.parentTargetId === expected.parentTargetId;
+}
+
+/**
+ * Match a terminal/control event to the run waiting for it.
+ *
+ * requestId, origin, scope and sessionId are the minimum identity. Treating an
+ * absent field as agreement is the bug: an old print terminal omitted all four
+ * except origin, and a newly installed Slack waiter could adopt it (#743).
+ * Remote conversation and destination become mandatory when the waiter owns
+ * them. This lets local/web runs keep their smaller identity while remote runs
+ * fail closed on the extra fields that separate one conversation from another.
+ */
+export function matchesRunPin(expected: RunPin, actual: Record<string, unknown>): boolean {
+    if (!expected.requestId || !expected.origin || !expected.scope || !expected.sessionId) return false;
+    if (actual['requestId'] !== expected.requestId
+        || actual['origin'] !== expected.origin
+        || actual['scope'] !== expected.scope
+        || actual['sessionId'] !== expected.sessionId) return false;
+    if (expected.remoteKey !== undefined && actual['remoteKey'] !== expected.remoteKey) return false;
+    if (expected.target !== undefined && !sameTarget(expected.target, actual['target'])) return false;
+    return true;
+}

src/orchestrator/collect.ts

@@ -16,6 +16,8 @@ import { settings } from '../core/config.js';
 import { getActiveChatSession } from '../core/chat-sessions.js';
 import { currentSessionScope } from '../core/session-context.js';
 import { resolveExecutionBinding } from './scope.js';
+import { matchesRunPin } from '../messaging/run-pin.js';
+import { isRemoteTarget } from '../messaging/types.js';
 
 export interface CollectedOrchestrateResult {
     text: string;
@@ -51,6 +53,16 @@ export function orchestrateAndCollectData(
         const runMeta = { ...meta, ...binding, origin: meta['origin'] || 'web',
             _onRuntimeActivity: onRuntimeActivity };
         const requestId = meta['requestId'] || undefined;
+        const strictTerminalPin = runMeta.origin === 'slack' && requestId
+            ? {
+                requestId,
+                origin: 'slack',
+                scope: binding.scope,
+                sessionId: binding.chatSessionId,
+                remoteKey: meta['remoteKey'],
+                target: isRemoteTarget(meta['target']) ? { ...meta['target'] } : undefined,
+            }
+            : null;
         let collected = '';
         let ownTerminalDiagnostic = '';
         let nativeSeen = false;
@@ -131,11 +143,14 @@ export function orchestrateAndCollectData(
             // must differ — a steer carrying our own id is not a supersession.
             if (type === 'steer_started'
                 && data['scope'] === binding.scope
-                && (data['sessionId'] === undefined || data['sessionId'] === binding.chatSessionId)
+                && data['sessionId'] === binding.chatSessionId
+                && data['origin'] === runMeta.origin
+                && (meta['remoteKey'] === undefined || data['remoteKey'] === meta['remoteKey'])
                 && (!requestId || data['requestId'] !== requestId)) {
                 superseded = true;
             }
             if (type === 'orchestrate_done') {
+                if (strictTerminalPin && !matchesRunPin(strictTerminalPin, data)) return;
                 // Filter by requestId (strongest), then origin, then chatId
                 if (meta?.["requestId"] && data?.["requestId"] && data["requestId"] !== meta["requestId"]) return;
                 if (meta?.["origin"] && data?.["origin"] && data["origin"] !== meta["origin"]) return;

src/orchestrator/gateway.ts

@@ -3,7 +3,7 @@
 // Replaces duplicated intent/queue/orchestrate logic in server.ts + bot.ts.
 
 import { randomUUID } from 'node:crypto';
-import { isAgentBusy, enqueueMessage, killActiveAgent, messageQueue, purgeQueueOnStop, steerAgent } from '../agent/spawn.js';
+import { getCurrentMainMeta, hasActiveMainReplacement, isAgentBusy, enqueueMessage, killActiveAgent, messageQueue, purgeQueueOnStop, steerAgent } from '../agent/spawn.js';
 import { hasBlockingWorkers } from './worker-registry.js';
 import { getSession, insertMessage } from '../core/db.js';
 import { resolveMainCli, type MainSessionRecord } from '../core/main-session.js';
@@ -19,12 +19,13 @@ import {
 } from './pipeline.js';
 import { getState } from './state-machine.js';
 import { channelGateOn, resolveOrcScope } from './scope.js';
-import type { RuntimeOrigin, RemoteTarget } from '../messaging/types.js';
+import { isRemoteTarget, type RuntimeOrigin, type RemoteTarget } from '../messaging/types.js';
 import { buildRemoteBindingKey, normalizedThreadId, type SessionScope } from '../messaging/session-key.js';
 import { sessionLanes } from './session-lanes.js';
 import { admitRequest, settleOnce } from './request-registry.js';
 import { beginSteerInput } from '../agent/steer-input-guard.js';
 import type { SlackWorkflowMetadata } from '../slack/workflow.js';
+import { sameRunConversation } from '../messaging/run-pin.js';
 
 export type SubmitResult = {
     action: 'started' | 'queued' | 'rejected';
@@ -102,6 +103,27 @@ function applyMidRunPolicy(
     };
 
     if (policy === 'steer') {
+        const owner = getCurrentMainMeta(ctx.scopeKey);
+        if (owner && !sameRunConversation(
+            { origin: owner.origin, remoteKey: owner.remoteKey
+                ?? (isRemoteTarget(owner.target) ? buildRemoteBindingKey(owner.target) : undefined) },
+            { origin: ctx.meta.origin, remoteKey: ctx.remoteKey
+                ?? (isRemoteTarget(ctx.meta.target) ? buildRemoteBindingKey(ctx.meta.target) : undefined) },
+        )) {
+            // Native replacement validates every owner dimension and returns a
+            // typed failure. Queueing here would preserve the owner's
+            // scope/session/target while swapping only remoteKey, which stores
+            // another conversation's prompt in the owner's transcript.
+            if (hasActiveMainReplacement(ctx.scopeKey)) {
+                // Fall through to steerAgent; its immutable owner check rejects.
+            } else {
+            // Steer changes the turn that is already running. A different
+            // remoteKey is a different conversation, so applying its text to
+            // this owner would also hand it the owner's progress and terminal
+            // delivery. Queue it as its own follow-up instead (#743).
+                return queue();
+            }
+        }
         // 'steer' means the message steers the agent — never a silent queue.
         // A steerable Codex App turn receives in-band input. Native Cursor/Grok
         // use their cancel-reprompt hooks; other runtimes take the kill-steer
@@ -171,7 +193,8 @@ export function __resetSubmitDedupForTest(): void {
 function runDetached(
     task: Promise<unknown>,
     label: string,
-    meta: { origin: RuntimeOrigin; target?: RemoteTarget; chatId?: string | number; requestId?: string; replyViaTarget?: boolean; eventScope?: { scope: string; sessionId: string } },
+    meta: { origin: RuntimeOrigin; target?: RemoteTarget; chatId?: string | number; requestId?: string;
+        remoteKey?: string; replyViaTarget?: boolean; eventScope?: { scope: string; sessionId: string } },
 ) {
     task.catch((err: unknown) => {
         const msg = (err as Error)?.message || String(err);
@@ -182,6 +205,7 @@ function runDetached(
             target: meta.target,
             chatId: meta.chatId,
             requestId: meta.requestId,
+            ...(meta.remoteKey ? { remoteKey: meta.remoteKey } : {}),
             replyViaTarget: meta.replyViaTarget,
             ...meta.eventScope,
             error: true,
@@ -231,15 +255,22 @@ export function submitMessage(
     // Admit the request the moment its id exists. Every exit below then settles
     // through settleOnce(), so a caller holding this id always hears exactly one
     // terminal event — including on paths that never emit orchestrate_done.
-    admitRequest(requestId, scope);
+    admitRequest(requestId, scope, Date.now(), {
+        origin: meta.origin,
+        sessionId: chatSessionId,
+        ...(remoteKey ? { remoteKey } : {}),
+        ...(meta.target ? { target: meta.target } : {}),
+    });
     try { meta.onAdmitted?.({ requestId, scope, chatSessionId }); }
     catch { settleOnce(requestId, 'failed', { error: 'slack_tool_context_unavailable' }); return { action: 'rejected', reason: 'slack_tool_context_unavailable', requestId }; }
     // OFF-mode byte-compat: only expose resolved identity when multi-session is on —
     // /api/message spreads SubmitResult into the HTTP response (routes/command.ts).
     const sessionContext = multiSessionEnabled
         ? { scope, chatSessionId, ...(remoteKey ? { remoteKey } : {}) }
         : undefined;
-    const eventScope = multiSessionEnabled ? { scope, sessionId: chatSessionId } : undefined;
+    const eventScope = multiSessionEnabled || meta.origin === 'slack'
+        ? { scope, sessionId: chatSessionId }
+        : undefined;
 
     // Reject before recording input, steering, interrupting or enqueueing. The
     // synchronous response must not advertise a rejected admission as steered.