Deterministic delegation receipts: route delegated outcomes back to the delegator

packages/agent-connector/src/adapters/aider.js

@@ -449,7 +449,7 @@ class AiderAdapter extends BaseAdapter {
 
     if (!this._aiderBin) this._aiderBin = this._findAiderBinary();
     if (!this._aiderBin) {
-      await this.sendError(msgChannel, `Aider CLI not found. Install with: ${aiderInstallHint()}`);
+      await this.sendFinalError(msg, msgChannel, `Aider CLI not found. Install with: ${aiderInstallHint()}`);
       return;
     }
 
@@ -458,7 +458,7 @@ class AiderAdapter extends BaseAdapter {
     // injects the key into the wrong provider).
     const resolution = this._resolveConfig();
     if (resolution.error) {
-      await this.sendError(msgChannel, `Configuration error: ${resolution.error}`);
+      await this.sendFinalError(msg, msgChannel, `Configuration error: ${resolution.error}`);
       return;
     }
 
@@ -472,7 +472,7 @@ class AiderAdapter extends BaseAdapter {
       result = await this._runAider(content, msgChannel);
     } catch (e) {
       this._log(`Error handling message: ${e.message}`);
-      await this.sendError(msgChannel, `Error processing message: ${e.message}`);
+      await this.sendFinalError(msg, msgChannel, `Error processing message: ${e.message}`);
       return;
     }
 
@@ -482,11 +482,13 @@ class AiderAdapter extends BaseAdapter {
     }
     const { text, error } = result;
     if (error) {
-      await this.sendError(msgChannel, error);
+      await this.sendFinalError(msg, msgChannel, error);
     } else if (text) {
-      await this.sendResponse(msgChannel, text);
+      await this.sendFinalResult(msg, msgChannel, text);
     } else {
-      await this.sendResponse(
+      // No text but the run succeeded (file edits were applied) — a result.
+      await this.sendFinalResult(
+        msg,
         msgChannel,
         'Aider finished with no textual output (any file changes were applied to the working directory).',
       );

packages/agent-connector/src/adapters/amp.js

@@ -246,7 +246,7 @@ class AmpAdapter extends BaseAdapter {
     if (!this._ampBin) {
       const message = `Amp CLI not found — install with: ${ampInstallHint()}`;
       this._reportStatus(REASON.RUNTIME_MISSING, message);
-      await this.sendError(msgChannel, message);
+      await this.sendFinalError(msg, msgChannel, message);
       return;
     }
 
@@ -269,10 +269,11 @@ class AmpAdapter extends BaseAdapter {
         });
         this._log(message);
         this._reportStatus(reason, message);
-        await this.sendError(msgChannel, message);
+        await this.sendFinalError(msg, msgChannel, message);
       } else {
         this._log(`Error handling message: ${redactDiagnostic(e.message)}`);
-        await this.sendError(
+        await this.sendFinalError(
+          msg,
           msgChannel,
           `Error processing message: ${redactDiagnostic(e.message)}`,
         );
@@ -288,9 +289,9 @@ class AmpAdapter extends BaseAdapter {
     // spawn/runtime error the agent row may be showing.
     this._reportStatus(null);
     if (responseText) {
-      await this.sendResponse(msgChannel, responseText);
+      await this.sendFinalResult(msg, msgChannel, responseText);
     } else {
-      await this.sendResponse(msgChannel, 'No response generated. Please try again.');
+      await this.sendFinalError(msg, msgChannel, 'No response generated. Please try again.');
     }
   }
 

packages/agent-connector/src/adapters/base.js

@@ -80,6 +80,12 @@ class BaseAdapter {
     // Per-channel task tracking for parallel execution
     this._channelBusy = new Set();
     this._channelQueues = {};
+    // Inbound message currently being handled per channel. Strict lifecycle:
+    // set right before _handleMessage, cleared right after. Consumed ONLY by
+    // cancellation paths (control "stop" has no trigger of its own) — never
+    // use it to attribute late/background output, which can arrive after the
+    // next message has already replaced the entry.
+    this._inflightTurns = {};
     // Cached workspace.browser_enabled. Populated lazily on first read so we
     // don't pay an HTTP roundtrip per message — adapters that toggle the
     // workspace flag must reconnect/restart to pick up the change (matches
@@ -746,11 +752,12 @@ class BaseAdapter {
 
   async _channelWorker(channel, msg) {
     this._channelBusy.add(channel);
+    this._inflightTurns[channel] = msg;
     try {
       await this._handleMessage(msg);
     } catch (e) {
       this._log(`Error in channel worker for ${channel}: ${e.message}`);
-      try { await this.sendError(channel, `Agent error: ${e.message}`); } catch {}
+      try { await this.sendFinalError(msg, channel, `Agent error: ${e.message}`); } catch {}
     }
 
     // Drain queue
@@ -761,13 +768,15 @@ class BaseAdapter {
       if (nextMsg._queueId) {
         try { await this.sendStatus(channel, 'processing queued message', { queue_id: nextMsg._queueId, queue_status: 'processed' }); } catch {}
       }
+      this._inflightTurns[channel] = nextMsg;
       try {
         await this._handleMessage(nextMsg);
       } catch (e) {
         this._log(`Error processing queued message in ${channel}: ${e.message}`);
-        try { await this.sendError(channel, `Agent error: ${e.message}`); } catch {}
+        try { await this.sendFinalError(nextMsg, channel, `Agent error: ${e.message}`); } catch {}
       }
     }
+    delete this._inflightTurns[channel];
     this._channelBusy.delete(channel);
   }
 
@@ -844,6 +853,77 @@ class BaseAdapter {
     }
   }
 
+  // ------------------------------------------------------------------
+  // Terminal replies (delegation receipts)
+  // ------------------------------------------------------------------
+  //
+  // Unlike sendResponse, these stamp the outgoing chat message with
+  // reply_kind + in_reply_to so the backend can deterministically route a
+  // delegated task's outcome back to the delegating agent instead of
+  // relying on the LLM router to guess "this is a report". `triggerMsg` is
+  // the inbound message that started the turn. Synthetic triggers
+  // (system:* senders, missing event id) degrade to a plain reply.
+  //
+  // reply_kind is only stamped when the trigger carries the SERVER-written
+  // delegation fields naming this agent: the UI renders its receipt badge
+  // purely from reply_kind, so stamping ordinary human-triggered replies
+  // would present them as delegation receipts. in_reply_to alone is kept
+  // for every real trigger — it is plain correlation data.
+
+  _receiptMeta(kind, triggerMsg) {
+    const meta = {};
+    const id = triggerMsg && triggerMsg.messageId;
+    const sender = (triggerMsg && triggerMsg.senderName) || '';
+    if (!id || sender.startsWith('system:')) return meta;
+    meta.in_reply_to = id;
+    const tm = (triggerMsg && triggerMsg.metadata) || {};
+    const delegatedTo = Array.isArray(tm.delegated_to) ? tm.delegated_to : [];
+    if (tm.delegated_by && delegatedTo.includes(this.agentName)) {
+      meta.reply_kind = kind;
+    }
+    return meta;
+  }
+
+  async _sendTerminal(kind, triggerMsg, channel, content) {
+    try {
+      await this.client.sendMessage(this.workspaceId, channel, this.token, content, {
+        senderType: 'agent',
+        senderName: this.agentName,
+        metadata: this._receiptMeta(kind, triggerMsg),
+        sessionId: this._sessionId,
+      });
+    } catch (e) {
+      if (e instanceof SessionRevokedError) {
+        this._onSessionRevoked();
+        return;
+      }
+      // Result delivery failures must surface to the caller (matching
+      // sendResponse); error/cancel/needs-input paths swallow like sendError
+      // so a failing receipt can't mask the original problem.
+      if (kind === 'result') throw e;
+    }
+  }
+
+  /** The turn's final answer. */
+  async sendFinalResult(triggerMsg, channel, content) {
+    return this._sendTerminal('result', triggerMsg, channel, content);
+  }
+
+  /** The turn failed — the delegator must not wait forever. */
+  async sendFinalError(triggerMsg, channel, content) {
+    return this._sendTerminal('error', triggerMsg, channel, content);
+  }
+
+  /** The agent needs more input before it can continue. */
+  async sendNeedsInput(triggerMsg, channel, content) {
+    return this._sendTerminal('needs_input', triggerMsg, channel, content);
+  }
+
+  /** The turn was cancelled (user stop). */
+  async sendCancelled(triggerMsg, channel, content) {
+    return this._sendTerminal('cancelled', triggerMsg, channel, content);
+  }
+
   async cleanupTodos(channel) {
     try {
       const result = await this.client.getTodos(this.workspaceId, channel, this.token, {

packages/agent-connector/src/adapters/claude.js

@@ -349,7 +349,11 @@ class ClaudeAdapter extends BaseAdapter {
   async _postStopNotice(channel) {
     if (!channel || this._stopNoticeSent.has(channel)) return;
     this._stopNoticeSent.add(channel);
-    try { await this.sendResponse(channel, 'Execution stopped by user.'); } catch {}
+    // The control-action stop has no inbound message of its own; the
+    // inflight-turn registry supplies the turn being cancelled so a
+    // delegating agent gets a `cancelled` receipt instead of silence.
+    const trigger = this._inflightTurns[channel] || null;
+    try { await this.sendCancelled(trigger, channel, 'Execution stopped by user.'); } catch {}
   }
 
   async _stopAllProcesses(completionMessage = 'Execution stopped.') {
@@ -365,7 +369,7 @@ class ClaudeAdapter extends BaseAdapter {
       delete this._channelProcesses[channel];
       delete this._channelQueues[channel];
       try {
-        await this.sendResponse(channel, completionMessage);
+        await this.sendCancelled(this._inflightTurns[channel] || null, channel, completionMessage);
       } catch {}
     }
   }
@@ -1076,18 +1080,18 @@ class ClaudeAdapter extends BaseAdapter {
    * error, or — when the CLI ended with neither — a session-reset notice so
    * the UI never hangs on "thinking…".
    */
-  async _postTurnOutcome(pp, msgChannel, finalResponse) {
+  async _postTurnOutcome(pp, msgChannel, finalResponse, triggerMsg) {
     if (finalResponse) {
-      try { await this.sendResponse(msgChannel, finalResponse); } catch {}
+      try { await this.sendFinalResult(triggerMsg, msgChannel, finalResponse); } catch {}
     } else if (pp.lastErrorText) {
-      try { await this.sendError(msgChannel, this._formatClaudeError(pp.lastErrorText)); } catch {}
+      try { await this.sendFinalError(triggerMsg, msgChannel, this._formatClaudeError(pp.lastErrorText)); } catch {}
     } else if (!pp.everPostedAnything) {
       if (this._channelSessions[msgChannel]) {
         delete this._channelSessions[msgChannel];
         try { this._saveSessions(); } catch {}
         this._log(`Empty-error result — cleared session for ${msgChannel}`);
       }
-      try { await this.sendError(msgChannel, 'The agent hit an error and could not respond. The session was reset — please send the message again.'); } catch {}
+      try { await this.sendFinalError(triggerMsg, msgChannel, 'The agent hit an error and could not respond. The session was reset — please send the message again.'); } catch {}
     }
   }
 
@@ -1196,7 +1200,7 @@ class ClaudeAdapter extends BaseAdapter {
             // so it rebuilds context from the channel recap.
             await this._resetSessionForPromptTooLong(msgChannel);
           } else {
-            await this._postTurnOutcome(existingPP, msgChannel, finalResponse);
+            await this._postTurnOutcome(existingPP, msgChannel, finalResponse, msg);
             this._resetIdleTimer(msgChannel);
             await this._queueTodoNudge(msgChannel, msg);
             return;
@@ -1292,7 +1296,7 @@ class ClaudeAdapter extends BaseAdapter {
         cmd = built.cmd;
         mcpConfigFile = built.mcpConfigFile;
       } catch (e) {
-        await this.sendError(msgChannel, e.message);
+        await this.sendFinalError(msg, msgChannel, e.message);
         return;
       }
 
@@ -1331,9 +1335,9 @@ class ClaudeAdapter extends BaseAdapter {
           }
           if (!pp.everPostedAnything) {
             if (pp.lastErrorText) {
-              try { await this.sendError(msgChannel, this._formatClaudeError(pp.lastErrorText)); } catch {}
+              try { await this.sendFinalError(msg, msgChannel, this._formatClaudeError(pp.lastErrorText)); } catch {}
             } else {
-              try { await this.sendResponse(msgChannel, 'No response generated. Please try again.'); } catch {}
+              try { await this.sendFinalError(msg, msgChannel, 'No response generated. Please try again.'); } catch {}
             }
           }
           break;
@@ -1346,13 +1350,13 @@ class ClaudeAdapter extends BaseAdapter {
           continue;
         }
 
-        await this._postTurnOutcome(pp, msgChannel, finalResponse);
+        await this._postTurnOutcome(pp, msgChannel, finalResponse, msg);
         this._resetIdleTimer(msgChannel);
         await this._queueTodoNudge(msgChannel, msg);
         break;
       } catch (e) {
         this._log(`Error handling message: ${e.message}`);
-        await this.sendError(msgChannel, `Error processing message: ${e.message}`);
+        await this.sendFinalError(msg, msgChannel, `Error processing message: ${e.message}`);
         break;
       }
     }

packages/agent-connector/src/adapters/cline.js

@@ -139,7 +139,7 @@ class ClineAdapter extends BaseAdapter {
         await this._stopProcess(this._channelProcesses[channel]);
         delete this._channelProcesses[channel];
         delete this._channelQueues[channel];
-        try { await this.sendResponse(channel, 'Execution stopped by user.'); } catch {}
+        try { await this.sendCancelled(this._inflightTurns[channel] || null, channel, 'Execution stopped by user.'); } catch {}
       } else {
         await this._stopAllProcesses('Execution stopped by user.');
       }
@@ -187,7 +187,7 @@ class ClineAdapter extends BaseAdapter {
       await this._stopProcess(proc);
       delete this._channelProcesses[channel];
       delete this._channelQueues[channel];
-      try { await this.sendResponse(channel, message); } catch {}
+      try { await this.sendCancelled(this._inflightTurns[channel] || null, channel, message); } catch {}
     }
   }
 
@@ -485,13 +485,13 @@ class ClineAdapter extends BaseAdapter {
     // back to the launcher/repo dir — return a clear error instead.
     const workingDir = this.workingDir || defaultAgentWorkdir(this.agentName);
     if (this.workingDir && !this._dirExists(this.workingDir)) {
-      await this.sendError(channel, `Working directory does not exist: ${this.workingDir}`);
+      await this.sendFinalError(msg, channel, `Working directory does not exist: ${this.workingDir}`);
       return;
     }
 
     const clineBin = this._findClineBinary();
     if (!clineBin) {
-      await this.sendError(channel,
+      await this.sendFinalError(msg, channel,
         'Cline CLI not found. Install it with: npm install -g cline');
       return;
     }
@@ -501,7 +501,7 @@ class ClineAdapter extends BaseAdapter {
     const ver = this._checkClineVersion(clineBin);
     if (ver.compatible === false) {
       this._log(`Refusing to start: Cline ${ver.version} < minimum ${MIN_CLINE_VERSION}`);
-      await this.sendError(channel,
+      await this.sendFinalError(msg, channel,
         `Cline CLI ${ver.version} is below the minimum supported version ${MIN_CLINE_VERSION}. ` +
         'Please upgrade with: npm install -g cline@latest');
       return;
@@ -560,7 +560,7 @@ class ClineAdapter extends BaseAdapter {
       });
 
       const spawnStartMs = Date.now();
-      const result = await this._runCline(channel, clineBin, args, workingDir);
+      const result = await this._runCline(channel, clineBin, args, workingDir, msg);
 
       if (result.userStopped) return;
 
@@ -583,12 +583,12 @@ class ClineAdapter extends BaseAdapter {
 
       // Emit final response or a classified error.
       if (result.finalText) {
-        try { await this.sendResponse(channel, result.finalText); } catch {}
+        try { await this.sendFinalResult(msg, channel, result.finalText); } catch {}
       } else if (result.errorMessage) {
         const { kind, userMessage } = classifyClineError(result.errorMessage);
-        try { await this.sendError(channel, this._withAuthHint(userMessage, kind)); } catch {}
+        try { await this.sendFinalError(msg, channel, this._withAuthHint(userMessage, kind)); } catch {}
       } else if (!result.anyOutput) {
-        try { await this.sendResponse(channel, 'No response generated. Please try again.'); } catch {}
+        try { await this.sendFinalError(msg, channel, 'No response generated. Please try again.'); } catch {}
       }
       return;
     }
@@ -626,7 +626,7 @@ class ClineAdapter extends BaseAdapter {
    *   { ok, finalText, errorMessage, anyOutput, userStopped }
    * `ok` reflects run_result.finishReason === "completed".
    */
-  _runCline(channel, clineBin, args, workingDir) {
+  _runCline(channel, clineBin, args, workingDir, triggerMsg) {
     const cleanEnv = { ...(this.agentEnv || process.env) };
     const [cmd, ...spawnArgs] = this._spawnableCmd(clineBin, args);
 
@@ -727,7 +727,7 @@ class ClineAdapter extends BaseAdapter {
               state.anyOutput = true;
               {
                 const opts = e.options && e.options.length ? `\n\nOptions: ${e.options.join(' · ')}` : '';
-                try { await this.sendResponse(channel, `❓ ${e.question || 'The agent is asking for input.'}${opts}`); } catch {}
+                try { await this.sendNeedsInput(triggerMsg, channel, `❓ ${e.question || 'The agent is asking for input.'}${opts}`); } catch {}
               }
               break;
             case 'notice':