fix(decopilot): publish user-prompt mirror after run purge so mid-run joiners see it (#4464)

Move the data-user-message publish from POST time (routes.ts) into
dispatchRun, right after streamBuffer.purge() clears the previous run.
Published before, the chunk was wiped by that purge, so a viewer JOINING
mid-run (deliverPolicy: "all") replayed only the assistant reply, not the
prompt, until a DB refetch. Publishing after the purge lets it survive the
whole run. Also aligns queued-turn UX (prompt appears at dispatch time).

Author: pedrofrxncx

apps/mesh/src/api/routes/decopilot/dispatch-run.ts

@@ -97,6 +97,7 @@ import {
   publishRunStatusStage,
   shouldPublishClusterRunStatus,
 } from "./run-status-stage";
+import { publishUserMessage } from "./user-message-stream";
 import type {
   HarnessStreamConsumerHooks,
   HarnessStreamTitleOptions,
@@ -1277,6 +1278,20 @@ async function prepareRun(
       }
     }
 
+    // Mirror the user prompt onto the run stream AFTER the purge above (which
+    // clears the previous run) so it survives the whole run — a viewer who
+    // JOINS mid-run replays it via `deliverPolicy: "all"`, not only viewers
+    // already connected at POST time. Best-effort and published before the
+    // run's assistant chunks so it sorts first. Uses the persisted message
+    // shape so the live chunk and a later DB refetch reconcile by id, no swap.
+    if (materializedRequestMessage) {
+      await publishUserMessage(
+        streamBuffer,
+        mem.thread.id,
+        materializedRequestMessage as UIMessage,
+      );
+    }
+
     const pendingOps: Promise<void>[] = [];
 
     // CLI-harness delta + resume only holds on the long-lived desktop daemon:

apps/mesh/src/api/routes/decopilot/routes.ts

@@ -45,7 +45,6 @@ import {
   publishRunStatusStage,
   shouldPublishClusterRunStatus,
 } from "./run-status-stage";
-import { publishUserMessage } from "./user-message-stream";
 import { wrapWithSseKeepalive } from "./sse-keepalive";
 import { resolveDispatchTarget } from "../../../links/resolve-dispatch-target";
 import {
@@ -747,10 +746,6 @@ export function createDecopilotRoutes(deps: DecopilotDeps) {
         })
       ) {
         await emitter.emitRequestMessage(persistedRequestMessage);
-        // Mirror the prompt onto the run stream so OTHER viewers of a shared
-        // thread render it live, not only after a DB refetch. Best-effort and
-        // published before the run's assistant chunks so it sorts first.
-        await publishUserMessage(streamBuffer, taskId, persistedRequestMessage);
       }
 
       const serializableRequest = buildDurableDispatchInput(input, {