feat(cell): the cell provider against the login-era cell: refused tokens end the session's reading with what to do, allowance and run bounds reported, run usage as token usage

apps/server/scripts/live-cell-api.mjs

@@ -1,7 +1,7 @@
 // A scripted OpenAI Responses API for the live cell run: point a cell's OPENAI_BASE_URL at it.
 //   node scripts/live-cell-api.mjs [port=9815] [holdMs=20000]   (CHILD_DELAY_MS=30000 for delete-delegate)
 // The latest user message picks the answer; the prompt table is in docs/internals/provider-cell.md (Testing).
-// Every request is logged with its step (function calls since the prompt) and prompt.
+// Every request is logged with its step and prompt; every answer reports usage (bytes / 4).
 import * as NodeHttp from "node:http";
 
 const port = Number(process.argv[2] ?? 9815);
@@ -58,7 +58,14 @@ NodeHttp.createServer((req, res) => {
       .find((item) => item.type === "message" && item.role === "assistant");
     const previousText =
       previous?.content?.find?.((part) => part.type === "output_text")?.text ?? "";
-    const ok = (output) => json(200, { id: `resp_${n}`, status: "completed", output });
+    const usage = (output) => ({
+      input_tokens: Math.ceil(data.length / 4),
+      input_tokens_details: { cached_tokens: 0 },
+      output_tokens: Math.ceil(JSON.stringify(output).length / 4),
+      output_tokens_details: { reasoning_tokens: 0 },
+    });
+    const ok = (output) =>
+      json(200, { id: `resp_${n}`, status: "completed", output, usage: usage(output) });
     console.log(
       `${stamp()} [api] #${n} step ${step} model=${body.model} prompt=${JSON.stringify(prompt.split("\n")[0].slice(0, 100))}`,
     );

apps/server/scripts/live-cell-driver.ts

@@ -1,4 +1,4 @@
-// @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalDate:off globalDateInEffect:off globalConsole:off preferSchemaOverJson:off globalFetchInEffect:off - a live driver: it runs the real orchestrator and adapter against a cell on celld, logs on the wall clock, and bounds its own reads of the cell with an AbortSignal.
+// @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalDate:off globalDateInEffect:off globalConsole:off preferSchemaOverJson:off globalFetchInEffect:off globalFetch:off - a live driver: it runs the real orchestrator and adapter against a cell on celld, logs on the wall clock, bounds its own reads of the cell with an AbortSignal, and posts the operator's request and other projects' prompts itself.
 /**
  * Drives the real orchestrator and cell adapter against a live cell for the scenarios the recorder
  * cannot express, logging every HTTP exchange and socket frame with a wall-clock stamp.
@@ -101,6 +101,11 @@ const PROMPTS = {
   reconnect: "Hold: Respond with exactly: reconnect ok",
   restart: "Hold: Respond with exactly: RESUMED",
   attachRetry: "Respond with exactly: attach retry ok",
+  access: "Respond with exactly: access ok",
+  revoke:
+    "Do not answer immediately. First run the local shell command `sleep 35`, then respond with exactly: revoked run should not finish.",
+  afterRevoke: "Respond with exactly: after the revocation",
+  holder: "Hold: Respond with exactly: holder",
 } as const;
 
 /** The adapter runs on the test clock; the cell on the wall clock. */
@@ -122,9 +127,11 @@ const makeWebSocket: Socket.WebSocketConstructor["Service"] = (url, options) =>
   const ordinal = socketOrdinal;
   latestSocket = socket;
   socket.on("open", () => log(`ws#${ordinal} open ${url}`));
-  socket.on("close", (code, reason) =>
-    log(`ws#${ordinal} closed code=${code} reason=${String(reason)}`),
-  );
+  socket.on("close", (code, reason) => {
+    log(`ws#${ordinal} closed code=${code} reason=${String(reason)}`);
+    if (code === 4401 && socketsAfterClose === undefined) socketsAfterClose = 0;
+  });
+  if (socketsAfterClose !== undefined) socketsAfterClose += 1;
   socket.on("error", (error) => log(`ws#${ordinal} error ${error.message}`));
   return socket as unknown as globalThis.WebSocket;
 };
@@ -157,6 +164,7 @@ const loggingTransport = (live: CellTransport): CellTransport => ({
     log(
       `>> ${input.method} ${input.url} headers=${JSON.stringify(redact(input.headers))}${input.body === undefined ? "" : ` body=${JSON.stringify(input.body)}`}`,
     );
+    if (input.method === "POST" && input.url.endsWith("/prompt")) promptsPosted += 1;
     const isTaskCancel = input.method === "POST" && /\/tasks\/[^/]+\/cancel$/.test(input.url);
     if (isTaskCancel) onTaskCancelSent?.({ url: input.url, headers: input.headers });
     const response = yield* live.request(input).pipe(
@@ -227,6 +235,30 @@ const loggingTransport = (live: CellTransport): CellTransport => ({
   }),
 });
 
+/** The login-era scenarios check what the thread shows; each exits non-zero on a miss. */
+const errorsOf = (projection: OrchestrationV2ThreadProjection | undefined) =>
+  (projection?.turnItems ?? []).flatMap((item) => (item.type === "error" ? [item.failure] : []));
+const check = (ok: boolean, what: string) =>
+  ok
+    ? Effect.sync(() => log(`${scenarioName}: check passed: ${what}`))
+    : Effect.die(new Error(`${scenarioName}: check failed: ${what}`));
+/** A POST of the driver's own, beside the adapter: the operator's request, or another project's prompt. */
+const postNow = async (path: string, body: string, headers: Record<string, string>) => {
+  const response = await fetch(`${baseUrl}${path}`, { method: "POST", headers, body });
+  const text = await response.text();
+  log(`driver POST ${path} -> ${response.status} ${text.slice(0, 300)}`);
+  return response.status;
+};
+const post = (path: string, body: string, headers: Record<string, string>) =>
+  Effect.promise(() => postNow(path, body, headers));
+const operateNow = (token: string) =>
+  postNow("/login/operator", new URLSearchParams({ token }).toString(), {
+    "content-type": "application/x-www-form-urlencoded",
+  });
+const operate = (token: string) => Effect.promise(() => operateNow(token));
+let promptsPosted = 0;
+let socketsAfterClose: number | undefined;
+
 const describeProjection = (projection: OrchestrationV2ThreadProjection) =>
   [
     ...projection.runs.map(
@@ -327,6 +359,17 @@ const stepsFor = (name: string): ReadonlyArray<OrchestratorFixtureInputStep> =>
       return [{ type: "message", text: PROMPTS.attachRetry }];
     case "delete-delegate":
       return [{ type: "message", text: PROMPTS.delegate }];
+    case "access":
+    case "runs-bound":
+      return [{ type: "message", text: PROMPTS.access }];
+    // Two model requests, so an allowance that covers the first runs out at the second.
+    case "allowance":
+      return [{ type: "message", text: PROMPTS.shell }];
+    case "revoke":
+      return [
+        { type: "message", text: PROMPTS.revoke },
+        { type: "message", text: PROMPTS.afterRevoke },
+      ];
     default:
       throw new Error(`Unknown scenario ${name}`);
   }
@@ -378,6 +421,27 @@ const main = Effect.gen(function* () {
     };
   }
 
+  const authorization: Record<string, string> = settings.token
+    ? { authorization: `Bearer ${settings.token}` }
+    : {};
+  if (scenarioName === "revoke") {
+    const suspend = readArgValue("--suspend");
+    if (suspend === undefined)
+      return yield* Effect.die(new Error("revoke needs --suspend <token>"));
+    onPromptAccepted = () => {
+      onPromptAccepted = undefined;
+      log("revoke: prompt accepted; the operator suspends the tenant");
+      void operateNow(suspend);
+    };
+  }
+  const holder = `${runTag.replace(/[^A-Za-z0-9._-]/g, "-")}-holder`.slice(0, 60);
+  if (scenarioName === "runs-bound") {
+    yield* post(`/cell/${holder}/prompt`, PROMPTS.holder, {
+      ...authorization,
+      "x-request-id": `holder-${runTag}`,
+    });
+  }
+
   if (scenarioName === "delete-delegate") {
     const workspace = yield* checkpointWorkspace("live-delete-delegate");
     const threadId = materialized.projectionThreadIds[0]!;
@@ -536,6 +600,52 @@ const main = Effect.gen(function* () {
         );
       log(`attach-retry: check passed: ${socketOrdinal} sockets, the prompt completed`);
     }
+    const [projection] = result.projections.values();
+    const failures = errorsOf(projection);
+    const first = failures[0];
+    if (scenarioName === "access") {
+      yield* check(
+        projection?.runs[0]?.status === "failed" &&
+          /Settings → Providers|not serving requests|refused T3's request/.test(
+            first?.message ?? "",
+          ),
+        `the run failed with the tenant's refusal and what to do: ${first?.message}`,
+      );
+      yield* check(promptsPosted === 0, `no prompt was posted (${promptsPosted})`);
+    }
+    if (scenarioName === "revoke") {
+      const resume = readArgValue("--resume");
+      yield* check(
+        projection?.runs[0]?.status === "failed" &&
+          first?.code === "cell_unauthorized" &&
+          /suspended/.test(first.message),
+        `the run failed with the refusal naming the suspension: ${first?.message}`,
+      );
+      yield* check(
+        projection?.runs[1]?.status === "failed" && /Access token/.test(failures[1]?.message ?? ""),
+        `the next message failed at once with the refusal: ${failures[1]?.message}`,
+      );
+      yield* check(
+        socketsAfterClose === 0,
+        `no socket opened after the 4401: ${socketsAfterClose ?? "no 4401 seen"}`,
+      );
+      if (resume !== undefined) yield* operate(resume);
+    }
+    if (scenarioName === "allowance") {
+      // vim's cell names the usage limit and its reset time on a prompt's 409 since vim #53.
+      yield* check(
+        first?.class === "usage_limit" && typeof first.resetAt === "string" && first.resetAt !== "",
+        `the run failed as the usage limit with its reset time: ${JSON.stringify(first)}`,
+      );
+    }
+    if (scenarioName === "runs-bound") {
+      yield* post(`/cell/${holder}/cancel`, "", authorization);
+      yield* check(
+        projection?.runs[0]?.status === "failed" &&
+          /send the prompt again/.test(first?.message ?? ""),
+        `the run failed saying to send the prompt again: ${first?.message}`,
+      );
+    }
     return;
   }
 

apps/server/scripts/live-cell-probe.ts

@@ -1,9 +1,8 @@
 // @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalConsole:off globalConsoleInEffect:off preferSchemaOverJson:off - a live probe driver that prints what it sees.
 /**
- * Runs the cell provider's status probe (the one Settings → Providers shows)
- * against a live tenant and against stand-ins for an old-protocol tenant, a
- * protocol-2 tenant, a non-cell HTTP server and a closed port.
- *   node scripts/live-cell-probe.ts --url http://127.0.0.1:9884
+ * Runs the cell provider's status probe against a live tenant (with --token, and without one) and
+ * stand-ins for old protocols, a non-cell server, a closed port, a 401 and a failing-closed 503.
+ *   node scripts/live-cell-probe.ts --url http://127.0.0.1:9884 [--token t]
  */
 import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient";
 import * as NodeServices from "@effect/platform-node/NodeServices";
@@ -15,18 +14,28 @@ import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
 import * as NodeHttp from "node:http";
 
-const liveUrl = process.argv[process.argv.indexOf("--url") + 1] ?? "http://127.0.0.1:9884";
+const argValue = (name: string) => {
+  const index = process.argv.indexOf(name);
+  return index === -1 ? undefined : process.argv[index + 1];
+};
+const liveUrl = argValue("--url") ?? "http://127.0.0.1:9884";
+const liveToken = argValue("--token") ?? "";
 const decode = Schema.decodeUnknownEffect(CellSettings);
 
 function serve(
-  answer: (path: string) => { status: number; type: string; body: string },
+  answer: (path: string) => {
+    status: number;
+    type: string;
+    body: string;
+    headers?: Record<string, string>;
+  },
 ): Promise<{ url: string; close: () => void }> {
   return new Promise((resolve) => {
     const requests: Array<string> = [];
     const server = NodeHttp.createServer((req, res) => {
       requests.push(`${req.method} ${req.url}`);
-      const { status, type, body } = answer(req.url ?? "");
-      res.writeHead(status, { "content-type": type });
+      const { status, type, body, headers } = answer(req.url ?? "");
+      res.writeHead(status, { "content-type": type, ...headers });
       res.end(body);
     });
     server.listen(0, "127.0.0.1", () => {
@@ -60,17 +69,19 @@ const main = Effect.gen(function* () {
   const probe = Effect.fn("LiveCellProbe.probe")(function* (
     label: string,
     baseUrl: string,
+    token = "",
     projectPrefix = "t3-",
   ) {
-    const settings = yield* decode({ enabled: true, baseUrl, token: "", projectPrefix });
+    const settings = yield* decode({ enabled: true, baseUrl, token, projectPrefix });
     const result = yield* checkCellProviderStatus(settings, transport);
     console.log(`${label}: ${baseUrl}`);
     console.log(
       `  status=${result.status} installed=${result.installed} version=${JSON.stringify(result.version)} models=${JSON.stringify(result.models.map((m) => m.slug))}`,
     );
-    console.log(`  message=${JSON.stringify(result.message)}`);
+    console.log(`  auth=${result.auth.status} message=${JSON.stringify(result.message)}`);
   });
-  yield* probe("live cell", liveUrl);
+  yield* probe(liveToken ? "live cell with the token" : "live cell", liveUrl, liveToken);
+  if (liveToken) yield* probe("live cell without a token", liveUrl);
   const legacy = yield* Effect.promise(() =>
     serve(() => ({ status: 200, type: "application/json", body: oldState({}) })),
   );
@@ -100,6 +111,26 @@ const main = Effect.gen(function* () {
   );
   yield* probe("protocol-3 tenant without an API key", unconfigured.url);
   unconfigured.close();
+  const refusing = yield* Effect.promise(() =>
+    serve(() => ({
+      status: 401,
+      type: "text/plain; charset=utf-8",
+      headers: { "www-authenticate": 'Bearer realm="acme"' },
+      body: "Unauthorized: the session was reset by a new invite code; log in at /login, or send the token of POST /login/token as Authorization: Bearer",
+    })),
+  );
+  yield* probe("tenant refusing a stale token (401)", refusing.url, "stale-token");
+  refusing.close();
+  const failingClosed = yield* Effect.promise(() =>
+    serve(() => ({
+      status: 503,
+      type: "text/plain; charset=utf-8",
+      headers: { "retry-after": "60" },
+      body: "Service unavailable: SESSION_SECRET is not set, so no request to a cell can be checked for a login; set SESSION_SECRET and INVITE_SECRET as secrets of this deployment (docs/cell-edition.md, Login)",
+    })),
+  );
+  yield* probe("tenant with a base domain and no SESSION_SECRET (503)", failingClosed.url);
+  failingClosed.close();
   yield* probe("closed port", "http://127.0.0.1:9");
 });