♻️ refactor(tasks): move main claim authority to daemon

apps/spark-daemon/src/daemon-start.ts

@@ -110,6 +110,11 @@ import {
   ARTIFACT_PROJECTION_RECONCILE_INTERVAL_MS,
   ArtifactProjectionReconciler,
 } from "./artifact-projection.ts";
+import {
+  MAIN_TASK_CLAIM_RECONCILE_INTERVAL_MS,
+  MAIN_TASK_CLAIM_STARTUP_RECOVERY_WINDOW_MS,
+} from "./task-claims/policy.ts";
+import { reconcileMainTaskClaims } from "./task-claims/reconciler.ts";
 
 export async function startSparkDaemon(options: StartSparkDaemonOptions): Promise<void> {
   const runtime = await createPreparedDaemonRuntime(options);
@@ -140,6 +145,7 @@ interface DaemonServingLoops {
   channelDelivery?: Promise<void>;
   channelReply?: Promise<void>;
   notification?: Promise<void>;
+  taskClaims?: Promise<void>;
 }
 
 interface RestartDrainController {
@@ -173,6 +179,7 @@ interface PreparedDaemonRuntime {
   servingGate: ServingLoopGate;
   loops: DaemonServingLoops;
   restartDrain: RestartDrainController;
+  taskClaimStartupRecoveryUntil: string;
   stopScheduler: () => void;
   stopDirectInvocations: () => void;
   stopChannelIngress: () => void;
@@ -271,6 +278,10 @@ async function createPreparedDaemonRuntime(
     closeRestartAdmission,
   });
   const servingGate = createServingLoopGate();
+  const taskClaimStartupRecoveryUntil = new Date(
+    Date.parse(options.taskClaimNow?.() ?? new Date().toISOString()) +
+      MAIN_TASK_CLAIM_STARTUP_RECOVERY_WINDOW_MS,
+  ).toISOString();
   const stopScheduler = () => scheduler?.stop();
   const stopDirectInvocations = () => invocationRegistry.stop();
   const stopChannelIngress = () => void shutdownChannelIngress("runtime-abort");
@@ -314,6 +325,7 @@ async function createPreparedDaemonRuntime(
     servingGate,
     loops: {},
     restartDrain,
+    taskClaimStartupRecoveryUntil,
     stopScheduler,
     stopDirectInvocations,
     stopChannelIngress,
@@ -441,6 +453,7 @@ function createServingLoopGate(): ServingLoopGate {
 
 async function prepareDaemonServing(runtime: PreparedDaemonRuntime): Promise<void> {
   const { options, runtimeSignal, channelIngress } = runtime;
+  await reconcileMainTaskClaimsBeforeAdmission(runtime);
   if (!runtimeSignal.aborted) {
     await options.onReady?.({
       channelIngress,
@@ -456,6 +469,40 @@ async function prepareDaemonServing(runtime: PreparedDaemonRuntime): Promise<voi
   commitDaemonServingFence(runtime);
 }
 
+async function reconcileMainTaskClaimsBeforeAdmission(
+  runtime: PreparedDaemonRuntime,
+): Promise<void> {
+  const result = await reconcileMainTaskClaims(runtime.options.db, {
+    now: runtime.options.taskClaimNow?.(),
+    startupRecoveryUntil: runtime.taskClaimStartupRecoveryUntil,
+  });
+  if (result.degraded.length > 0) {
+    throw new Error(
+      `Task claim startup reconciliation failed: ${result.degraded
+        .map((entry) => `${entry.workspaceId}: ${entry.error}`)
+        .join("; ")}`,
+    );
+  }
+}
+
+async function runMainTaskClaimReconcileLoop(runtime: PreparedDaemonRuntime): Promise<void> {
+  const intervalMs =
+    runtime.options.taskClaimReconcileIntervalMs ?? MAIN_TASK_CLAIM_RECONCILE_INTERVAL_MS;
+  while (!runtime.runtimeSignal.aborted) {
+    await delayUnlessAborted(intervalMs, runtime.runtimeSignal);
+    if (runtime.runtimeSignal.aborted) return;
+    const result = await reconcileMainTaskClaims(runtime.options.db, {
+      now: runtime.options.taskClaimNow?.(),
+      startupRecoveryUntil: runtime.taskClaimStartupRecoveryUntil,
+    });
+    for (const degraded of result.degraded) {
+      console.error(
+        `[spark-daemon] task claim reconcile degraded for ${degraded.workspaceId}: ${degraded.error}`,
+      );
+    }
+  }
+}
+
 function canOpenDaemonAdmission(runtime: PreparedDaemonRuntime): boolean {
   return !runtime.runtimeSignal.aborted && !runtime.options.drainSignal?.aborted;
 }
@@ -470,6 +517,12 @@ async function activateDaemonAdmission(runtime: PreparedDaemonRuntime): Promise<
 
 function startDaemonServingLoops(runtime: PreparedDaemonRuntime): void {
   const { scheduler, channelIngress, options, runtimeSignal, servingGate, loops } = runtime;
+  if (!options.once) {
+    loops.taskClaims = servingGate.promise.then(async (committed) => {
+      if (!committed || runtimeSignal.aborted) return;
+      await runMainTaskClaimReconcileLoop(runtime);
+    });
+  }
   if (scheduler && !options.once) {
     loops.scheduler = servingGate.promise.then(async (committed) => {
       if (committed && !runtimeSignal.aborted) await runSchedulerLoop(runtime);
@@ -612,6 +665,7 @@ async function cleanupPreparedDaemonRuntime(runtime: PreparedDaemonRuntime): Pro
   await runtime.loops.channelDelivery;
   await runtime.loops.notification;
   await runtime.loops.channelReply;
+  await runtime.loops.taskClaims;
   if (options.managePidFile !== false && existsSync(options.paths.pidFile)) {
     rmSync(options.paths.pidFile, { force: true });
   }

apps/spark-daemon/src/daemon.ts

@@ -166,6 +166,9 @@ export interface StartSparkDaemonOptions {
   mailStore?: SparkSessionMailStore;
   notificationReconcileIntervalMs?: number;
   channelDeliveryReconcileIntervalMs?: number;
+  /** Testable clock for daemon-owned main task claim reconciliation. */
+  taskClaimNow?: () => string;
+  taskClaimReconcileIntervalMs?: number;
   /** Bind readiness transport while externally observable work admission is still closed. */
   onReady?: (runtime: {
     channelIngress: DaemonChannelIngressRuntime | null;

apps/spark-daemon/src/local-rpc/handlers/task-claim.ts

@@ -0,0 +1,28 @@
+import {
+  acquireMainTaskClaim,
+  recoverMainTaskClaim,
+  releaseMainTaskClaim,
+} from "../../task-claims/authority.ts";
+import type { LocalRpcDispatchContext } from "./context.ts";
+import type { LocalRpcServiceOutput, LocalRpcServiceRequest } from "../types.ts";
+
+type TaskClaimRequest = Extract<
+  LocalRpcServiceRequest,
+  {
+    method: "task.claim.acquire" | "task.claim.release" | "task.claim.recover";
+  }
+>;
+
+export async function handleTaskClaimRequest(
+  ctx: LocalRpcDispatchContext,
+  request: TaskClaimRequest,
+): Promise<LocalRpcServiceOutput<TaskClaimRequest>> {
+  switch (request.method) {
+    case "task.claim.acquire":
+      return await acquireMainTaskClaim(ctx.db, request.params);
+    case "task.claim.release":
+      return await releaseMainTaskClaim(ctx.db, request.params);
+    case "task.claim.recover":
+      return await recoverMainTaskClaim(ctx.db, request.params);
+  }
+}

apps/spark-daemon/src/local-rpc/orpc-router.ts

@@ -1,8 +1,3 @@
-/**
- * oRPC router for local-rpc methods.
- *
- * Every contracted method calls the transport-neutral daemon service directly.
- */
 import type { DatabaseSync } from "node:sqlite";
 import { ORPCError, implement } from "@orpc/server";
 import {
@@ -171,6 +166,19 @@ export function createLocalRpcOrpcRouter(input: CreateLocalRpcOrpcRouterOptions)
         ),
       },
     },
+    task: {
+      claim: {
+        acquire: os.task.claim.acquire.handler(async ({ input: params }) =>
+          invoke("task.claim.acquire", params),
+        ),
+        release: os.task.claim.release.handler(async ({ input: params }) =>
+          invoke("task.claim.release", params),
+        ),
+        recover: os.task.claim.recover.handler(async ({ input: params }) =>
+          invoke("task.claim.recover", params),
+        ),
+      },
+    },
     uplink: {
       park: os.uplink.park.handler(async ({ input: params }) => invoke("uplink.park", params)),
       unpark: os.uplink.unpark.handler(async ({ input: params }) =>

apps/spark-daemon/src/local-rpc/parse.test.ts

@@ -43,9 +43,9 @@ describe("side-thread local RPC parsing", () => {
     }
   });
 
-  it("recognizes exactly the 67 protocol-owned methods", () => {
+  it("recognizes exactly the 70 protocol-owned methods", () => {
     const methods = Object.keys(sparkLocalRpcProcedureSchemas);
-    expect(methods).toHaveLength(67);
+    expect(methods).toHaveLength(70);
     expect(methods.every(isSparkLocalRpcMethod)).toBe(true);
     expect(() =>
       parseLocalRpcRequest(JSON.stringify({ id: "unknown", method: "legacy.unknown", params: {} })),

apps/spark-daemon/src/local-rpc/service.test.ts

@@ -103,11 +103,11 @@ describe("transport-neutral local RPC service", () => {
     db.close();
   });
 
-  it("exhaustively groups all 67 methods behind their protocol output parser", () => {
+  it("exhaustively groups all 69 methods behind their protocol output parser", () => {
     const groupedMethods = Object.values(localRpcServiceHandlerMethodGroups).flat();
     const catalogMethods = Object.keys(sparkLocalRpcProcedureSchemas) as SparkLocalRpcMethod[];
 
-    expect(groupedMethods).toHaveLength(67);
+    expect(groupedMethods).toHaveLength(70);
     expect(new Set(groupedMethods).size).toBe(groupedMethods.length);
     expect([...groupedMethods].sort()).toEqual([...catalogMethods].sort());
 

apps/spark-daemon/src/local-rpc/service.ts

@@ -19,6 +19,7 @@ import { handleHumanRequest } from "./handlers/human.ts";
 import { handleModelRequest } from "./handlers/model.ts";
 import { handleSessionRequest } from "./handlers/session.ts";
 import { handleSideThreadRequest } from "./handlers/side-thread.ts";
+import { handleTaskClaimRequest } from "./handlers/task-claim.ts";
 import { handleTurnRequest } from "./handlers/turn.ts";
 import { handleUplinkRequest } from "./handlers/uplink.ts";
 import { handleWorkspaceRequest } from "./handlers/workspace.ts";
@@ -80,6 +81,7 @@ export const localRpcServiceHandlerMethodGroups = {
     "workspace.client.release",
     "workspace.executor.ensure",
   ],
+  taskClaim: ["task.claim.acquire", "task.claim.release", "task.claim.recover"],
   session: [
     "session.notification.deliver",
     "session.list",
@@ -223,6 +225,9 @@ async function dispatchLocalRpcServiceRequest(
   if (requestBelongsToHandlerGroup(request, "workspace")) {
     return handleWorkspaceRequest(context, request);
   }
+  if (requestBelongsToHandlerGroup(request, "taskClaim")) {
+    return handleTaskClaimRequest(context, request);
+  }
   if (requestBelongsToHandlerGroup(request, "session")) {
     return handleSessionRequest(context, request);
   }

apps/spark-daemon/src/store/workspaces.ts

@@ -14,6 +14,7 @@ import {
   type WorkspaceOccupancySession,
   type WorkspaceSessionSurface,
 } from "@zendev-lab/spark-protocol";
+import type { SparkTaskClaimLeaseIdentity } from "@zendev-lab/spark-protocol/task-claim";
 import { asciiSlug } from "@zendev-lab/spark-system";
 import { SparkDaemonControlError } from "../control-error.ts";
 
@@ -694,6 +695,30 @@ export function workspaceKeyForPath(localPath: string): string {
   return workspaceKeyForName(basename(normalizeLocalPath(localPath)));
 }
 
+export interface SparkDaemonWorkspaceClaimTarget {
+  id: string;
+  localPath: string;
+}
+
+export function listWorkspaceClaimTargets(db: DatabaseSync): SparkDaemonWorkspaceClaimTarget[] {
+  return db
+    .prepare("SELECT id, local_path AS localPath FROM daemon_workspaces ORDER BY id")
+    .all() as unknown as SparkDaemonWorkspaceClaimTarget[];
+}
+
+export function requireWorkspaceClaimTarget(
+  db: DatabaseSync,
+  workspaceId: string,
+): SparkDaemonWorkspaceClaimTarget {
+  const target = db
+    .prepare("SELECT id, local_path AS localPath FROM daemon_workspaces WHERE id = ? LIMIT 1")
+    .get(workspaceId) as SparkDaemonWorkspaceClaimTarget | undefined;
+  if (!target) {
+    throw new SparkDaemonControlError("workspace_not_found", `Unknown workspace: ${workspaceId}`);
+  }
+  return target;
+}
+
 export function listWorkspaces(db: DatabaseSync): SparkDaemonWorkspace[] {
   const rows = db
     .prepare(
@@ -1177,6 +1202,34 @@ export function expireWorkspaceClientLeases(
   return Number(result.changes ?? 0);
 }
 
+export function requireFencedSessionWorkspaceClient(
+  db: DatabaseSync,
+  identity: SparkTaskClaimLeaseIdentity,
+  now = new Date().toISOString(),
+): SparkDaemonWorkspaceClient {
+  const client = getWorkspaceClientById(db, identity.clientId);
+  if (
+    !client ||
+    client.workspaceId !== identity.workspaceId ||
+    client.sessionId !== identity.sessionId ||
+    client.kind !== "interactive"
+  ) {
+    throw new SparkDaemonControlError(
+      "task_claim_lease_invalid",
+      `Workspace client ${identity.clientId} is not the active interactive lease for ${identity.sessionId}.`,
+    );
+  }
+  try {
+    assertWorkspaceClientLease(client, identity.leaseFence, now);
+  } catch (error) {
+    throw new SparkDaemonControlError(
+      "task_claim_lease_invalid",
+      error instanceof Error ? error.message : `Invalid task claim lease: ${identity.clientId}`,
+    );
+  }
+  return client;
+}
+
 export function listWorkspaceClients(
   db: DatabaseSync,
   workspaceId?: string,