@@ -1,45 +1,15 @@
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
-import { logger } from "~/services/logger.server";
 import { BoundedTtlCache } from "~/services/realtime/boundedTtlCache";
 import { singleton } from "~/utils/singleton";
+import { logger } from "~/services/logger.server";
 import { FEATURE_FLAG } from "~/v3/featureFlags";
+import { computeWaitpointMintKind, type WaitpointMintKind } from "./waitpointMintKind.js";
 
-/**
- * Which coordinator mints a NEW waitpoint. Consulted at the mint and never again: every
- * later operation routes by id shape. A flip therefore changes only where the NEXT
- * waitpoint is born, which is why this needs no flip-grace machinery.
- */
-export type WaitpointMintKind = "legacy" | "store";
+export { computeWaitpointMintKind, type WaitpointMintKind };
 
-/** The flag's vocabulary, deliberately not the coordinator's. */
 type WaitpointSystemFlag = "legacy" | "redis";
 
-type MintKindDeps = {
-  globalDefault: WaitpointSystemFlag;
-  /** Undefined when the org has no override. Must not hit the DB when given org flags. */
-  flag: (
-    orgId: string,
-    orgFeatureFlags: unknown | undefined
-  ) => Promise<WaitpointSystemFlag | undefined>;
-};
-
-// PURE CORE — no env import; the tests drive this directly.
-export async function computeWaitpointMintKind(
-  environment: { organizationId: string; id: string; orgFeatureFlags?: unknown },
-  deps: MintKindDeps
-): Promise<WaitpointMintKind> {
-  try {
-    const perOrg = await deps.flag(environment.organizationId, environment.orgFeatureFlags);
-    return (perOrg ?? deps.globalDefault) === "redis" ? "store" : "legacy";
-  } catch (error) {
-    // Fail safe, as computeRunIdMintKind does: a flag-read failure degrades to the old
-    // path rather than becoming a trigger-path outage.
-    logger.error("[waitpointMintKind] flag read failed; minting legacy (fail-safe)", { error });
-    return "legacy";
-  }
-}
-
 const mintCache = singleton(
   "waitpointMintCache",
   () =>
@@ -58,6 +28,8 @@ export async function resolveWaitpointMintKind(environment: {
 }): Promise<WaitpointMintKind> {
   return computeWaitpointMintKind(environment, {
     globalDefault: env.WAITPOINT_SYSTEM_DEFAULT,
+    onError: (error) =>
+      logger.error("[waitpointMintKind] flag read failed; minting legacy (fail-safe)", { error }),
     flag: async (orgId, orgFeatureFlags) => {
       // null is a cached "this org has no override", which must stay distinct from a miss:
       // BoundedTtlCache reports a stored undefined as a miss, so never store undefined.
 
@@ -1,5 +1,5 @@
 import { describe, expect, it, vi } from "vitest";
-import { computeWaitpointMintKind } from "./waitpointMintKind.server";
+import { computeWaitpointMintKind } from "./waitpointMintKind";
 
 const environment = { organizationId: "org_1", id: "env_1" };
 
 
@@ -0,0 +1,37 @@
+// Pure: no server-only imports, so a test can drive this without loading env.server.
+/**
+ * Which coordinator mints a NEW waitpoint. Consulted at the mint and never again: every
+ * later operation routes by id shape. A flip therefore changes only where the NEXT
+ * waitpoint is born, which is why this needs no flip-grace machinery.
+ */
+export type WaitpointMintKind = "legacy" | "store";
+
+/** The flag's vocabulary, deliberately not the coordinator's. */
+type WaitpointSystemFlag = "legacy" | "redis";
+
+type MintKindDeps = {
+  globalDefault: WaitpointSystemFlag;
+  /** Undefined when the org has no override. Must not hit the DB when given org flags. */
+  flag: (
+    orgId: string,
+    orgFeatureFlags: unknown | undefined
+  ) => Promise<WaitpointSystemFlag | undefined>;
+  /** Surfaced instead of logged, so this module pulls in no server-only import. */
+  onError?: (error: unknown) => void;
+};
+
+// PURE CORE — no env import; the tests drive this directly.
+export async function computeWaitpointMintKind(
+  environment: { organizationId: string; id: string; orgFeatureFlags?: unknown },
+  deps: MintKindDeps
+): Promise<WaitpointMintKind> {
+  try {
+    const perOrg = await deps.flag(environment.organizationId, environment.orgFeatureFlags);
+    return (perOrg ?? deps.globalDefault) === "redis" ? "store" : "legacy";
+  } catch (error) {
+    // Fail safe, as computeRunIdMintKind does: a flag-read failure degrades to the old
+    // path rather than becoming a trigger-path outage.
+    deps.onError?.(error);
+    return "legacy";
+  }
+}
@@ -295,6 +295,60 @@ describe("StoreWaitpointCoordinatorArm", () => {
     }
   );
 
+  // The token API and dashboard read status, output and completedAt from the projection
+  // row, so a completion that never reaches it reports a finished token as still waiting.
+  containerTest(
+    "reflects a MANUAL completion onto the projection row",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupEnvironment(prisma);
+      const { store, arm } = setup(redisOptions, prisma);
+
+      try {
+        const created = await arm.createManualWaitpoint({
+          mintKind: "store",
+          environmentId: environment.id,
+          projectId: environment.projectId,
+        });
+
+        await arm.complete({
+          waitpointId: created.waitpoint.id,
+          output: { value: '{"done":true}', type: "application/json", isError: false },
+        });
+
+        const row = await prisma.waitpoint.findFirst({ where: { id: created.waitpoint.id } });
+        expect(row?.status).toBe("COMPLETED");
+        expect(row?.output).toBe('{"done":true}');
+        expect(row?.outputIsError).toBe(false);
+        expect(row?.completedAt).not.toBeNull();
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  // An unwired caller must fail, never silently disable the guard.
+  containerTest(
+    "refuses a lockless absorb that arrives with no parent BATCH waitpoint id",
+    async ({ prisma, redisOptions }) => {
+      const environment = await setupEnvironment(prisma);
+      const { store, arm } = setup(redisOptions, prisma);
+
+      try {
+        await expect(
+          arm.registerBlocksLockless({
+            runId: RUN_ID,
+            waitpointIds: [generateWaitpointId("RUN")],
+            projectId: environment.projectId,
+            batchId: "batch_1",
+            batchIndex: 0,
+          })
+        ).rejects.toThrow(/no parent .*BATCH waitpoint id/);
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
   containerTest(
     "returns the cached waitpoint for a repeated idempotency key",
     async ({ prisma, redisOptions }) => {
 
@@ -164,12 +164,24 @@ export class StoreWaitpointCoordinatorArm implements WaitpointCoordinator {
    * open. If it is absent or already complete, a concurrent completion could see an empty
    * pending set mid-absorb and resume the parent early.
    *
-   * Neither TLA+ campaign models this variant, so this assertion is its only protection
-   * until the race harness covers it.
+   * Scope, stated precisely: this is a PREFLIGHT DETECTOR, not a barrier. It reads the run
+   * shard, then the absorb writes in a separate operation, so a completion landing between
+   * the two is detected on the next call, not prevented. Closing that window means moving
+   * the pending-set assertion inside the absorb script, so check and write share one
+   * atomic action.
+   *
+   * Neither TLA+ campaign models this variant, so until the race harness covers it this
+   * detector plus the fail-loud on a missing id is the whole protection.
    */
   async #assertBatchWaitpointPending(params: RegisterBlocksLocklessParams): Promise<void> {
     if (!params.batchWaitpointId) {
-      return;
+      // Never silently skip. An unwired caller would disable the guard rather than fail,
+      // which is the failure mode the guard exists to prevent.
+      this.batchGuardViolations.add(1);
+      throw new Error(
+        `Lockless absorb for run ${params.runId} reached the store arm with no parent ` +
+          `BATCH waitpoint id, so the pending-set guard has nothing to assert on`
+      );
     }
 
     const state = await this.store.readBlockState(params.runId);
@@ -256,6 +268,10 @@ export class StoreWaitpointCoordinatorArm implements WaitpointCoordinator {
       throw new Error(`Waitpoint ${waitpointId} is not present in the store`);
     }
 
+    if (held.record.type === "MANUAL") {
+      await this.#completeManualProjection(waitpointId, held.completion ?? completion);
+    }
+
     return {
       waitpoint: toPrismaWaitpoint(held.record, held.status, held.completion),
       blockedRuns: result.watchers.map((watcher) => ({
@@ -470,6 +486,41 @@ export class StoreWaitpointCoordinatorArm implements WaitpointCoordinator {
     }
   }
 
+  /**
+   * Reflect a MANUAL completion onto the projection row.
+   *
+   * The token API and the dashboard read status, output and completedAt from this row, so
+   * leaving it PENDING would report a completed token as still waiting. Best effort, for
+   * the same reason as the create-time write: the store already completed the waitpoint.
+   */
+  async #completeManualProjection(
+    waitpointId: string,
+    completion: WaitpointCompletion
+  ): Promise<void> {
+    const output = completion.output;
+
+    const [error] = await tryCatch(
+      this.runStore.updateManyWaitpoints({
+        where: { id: waitpointId },
+        data: {
+          status: "COMPLETED",
+          completedAt: new Date(completion.completedAt),
+          output: output ? ("inline" in output ? output.inline : output.ref) : null,
+          outputType: completion.outputType,
+          outputIsError: completion.outputIsError,
+        },
+      })
+    );
+
+    if (error) {
+      this.projectionWriteFailures.add(1);
+      this.logger.error("failed to complete the MANUAL waitpoint projection row", {
+        waitpointId,
+        error,
+      });
+    }
+  }
+
   #record(params: {
     id: string;
     friendlyId?: string;
 
@@ -28,6 +28,12 @@ export type WaitpointCoordinator = {
   mintAssociatedWaitpointData(params: {
     projectId: string;
     environmentId: string;
+    /**
+     * The run this waitpoint belongs to. A store arm derives the waitpoint id from the
+     * run's own id body, so the derivation is a pure function of the anchor and needs no
+     * lock. A Postgres arm mints a fresh id and ignores this.
+     */
+    anchorRunId?: string;
   }): AssociatedWaitpointData;
   createAssociatedWaitpoint(params: {
     runId: string;