Make managed workspaces survive pauses and report richer activity

.github/workflows/devpc-managed-release.yml

@@ -128,7 +128,7 @@ jobs:
               activity="$(curl --connect-timeout 2 --max-time 5 --fail --silent --show-error \
                 -H 'x-devpc-gateway-token: release-smoke-managed-gateway-token' \
                 http://127.0.0.1:4311/api/_devpc/activity)"
-              test "$activity" = '{"active":false}' || {
+              test "$activity" = '{"active":false,"working":false,"pendingWork":false}' || {
                 echo "Unexpected managed activity response: $activity"
                 exit 1
               }

apps/server/src/managedDevPcActivity.test.ts

@@ -1,7 +1,14 @@
 import type { OrchestrationThreadShell } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 import { describe, expect, it } from "vite-plus/test";
 
-import { hasRunningManagedTurn } from "./managedDevPcActivity.ts";
+import {
+  hasPendingManagedWork,
+  hasQueuedManagedTurnStart,
+  hasRunningManagedTurn,
+  hasStartingManagedSession,
+  QUEUED_TURN_START_GRACE_MS,
+} from "./managedDevPcActivity.ts";
 import {
   clearPrimeAgentSessionActivity,
   hasRunningPrimeAgentSubagents,
@@ -14,11 +21,34 @@ const thread = (
 ) =>
   ({
     latestTurn: state === null ? null : { state },
+    latestUserMessageAt: null,
+    session: null,
     hasPendingApprovals: waiting === "approval",
     hasPendingUserInput: waiting === "input",
     hasActionableProposedPlan: waiting === "plan",
   }) as OrchestrationThreadShell;
 
+const NOW_MS = Date.parse("2026-08-09T12:00:00.000Z");
+const iso = (offsetMs: number) => DateTime.formatIso(DateTime.makeUnsafe(NOW_MS + offsetMs));
+
+const queuedThread = (
+  overrides: Partial<{
+    latestUserMessageAt: string | null;
+    latestTurn: OrchestrationThreadShell["latestTurn"];
+    sessionStatus: "idle" | "starting" | "running" | "ready" | "interrupted" | "stopped" | "error";
+  }> = {},
+) =>
+  ({
+    latestTurn: overrides.latestTurn ?? null,
+    latestUserMessageAt:
+      overrides.latestUserMessageAt === undefined ? iso(-5_000) : overrides.latestUserMessageAt,
+    session:
+      overrides.sessionStatus === undefined ? null : ({ status: overrides.sessionStatus } as never),
+    hasPendingApprovals: false,
+    hasPendingUserInput: false,
+    hasActionableProposedPlan: false,
+  }) as OrchestrationThreadShell;
+
 describe("managed DevPC activity", () => {
   it("reports activity only while an AI turn is running", () => {
     expect(hasRunningManagedTurn([thread("completed"), thread("running")])).toBe(true);
@@ -30,6 +60,68 @@ describe("managed DevPC activity", () => {
     expect(hasRunningManagedTurn([])).toBe(false);
   });
 
+  it("counts a fresh user message no turn adopted yet as queued work", () => {
+    expect(hasQueuedManagedTurnStart([queuedThread()], NOW_MS)).toBe(true);
+    // Adopted: the turn's requestedAt is newer than the message.
+    expect(
+      hasQueuedManagedTurnStart(
+        [
+          queuedThread({
+            latestTurn: {
+              state: "running",
+              requestedAt: iso(-1_000),
+              startedAt: null,
+              completedAt: null,
+            } as never,
+          }),
+        ],
+        NOW_MS,
+      ),
+    ).toBe(false);
+    // Stale: outside the grace window in either direction.
+    expect(
+      hasQueuedManagedTurnStart(
+        [queuedThread({ latestUserMessageAt: iso(-QUEUED_TURN_START_GRACE_MS - 1_000) })],
+        NOW_MS,
+      ),
+    ).toBe(false);
+    expect(
+      hasQueuedManagedTurnStart(
+        [queuedThread({ latestUserMessageAt: iso(QUEUED_TURN_START_GRACE_MS + 1_000) })],
+        NOW_MS,
+      ),
+    ).toBe(false);
+    // An errored session cannot adopt the message.
+    expect(hasQueuedManagedTurnStart([queuedThread({ sessionStatus: "error" })], NOW_MS)).toBe(
+      false,
+    );
+    expect(hasQueuedManagedTurnStart([queuedThread({ latestUserMessageAt: null })], NOW_MS)).toBe(
+      false,
+    );
+  });
+
+  it("counts a booting provider session as work in progress", () => {
+    expect(
+      hasStartingManagedSession([
+        queuedThread({ latestUserMessageAt: null, sessionStatus: "starting" }),
+      ]),
+    ).toBe(true);
+    expect(
+      hasStartingManagedSession([
+        queuedThread({ latestUserMessageAt: null, sessionStatus: "ready" }),
+      ]),
+    ).toBe(false);
+    expect(hasStartingManagedSession([queuedThread({ latestUserMessageAt: null })])).toBe(false);
+  });
+
+  it("reports human-blocked work as pending, separately from running work", () => {
+    expect(hasPendingManagedWork([thread("running", "approval")])).toBe(true);
+    expect(hasPendingManagedWork([thread("completed", "input")])).toBe(true);
+    expect(hasPendingManagedWork([thread("interrupted", "plan")])).toBe(true);
+    expect(hasPendingManagedWork([thread("running")])).toBe(false);
+    expect(hasPendingManagedWork([])).toBe(false);
+  });
+
   it("keeps a workspace active for detached Prime subagents, not a resident process", () => {
     const sessionKey = "primeAgent:thread-managed";
     expect(hasRunningPrimeAgentSubagents()).toBe(false);

apps/server/src/managedDevPcActivity.ts

@@ -1,6 +1,7 @@
 import * as NodeCrypto from "node:crypto";
 
 import type { OrchestrationThreadShell } from "@t3tools/contracts";
+import * as Clock from "effect/Clock";
 import * as Effect from "effect/Effect";
 import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http";
 
@@ -31,6 +32,60 @@ export function hasRunningManagedTurn(threads: ReadonlyArray<OrchestrationThread
   );
 }
 
+/**
+ * Mirrors the decider's queued-turn-start grace: a user message no turn has
+ * adopted yet is imminent agent work, and the workspace must not be paused in
+ * the window between "message sent" and "turn running". Bounded on both sides
+ * because message timestamps are client-supplied (see decider.ts).
+ */
+export const QUEUED_TURN_START_GRACE_MS = 2 * 60_000;
+
+export function hasQueuedManagedTurnStart(
+  threads: ReadonlyArray<OrchestrationThreadShell>,
+  nowMs: number,
+): boolean {
+  return threads.some((thread) => {
+    if (thread.session?.status === "error") return false;
+    if (thread.latestUserMessageAt === null) return false;
+    const messageAtMs = Date.parse(thread.latestUserMessageAt);
+    if (!Number.isFinite(messageAtMs)) return false;
+    const latestTurnAtMs =
+      thread.latestTurn === null
+        ? Number.NEGATIVE_INFINITY
+        : Math.max(
+            ...[
+              thread.latestTurn.requestedAt,
+              thread.latestTurn.startedAt,
+              thread.latestTurn.completedAt,
+            ].map((candidate) =>
+              candidate == null ? Number.NEGATIVE_INFINITY : Date.parse(candidate),
+            ),
+          );
+    return (
+      messageAtMs > latestTurnAtMs && Math.abs(nowMs - messageAtMs) <= QUEUED_TURN_START_GRACE_MS
+    );
+  });
+}
+
+/** A provider session mid-boot is about to run a turn; that is agent work too. */
+export function hasStartingManagedSession(
+  threads: ReadonlyArray<OrchestrationThreadShell>,
+): boolean {
+  return threads.some((thread) => thread.session?.status === "starting");
+}
+
+/**
+ * Work that exists but is blocked on the human: pause-safe (the workspace may
+ * idle out while an approval waits), yet worth surfacing so the platform can
+ * distinguish "nothing to do" from "waiting on the user".
+ */
+export function hasPendingManagedWork(threads: ReadonlyArray<OrchestrationThreadShell>): boolean {
+  return threads.some(
+    (thread) =>
+      thread.hasPendingApprovals || thread.hasPendingUserInput || thread.hasActionableProposedPlan,
+  );
+}
+
 const handleManagedDevPcActivity = Effect.gen(function* () {
   const request = yield* HttpServerRequest.HttpServerRequest;
   const config = yield* ServerConfig.ServerConfig;
@@ -45,20 +100,34 @@ const handleManagedDevPcActivity = Effect.gen(function* () {
   }
 
   const snapshots = yield* ProjectionSnapshotQuery;
+  const nowMs = yield* Clock.currentTimeMillis;
   return yield* snapshots.getShellSnapshot().pipe(
     Effect.match({
       onFailure: () =>
         HttpServerResponse.jsonUnsafe(
           { error: { code: "ACTIVITY_UNAVAILABLE", message: "Activity is unavailable." } },
           { status: 503, headers: { "cache-control": "no-store" } },
         ),
-      onSuccess: (snapshot) =>
-        HttpServerResponse.jsonUnsafe(
+      onSuccess: (snapshot) => {
+        // `active` keeps its original meaning (a genuinely running turn) so
+        // control planes reading only that field see unchanged behavior.
+        // `working` widens it with imminent work — a queued turn start or a
+        // booting session — which must hold a work claim before the turn's
+        // running state lands. `pendingWork` is human-blocked work: pause-safe
+        // but not "idle".
+        const active = hasRunningManagedTurn(snapshot.threads) || hasRunningPrimeAgentSubagents();
+        return HttpServerResponse.jsonUnsafe(
           {
-            active: hasRunningManagedTurn(snapshot.threads) || hasRunningPrimeAgentSubagents(),
+            active,
+            working:
+              active ||
+              hasQueuedManagedTurnStart(snapshot.threads, nowMs) ||
+              hasStartingManagedSession(snapshot.threads),
+            pendingWork: hasPendingManagedWork(snapshot.threads),
           },
           { status: 200, headers: { "cache-control": "no-store" } },
-        ),
+        );
+      },
     }),
   );
 });