@@ -328,6 +328,137 @@ describe("CloudTaskService", () => {
     );
   });
 
+  it("replays a current snapshot when a subscriber attaches to an existing watcher", async () => {
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+
+    const historicalEntry = {
+      type: "notification",
+      timestamp: "2026-01-01T00:00:00Z",
+      notification: {
+        jsonrpc: "2.0",
+        method: "_posthog/console",
+        params: {
+          sessionId: "run-1",
+          level: "info",
+          message: "older history",
+        },
+      },
+    };
+    const liveEntry = {
+      type: "notification",
+      timestamp: "2026-01-01T00:00:01Z",
+      notification: {
+        jsonrpc: "2.0",
+        method: "_posthog/console",
+        params: {
+          sessionId: "run-1",
+          level: "info",
+          message: "live tail",
+        },
+      },
+    };
+
+    const runResponse = {
+      id: "run-1",
+      status: "in_progress",
+      stage: "build",
+      output: null,
+      error_message: null,
+      branch: "main",
+      updated_at: "2026-01-01T00:00:00Z",
+    };
+
+    mockNetFetch
+      .mockResolvedValueOnce(createJsonResponse(runResponse))
+      .mockResolvedValueOnce(
+        createJsonResponse([historicalEntry], 200, { "X-Has-More": "false" }),
+      )
+      .mockResolvedValueOnce(createJsonResponse(runResponse))
+      .mockResolvedValueOnce(
+        createJsonResponse([historicalEntry], 200, { "X-Has-More": "false" }),
+      );
+
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(`id: 1\ndata: ${JSON.stringify(liveEntry)}\n\n`),
+    );
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() => updates.length >= 2);
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() =>
+      updates.some(
+        (update) =>
+          typeof update === "object" &&
+          update !== null &&
+          (update as { kind?: string; totalEntryCount?: number }).kind ===
+            "snapshot" &&
+          (update as { totalEntryCount?: number }).totalEntryCount === 2,
+      ),
+    );
+
+    const replayedSnapshot = updates.find(
+      (update) =>
+        typeof update === "object" &&
+        update !== null &&
+        (update as { kind?: string; totalEntryCount?: number }).kind ===
+          "snapshot" &&
+        (update as { totalEntryCount?: number }).totalEntryCount === 2,
+    );
+
+    expect(replayedSnapshot).toEqual({
+      taskId: "task-1",
+      runId: "run-1",
+      kind: "snapshot",
+      newEntries: [historicalEntry, liveEntry],
+      totalEntryCount: 2,
+      status: "in_progress",
+      stage: "build",
+      output: null,
+      errorMessage: null,
+      branch: "main",
+    });
+
+    const getWatcherEmittedEntryCount = (): number => {
+      const watcher = (
+        service as unknown as {
+          watchers: Map<string, { emittedLogEntries: unknown[] }>;
+        }
+      ).watchers.get("task-1:run-1");
+      return watcher?.emittedLogEntries.length ?? 0;
+    };
+
+    expect(getWatcherEmittedEntryCount()).toBe(1);
+
+    mockNetFetch.mockResolvedValueOnce(
+      createJsonResponse([historicalEntry, liveEntry], 200, {
+        "X-Has-More": "false",
+      }),
+    );
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() => getWatcherEmittedEntryCount() === 0);
+  });
+
   it("ignores keepalive SSE events while keeping the stream open", async () => {
     const updates: unknown[] = [];
     service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
 
@@ -92,6 +92,7 @@ interface WatcherState {
   isBootstrapping: boolean;
   hasEmittedSnapshot: boolean;
   bufferedLogBatches: StoredLogEntry[][];
+  emittedLogEntries: StoredLogEntry[];
   failed: boolean;
   needsPostBootstrapReconnect: boolean;
   needsStopAfterBootstrap: boolean;
@@ -224,6 +225,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
         key,
         subscribers: existing.subscriberCount,
       });
+      void this.emitCurrentSnapshot(key);
       return;
     }
 
@@ -399,6 +401,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       isBootstrapping: false,
       hasEmittedSnapshot: false,
       bufferedLogBatches: [],
+      emittedLogEntries: [],
       failed: false,
       needsPostBootstrapReconnect: false,
       needsStopAfterBootstrap: false,
@@ -767,6 +770,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
     }
 
     watcher.totalEntryCount += entries.length;
