interactive cloud runs

Author: joshsny

apps/twig/src/main/services/cloud-task/schemas.ts

@@ -44,6 +44,12 @@ export const onUpdateInput = z.object({
   runId: z.string(),
 });
 
+export const setViewingInput = z.object({
+  taskId: z.string(),
+  runId: z.string(),
+  viewing: z.boolean(),
+});
+
 export const sendCommandInput = z.object({
   taskId: z.string(),
   runId: z.string(),

apps/twig/src/main/services/cloud-task/service.ts

@@ -16,7 +16,9 @@ import {
 const log = logger.scope("cloud-task");
 
 const LOG_POLL_INTERVAL_MS = 5_000;
+const LOG_POLL_INTERVAL_FAST_MS = 1_000;
 const STATUS_POLL_INTERVAL_MS = 10_000;
+const STATUS_POLL_INTERVAL_FAST_MS = 3_000;
 
 interface TaskRunResponse {
   id: string;
@@ -43,6 +45,7 @@ interface WatcherState {
   lastBranch: string | null;
   lastStatusPollTime: number;
   subscriberCount: number;
+  viewing: boolean;
 }
 
 interface PendingWatchState {
@@ -130,6 +133,23 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
     }
   }
 
+  setViewing(taskId: string, runId: string, viewing: boolean): void {
+    const key = watcherKey(taskId, runId);
+    const watcher = this.watchers.get(key);
+    if (!watcher) return;
+
+    if (watcher.viewing === viewing) return;
+    watcher.viewing = viewing;
+
+    if (viewing && watcher.pollTimeoutId) {
+      clearTimeout(watcher.pollTimeoutId);
+      watcher.pollTimeoutId = null;
+      this.schedulePoll(key);
+    }
+
+    log.info("Cloud task watcher viewing changed", { key, viewing });
+  }
+
   async sendCommand(input: SendCommandInput): Promise<SendCommandOutput> {
     if (!this.apiKey) {
       return { success: false, error: "No API token available" };
@@ -242,6 +262,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       lastBranch: null,
       lastStatusPollTime: 0,
       subscriberCount,
+      viewing: false,
     };
 
     this.watchers.set(key, watcher);
@@ -268,10 +289,14 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
     const watcher = this.watchers.get(key);
     if (!watcher) return;
 
+    const interval = watcher.viewing
+      ? LOG_POLL_INTERVAL_FAST_MS
+      : LOG_POLL_INTERVAL_MS;
+
     watcher.pollTimeoutId = setTimeout(() => {
       watcher.pollTimeoutId = null;
       this.poll(key, false);
-    }, LOG_POLL_INTERVAL_MS);
+    }, interval);
   }
 
   private async poll(key: string, isSnapshot: boolean): Promise<void> {
@@ -284,9 +309,11 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
 
       // Fetch status if snapshot or interval elapsed
       const now = Date.now();
+      const statusInterval = watcher.viewing
+        ? STATUS_POLL_INTERVAL_FAST_MS
+        : STATUS_POLL_INTERVAL_MS;
       const shouldFetchStatus =
-        isSnapshot ||
-        now - watcher.lastStatusPollTime >= STATUS_POLL_INTERVAL_MS;
+        isSnapshot || now - watcher.lastStatusPollTime >= statusInterval;
 
       let statusResult: TaskRunResponse | null = null;
       let statusChanged = false;

apps/twig/src/main/trpc/routers/cloud-task.ts

@@ -5,6 +5,7 @@ import {
   onUpdateInput,
   sendCommandInput,
   sendCommandOutput,
+  setViewingInput,
   unwatchInput,
   updateTokenInput,
   watchInput,
@@ -28,6 +29,12 @@ export const cloudTaskRouter = router({
     .input(updateTokenInput)
     .mutation(({ input }) => getService().updateToken(input.token)),
 
+  setViewing: publicProcedure
+    .input(setViewingInput)
+    .mutation(({ input }) =>
+      getService().setViewing(input.taskId, input.runId, input.viewing),
+    ),
+
   sendCommand: publicProcedure
     .input(sendCommandInput)
     .output(sendCommandOutput)

apps/twig/src/renderer/features/sessions/components/buildConversationItems.ts

@@ -123,14 +123,7 @@ export function buildConversationItems(
   if (!isPromptPending) {
     for (const turn of b.pendingPrompts.values()) {
       turn.isComplete = true;
-      turn.stopReason = "cancelled";
-      turn.context.turnCancelled = true;
       turn.context.turnComplete = true;
-      b.items.push({
-        type: "turn_cancelled",
-        id: `${turn.id}-cancelled`,
-        interruptReason: turn.interruptReason,
-      });
     }
   }
 

apps/twig/src/renderer/features/sessions/service/service.ts

@@ -761,34 +761,49 @@ export class SessionService {
     this.previewAbort = null;
   }
 
+  private updatePromptStateFromEvents(
+    taskRunId: string,
+    events: AcpMessage[],
+  ): void {
+    for (const acpMsg of events) {
+      const msg = acpMsg.message;
+      if (isJsonRpcRequest(msg) && msg.method === "session/prompt") {
+        sessionStoreSetters.updateSession(taskRunId, {
+          isPromptPending: true,
+          promptStartedAt: acpMsg.ts,
+        });
+      }
+      if (
+        "id" in msg &&
+        "result" in msg &&
+        typeof msg.result === "object" &&
+        msg.result !== null &&
+        "stopReason" in msg.result
+      ) {
+        sessionStoreSetters.updateSession(taskRunId, {
+          isPromptPending: false,
+          promptStartedAt: null,
+        });
+      }
+    }
+  }
+
   private handleSessionEvent(taskRunId: string, acpMsg: AcpMessage): void {
     const session = sessionStoreSetters.getSessions()[taskRunId];
     if (!session) return;
 
     sessionStoreSetters.appendEvents(taskRunId, [acpMsg]);
+    this.updatePromptStateFromEvents(taskRunId, [acpMsg]);
 
     const msg = acpMsg.message;
 
-    if (isJsonRpcRequest(msg) && msg.method === "session/prompt") {
-      // acpMsg.ts is local time (set in-process) — matches Date.now() used in sendLocalPrompt
-      sessionStoreSetters.updateSession(taskRunId, {
-        isPromptPending: true,
-        promptStartedAt: acpMsg.ts,
-      });
-    }
-
     if (
       "id" in msg &&
       "result" in msg &&
       typeof msg.result === "object" &&
       msg.result !== null &&
       "stopReason" in msg.result
     ) {
-      sessionStoreSetters.updateSession(taskRunId, {
-        isPromptPending: false,
-        promptStartedAt: null,
-      });
-
       const stopReason = (msg.result as { stopReason?: string }).stopReason;
       if (stopReason) {
         notifyPromptComplete(session.taskTitle, stopReason);
@@ -1068,6 +1083,11 @@ export class SessionService {
     const session = sessionStoreSetters.getSessionByTaskId(taskId);
     if (!session) return false;
 
+    sessionStoreSetters.updateSession(session.taskRunId, {
+      isPromptPending: false,
+      promptStartedAt: null,
+    });
+
     if (session.isCloud) {
       return this.cancelCloudPrompt(session);
     }
@@ -1704,6 +1724,7 @@ export class SessionService {
         const entriesToAppend = update.newEntries.slice(-delta);
         const newEvents = convertStoredEntriesToEvents(entriesToAppend);
         sessionStoreSetters.appendEvents(taskRunId, newEvents, expectedCount);
+        this.updatePromptStateFromEvents(taskRunId, newEvents);
       } else {
         // Gap in data — append everything we have but don't jump processedLineCount
         log.warn("Cloud task log count inconsistency", {
@@ -1718,6 +1739,7 @@ export class SessionService {
           newEvents,
           currentCount + update.newEntries.length,
         );
+        this.updatePromptStateFromEvents(taskRunId, newEvents);
       }
     }
     // Update cloud status fields if present

apps/twig/src/renderer/features/task-detail/components/TaskLogsPanel.tsx

@@ -106,13 +106,23 @@ export function TaskLogsPanel({ taskId, task }: TaskLogsPanelProps) {
   // Cloud task watching — logs + status via main-process CloudTaskService subscription
   useEffect(() => {
     if (!isCloud || !task.latest_run?.id) return;
-    return getSessionService().watchCloudTask(
+    const runId = task.latest_run.id;
+    trpcVanilla.cloudTask.setViewing
+      .mutate({ taskId: task.id, runId, viewing: true })
+      .catch(() => {});
+    const cleanup = getSessionService().watchCloudTask(
       task.id,
-      task.latest_run.id,
+      runId,
       () => {
         queryClient.invalidateQueries({ queryKey: ["tasks"] });
       },
     );
+    return () => {
+      trpcVanilla.cloudTask.setViewing
+        .mutate({ taskId: task.id, runId, viewing: false })
+        .catch(() => {});
+      cleanup?.();
+    };
   }, [isCloud, task.id, task.latest_run?.id, queryClient]);
 
   // Local session connection