feat(server): Prism stream clock for worker liveness, plus prism_cancel

apps/mobile/src/lib/threadActivity.ts

@@ -5,7 +5,11 @@ import {
   requestKindFromRequestType,
   type PendingApproval,
 } from "@t3tools/client-runtime/pending-requests";
-import { UserInputAttachmentAnswerPayload, isToolLifecycleItemType } from "@t3tools/contracts";
+import {
+  PRISM_STREAM_STATS_ACTIVITY_KIND,
+  UserInputAttachmentAnswerPayload,
+  isToolLifecycleItemType,
+} from "@t3tools/contracts";
 import type {
   OrchestrationLatestTurn,
   OrchestrationThread,
@@ -441,6 +445,7 @@ function deriveWorkLogEntries(
     if (activity.kind === "task.updated" && !isTerminalTaskUpdate(activity)) continue;
     if (activity.kind === "tool.progress") continue;
     if (activity.kind === "context-window.updated") continue;
+    if (activity.kind === PRISM_STREAM_STATS_ACTIVITY_KIND) continue;
     if (activity.summary === "Checkpoint captured") continue;
     if (isNoContentRuntimeWarning(activity)) continue;
     if (isPlanBoundaryToolActivity(activity)) continue;

apps/server/src/mcp/McpHttpServer.ts

@@ -35,6 +35,7 @@ import { PullRequestsToolkit } from "./toolkits/pullRequests/tools.ts";
 import { IssuesToolkitHandlersLive } from "./toolkits/issues/handlers.ts";
 import { IssuesToolkit } from "./toolkits/issues/tools.ts";
 import * as ProcessRunner from "../processRunner.ts";
+import { prismLivenessRouteLayer, streamStatsRecorderLayer } from "../prism/livenessRoute.ts";
 import { prismSnapshotRouteLayer } from "../prism/snapshotRoute.ts";
 import { PrismToolkitHandlersLive } from "./toolkits/prism/handlers.ts";
 import { PrismToolkit } from "./toolkits/prism/tools.ts";
@@ -697,4 +698,7 @@ export const layer = Layer.mergeAll(
   PrismToolkitRegistrationLive,
   DeviceToolkitRegistrationLive,
   prismSnapshotRouteLayer,
+  // Fork: Prism stream clock (toolboxmd/t3code#55).
+  prismLivenessRouteLayer,
+  streamStatsRecorderLayer,
 ).pipe(Layer.provideMerge(McpTransportLive));

apps/server/src/mcp/toolkits/prism/handlers.test.ts

@@ -3,7 +3,13 @@ import { assert, it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 
-import { findRouterBin, parseRouterOutput, routerStateDir, submitArgs } from "./handlers.ts";
+import {
+  cancelOwnedJob,
+  findRouterBin,
+  parseRouterOutput,
+  routerStateDir,
+  submitArgs,
+} from "./handlers.ts";
 
 it("submits with the calling thread as the T3 planner and starts the job", () => {
   const args = submitArgs({
@@ -74,3 +80,59 @@ it.layer(NodeServices.layer)("router discovery", (it) => {
     }).pipe(Effect.scoped),
   );
 });
+
+/** A router that owns job `prism-1` for `thread-1` and records every call. */
+function fakeRouter(owner: string) {
+  const calls: Array<{ args: ReadonlyArray<string>; env?: NodeJS.ProcessEnv }> = [];
+  let status = "running";
+  const runRouter = (args: ReadonlyArray<string>, env?: NodeJS.ProcessEnv) =>
+    Effect.sync(() => {
+      calls.push(env ? { args, env } : { args });
+      if (args[0] === "cancel") status = "cancelling";
+      if (args[0] === "recover") status = "cancelled";
+      if (args[0] === "status") return { job: { planner_t3_thread: owner, status } };
+      return { acknowledged: true, status };
+    });
+  return { calls, runRouter };
+}
+
+const tokenEnv = { T3_SERVER_URL: "http://127.0.0.1:3999", T3_SERVER_TOKEN: "secret" };
+
+it.effect("cancels then recovers a job the calling thread owns, with Prism's token", () =>
+  Effect.gen(function* () {
+    const router = fakeRouter("thread-1");
+    const result = yield* cancelOwnedJob({
+      requestId: "prism-1",
+      callerThreadId: "thread-1",
+      runRouter: router.runRouter,
+      issueToken: Effect.succeed(tokenEnv),
+    });
+    assert.deepStrictEqual(
+      router.calls.map((call) => [call.args[0], call.env?.T3_SERVER_TOKEN]),
+      [
+        ["status", undefined],
+        ["cancel", "secret"],
+        ["recover", "secret"],
+        ["status", undefined],
+      ],
+    );
+    assert.strictEqual(result.router.status, "cancelled");
+  }),
+);
+
+it.effect("refuses to cancel another thread's job without touching it", () =>
+  Effect.gen(function* () {
+    const router = fakeRouter("thread-2");
+    const error = yield* cancelOwnedJob({
+      requestId: "prism-1",
+      callerThreadId: "thread-1",
+      runRouter: router.runRouter,
+      issueToken: Effect.succeed(tokenEnv),
+    }).pipe(Effect.flip);
+    assert.strictEqual(error.reason, "Prism job prism-1 was not started by this thread.");
+    assert.deepStrictEqual(
+      router.calls.map((call) => call.args[0]),
+      ["status"],
+    );
+  }),
+);

apps/server/src/mcp/toolkits/prism/handlers.ts

@@ -30,6 +30,8 @@ const ROUTER_PLUGIN_DIR = [".codex", "plugins", "cache", "toolboxmd", "model-rou
 const ROUTER_TIMEOUT = "120 seconds";
 /** Jobs run for hours; the token must outlive the job, not the tool call. */
 const ROUTER_TOKEN_TTL = Duration.days(30);
+/** Cancel and recover finish within the call; the token only needs to outlive them. */
+const ROUTER_CANCEL_TOKEN_TTL = Duration.hours(1);
 
 const fail = (reason: string) => Effect.fail(new PrismToolError({ reason }));
 
@@ -136,6 +138,60 @@ export function parseRouterOutput(
   return { ok: parsed };
 }
 
+/** The T3 planner thread a router `status` reply names as the job's owner. */
+function plannerThreadOf(status: unknown): string | null {
+  const job =
+    status !== null && typeof status === "object" && "job" in status
+      ? (status as { job: unknown }).job
+      : null;
+  const thread =
+    job !== null && typeof job === "object" && "planner_t3_thread" in job
+      ? (job as { planner_t3_thread: unknown }).planner_t3_thread
+      : null;
+  return typeof thread === "string" && thread.length > 0 ? thread : null;
+}
+
+/** The job's status from a router `status` reply. */
+function jobStatusOf(status: unknown): string | null {
+  const job =
+    status !== null && typeof status === "object" && "job" in status
+      ? (status as { job: unknown }).job
+      : null;
+  const value =
+    job !== null && typeof job === "object" && "status" in job
+      ? (job as { status: unknown }).status
+      : null;
+  return typeof value === "string" ? value : null;
+}
+
+/**
+ * Router `cancel` then `recover` for a job the caller started, then its final
+ * status. `cancel` persists the intent; `recover` stops the job's processes
+ * and T3 child threads and finalizes it, or leaves it `cancelling` when
+ * something is still alive.
+ */
+export const cancelOwnedJob = <E, R>(input: {
+  readonly requestId: string;
+  readonly callerThreadId: string;
+  readonly runRouter: (
+    args: ReadonlyArray<string>,
+    env?: NodeJS.ProcessEnv,
+  ) => Effect.Effect<unknown, PrismToolError | E, R>;
+  readonly issueToken: Effect.Effect<NodeJS.ProcessEnv, PrismToolError | E, R>;
+}) =>
+  Effect.gen(function* () {
+    const { requestId, runRouter } = input;
+    const owner = plannerThreadOf(yield* runRouter(["status", "--request-id", requestId]));
+    if (owner !== input.callerThreadId) {
+      return yield* fail(`Prism job ${requestId} was not started by this thread.`);
+    }
+    const env = yield* input.issueToken;
+    const cancel = yield* runRouter(["cancel", "--request-id", requestId], env);
+    const recover = yield* runRouter(["recover", "--request-id", requestId], env);
+    const status = jobStatusOf(yield* runRouter(["status", "--request-id", requestId]));
+    return { requestId, router: { status, cancel, recover } };
+  });
+
 /** A wildcard bind is reachable on loopback, where the router runs. */
 function serverBaseUrl(address: HttpServer.HttpServer["Service"]["address"]): string {
   if (!NetAddress.isInetAddress(address)) return "http://127.0.0.1:3773";
@@ -260,6 +316,28 @@ const make = Effect.gen(function* () {
         ),
         Effect.map((router) => ({ requestId, router })),
       ),
+    prism_cancel: ({ requestId }) =>
+      Effect.gen(function* () {
+        const caller = yield* plannerCaller;
+        // Cancelling interrupts the job's T3 child threads, which needs a
+        // server token like submit's.
+        const issueToken = auth
+          .issueSession({
+            scopes: [AuthOrchestrationReadScope, AuthOrchestrationOperateScope],
+            label: `Prism router cancel ${requestId}`,
+            ttl: ROUTER_CANCEL_TOKEN_TTL,
+          })
+          .pipe(
+            Effect.map((issued) => ({ T3_SERVER_URL: serverUrl, T3_SERVER_TOKEN: issued.token })),
+            Effect.catchCause(() => fail("Could not issue a server token for Prism.")),
+          );
+        return yield* cancelOwnedJob({
+          requestId,
+          callerThreadId: caller.id,
+          runRouter,
+          issueToken,
+        });
+      }),
   });
 });
 

apps/server/src/mcp/toolkits/prism/tools.ts

@@ -113,9 +113,24 @@ const PrismAnswerTool = Tool.make("prism_answer", {
   .annotate(Tool.Idempotent, false)
   .annotate(Tool.OpenWorld, true);
 
+const PrismCancelTool = Tool.make("prism_cancel", {
+  description:
+    "Cancel a Prism job this thread started: Prism records the cancellation, stops the job's workers and child threads, and returns the final state (cancelled, or still cancelling when something could not be stopped yet; call again to retry). Use it only when the user asks to stop the job or the job is clearly stuck.",
+  parameters: Schema.Struct({ requestId: RequestId }),
+  success: PrismResult,
+  failure: PrismToolError,
+  dependencies,
+})
+  .annotate(Tool.Title, "Cancel Prism job")
+  .annotate(Tool.Readonly, false)
+  .annotate(Tool.Destructive, true)
+  .annotate(Tool.Idempotent, true)
+  .annotate(Tool.OpenWorld, true);
+
 export const PrismToolkit = Toolkit.make(
   PrismSubmitTool,
   PrismStatusTool,
   PrismQuestionsTool,
   PrismAnswerTool,
+  PrismCancelTool,
 );

apps/server/src/prism/livenessRoute.ts

@@ -0,0 +1,129 @@
+import {
+  AuthOrchestrationReadScope,
+  CommandId,
+  EventId,
+  PRISM_STREAM_STATS_ACTIVITY_KIND,
+  ThreadId,
+  TurnId,
+} from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
+import * as Clock from "effect/Clock";
+import * as Crypto from "effect/Crypto";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http";
+
+import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
+import * as OrchestrationEngine from "../orchestration/Services/OrchestrationEngine.ts";
+import { StreamClock, type ThreadLiveness, type TurnStreamStats } from "./streamClock.ts";
+
+const PRISM_LIVENESS_PATH = "/api/prism/liveness";
+
+const iso = (ms: number) => DateTime.formatIso(DateTime.makeUnsafe(ms));
+
+/**
+ * What `GET /api/prism/liveness?threadId=` returns: the server's `now` beside
+ * the thread's last stream event, so the caller measures silence without
+ * clock skew. `lastStreamAt` is null until the thread streams after startup.
+ */
+export function livenessView(threadId: string, now: number, liveness: ThreadLiveness) {
+  return {
+    threadId,
+    now: iso(now),
+    lastStreamAt: liveness.lastStreamAt === null ? null : iso(liveness.lastStreamAt),
+    silenceMs: liveness.lastStreamAt === null ? null : now - liveness.lastStreamAt,
+    openTool: liveness.openTool
+      ? { ...liveness.openTool, startedAt: iso(liveness.openTool.startedAt) }
+      : null,
+    turn: liveness.turn ? { ...liveness.turn, startedAt: iso(liveness.turn.startedAt) } : null,
+  };
+}
+
+const seconds = (ms: number) => `${(ms / 1000).toFixed(1)} s`;
+
+export function streamStatsActivity(stats: TurnStreamStats) {
+  const firstToken =
+    stats.timeToFirstTokenMs === null
+      ? "no tokens"
+      : `first token ${seconds(stats.timeToFirstTokenMs)}`;
+  return {
+    kind: PRISM_STREAM_STATS_ACTIVITY_KIND,
+    summary: `Stream ${stats.outcome}: ${firstToken}, max gap ${seconds(stats.maxGapMs)}, ${stats.eventCount} events`,
+    payload: {
+      provider: stats.provider,
+      model: stats.model,
+      outcome: stats.outcome,
+      state: stats.state,
+      startedAt: iso(stats.startedAt),
+      endedAt: iso(stats.endedAt),
+      durationMs: stats.endedAt - stats.startedAt,
+      timeToFirstTokenMs: stats.timeToFirstTokenMs,
+      maxGapMs: stats.maxGapMs,
+      eventCount: stats.eventCount,
+    },
+  };
+}
+
+const jsonError = (status: number, error: string) =>
+  HttpServerResponse.jsonUnsafe({ error }, { status });
+
+/** `GET /api/prism/liveness?threadId=…`, bearer token with `orchestration:read`. */
+export const prismLivenessRouteLayer = HttpRouter.add(
+  "GET",
+  PRISM_LIVENESS_PATH,
+  Effect.gen(function* () {
+    const request = yield* HttpServerRequest.HttpServerRequest;
+    const auth = yield* EnvironmentAuth.EnvironmentAuth;
+    const session = yield* auth.authenticateHttpRequest(request).pipe(Effect.option);
+    if (Option.isNone(session)) return jsonError(401, "unauthorized");
+    if (!session.value.scopes.includes(AuthOrchestrationReadScope)) {
+      return jsonError(403, `scope ${AuthOrchestrationReadScope} required`);
+    }
+    const url = HttpServerRequest.toURL(request);
+    const threadId = (Option.isSome(url) ? url.value.searchParams.get("threadId") : null)?.trim();
+    if (!threadId) return jsonError(400, "threadId required");
+    // Optional so route harnesses without provider wiring still build.
+    const clock = yield* Effect.serviceOption(StreamClock);
+    if (Option.isNone(clock)) return jsonError(503, "stream clock unavailable");
+    const liveness = yield* clock.value.liveness(ThreadId.make(threadId));
+    const now = yield* Clock.currentTimeMillis;
+    return HttpServerResponse.jsonUnsafe(livenessView(threadId, now, liveness));
+  }),
+);
+
+/** Records each finished turn's stream statistics as one thread activity. */
+export const streamStatsRecorderLayer = Layer.effectDiscard(
+  Effect.gen(function* () {
+    const clock = yield* Effect.serviceOption(StreamClock);
+    if (Option.isNone(clock)) return;
+    const engine = yield* OrchestrationEngine.OrchestrationEngineService;
+    const crypto = yield* Crypto.Crypto;
+    const record = (stats: TurnStreamStats) =>
+      Effect.gen(function* () {
+        const id = yield* crypto.randomUUIDv4.pipe(Effect.orDie);
+        const createdAt = iso(stats.endedAt);
+        yield* engine.dispatch({
+          type: "thread.activity.append",
+          commandId: CommandId.make(`server:prism-stream-stats:${id}`),
+          threadId: stats.threadId,
+          activity: {
+            id: EventId.make(id),
+            tone: "info",
+            ...streamStatsActivity(stats),
+            turnId: stats.turnId === null ? null : TurnId.make(stats.turnId),
+            createdAt,
+          },
+          createdAt,
+        });
+      }).pipe(
+        Effect.catchCause((cause) =>
+          Cause.hasInterruptsOnly(cause)
+            ? Effect.failCause(cause)
+            : Effect.logWarning("Could not record Prism stream statistics", { cause }),
+        ),
+      );
+    yield* Effect.forkScoped(Effect.forever(Effect.flatMap(clock.value.takeTurnEnd, record)));
+  }),
+);