+    this.rememberEmittedLogEntries(watcher, entries);
 
     this.emit(CloudTaskEvent.Update, {
       taskId: watcher.taskId,
@@ -812,6 +816,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       }
 
       watcher.totalEntryCount += dedupedEntries.length;
+      this.rememberEmittedLogEntries(watcher, dedupedEntries);
       this.emit(CloudTaskEvent.Update, {
         taskId: watcher.taskId,
         runId: watcher.runId,
@@ -824,6 +829,92 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
     watcher.bufferedLogBatches = [];
   }
 
+  private rememberEmittedLogEntries(
+    watcher: WatcherState,
+    entries: StoredLogEntry[],
+  ): void {
+    watcher.emittedLogEntries.push(...entries);
+  }
+
+  private mergeHistoricalAndEmittedEntries(
+    historicalEntries: StoredLogEntry[],
+    emittedEntries: StoredLogEntry[],
+  ): {
+    snapshotEntries: StoredLogEntry[];
+    missingEmittedEntries: StoredLogEntry[];
+  } {
+    if (emittedEntries.length === 0) {
+      return { snapshotEntries: historicalEntries, missingEmittedEntries: [] };
+    }
+
+    const historicalCounts = new Map<string, number>();
+    for (const entry of historicalEntries) {
+      const serialized = JSON.stringify(entry);
+      historicalCounts.set(
+        serialized,
+        (historicalCounts.get(serialized) ?? 0) + 1,
+      );
+    }
+
+    const missingEmittedEntries = emittedEntries.filter((entry) => {
+      const serialized = JSON.stringify(entry);
+      const remaining = historicalCounts.get(serialized) ?? 0;
+      if (remaining <= 0) {
+        return true;
+      }
+
+      historicalCounts.set(serialized, remaining - 1);
+      return false;
+    });
+
+    return {
+      snapshotEntries: [...historicalEntries, ...missingEmittedEntries],
+      missingEmittedEntries,
+    };
+  }
+
+  private async emitCurrentSnapshot(key: string): Promise<void> {
+    const watcher = this.watchers.get(key);
+    if (!watcher || watcher.failed) return;
+
+    const historicalEntries = await this.fetchAllSessionLogs(watcher);
+    const currentWatcher = this.watchers.get(key);
+    if (!currentWatcher || currentWatcher !== watcher || watcher.failed) {
+      return;
+    }
+
+    if (!historicalEntries) {
+      log.warn("Cloud task snapshot replay failed", {
+        taskId: watcher.taskId,
+        runId: watcher.runId,
+      });
+      return;
+    }
+
+    const { snapshotEntries, missingEmittedEntries } =
+      this.mergeHistoricalAndEmittedEntries(
+        historicalEntries,
+        watcher.emittedLogEntries,
+      );
+    watcher.emittedLogEntries = missingEmittedEntries;
+    if (snapshotEntries.length > watcher.totalEntryCount) {
+      watcher.totalEntryCount = snapshotEntries.length;
+    }
+
+    this.emit(CloudTaskEvent.Update, {
+      taskId: watcher.taskId,
+      runId: watcher.runId,
+      kind: "snapshot",
+      newEntries: snapshotEntries,
+      totalEntryCount: snapshotEntries.length,
+      status: watcher.lastStatus ?? undefined,
+      stage: watcher.lastStage,
+      output: watcher.lastOutput,
+      errorMessage: watcher.lastErrorMessage,
+      branch: watcher.lastBranch,
+    });
+  }
+
   private failWatcher(key: string, error: CloudTaskConnectionError): void {
     const watcher = this.watchers.get(key);
     if (!watcher) return;
 
@@ -37,15 +37,19 @@ export const cloudTaskRouter = router({
     .input(onUpdateInput)
     .subscription(async function* (opts) {
       const service = getService();
-      for await (const data of service.toIterable(CloudTaskEvent.Update, {
-        signal: opts.signal,
-      })) {
-        if (
-          data.taskId === opts.input.taskId &&
-          data.runId === opts.input.runId
-        ) {
-          yield data;
+      try {
+        for await (const data of service.toIterable(CloudTaskEvent.Update, {
+          signal: opts.signal,
+        })) {
+          if (
+            data.taskId === opts.input.taskId &&
+            data.runId === opts.input.runId
+          ) {
+            yield data;
+          }
         }
+      } finally {
+        service.unwatch(opts.input.taskId, opts.input.runId);
       }
     }),
 });
@@ -62,6 +62,18 @@ function promptResponseMsg(ts: number, id: number): AcpMessage {
   };
 }
 
+function turnCompleteMsg(ts: number, stopReason = "end_turn"): AcpMessage {
+  return {
+    type: "acp_message",
+    ts,
+    message: {
+      jsonrpc: "2.0",
+      method: "_posthog/turn_complete",
+      params: { sessionId: "session-1", stopReason },
+    },
+  };
+}
+
 describe("buildConversationItems", () => {
   it("extracts cloud prompt attachments into user messages", () => {
     const uri = makeAttachmentUri("/tmp/hello world.txt");
@@ -109,6 +121,19 @@ describe("buildConversationItems", () => {
     ]);
   });
 
+  it("marks cloud turns complete from structured turn completion notifications", () => {
+    const result = buildConversationItems(
+      [userPromptMsg(10, 42, "hello"), turnCompleteMsg(25)],
+      null,
+    );
+
+    expect(result.lastTurnInfo).toEqual({
+      isComplete: true,
+      durationMs: 15,
+      stopReason: "end_turn",
+    });
+  });
+
   it("keeps attachment-only prompts visible", () => {
     const uri = makeAttachmentUri("/tmp/test.txt");