chittyos
diff --git a/‎daemon/coordinator-lease.ts‎
Lines changed: 172 additions & 0 deletions b/‎daemon/coordinator-lease.ts‎
Lines changed: 172 additions & 0 deletions
diff --git a/‎meta/coordinator.ts‎
Lines changed: 221 additions & 0 deletions b/‎meta/coordinator.ts‎
Lines changed: 221 additions & 0 deletions
@@ -0,0 +1,172 @@
+/**
+ * Coordinator-backed lease client — drop-in replacement for daemon/leader.ts.
+ *
+ * Exports the same four functions with the same signatures and return shapes,
+ * so daemon/loop.ts switches by changing one import path. The arbiter moves
+ * from Neon `cc_node_leases` to the CommandCoordinator Durable Object
+ * (meta/coordinator.ts).
+ *
+ * Why: ADR-001 used Neon leases to elect a leader across chittymini-01..06.
+ * A DO is already a strongly-consistent singleton, so election is unnecessary —
+ * and the Neon dependency, which is cost-driven pressure, leaves this layer.
+ *
+ * Fails closed. With no coordinator configured this throws rather than
+ * degrading to an unarbitrated local decision, which would permit split-brain.
+ *
+ * @canonical-uri chittycanon://docs/architecture/chittycommand/ADR-001
+ */
+
+// Import from lease-types, NOT meta/coordinator: the latter evaluates
+// `cloudflare:workers`, which does not resolve on Node and would crash the
+// daemon at module load, before main() and before any log line.
+import { META_LEADER_ROLE, type StoredLease } from '../meta/lease-types';
+
+export { META_LEADER_ROLE };
+
+export const POLICY_BLOCKED_COORDINATOR_UNAVAILABLE =
+  'POLICY_BLOCKED_COORDINATOR_UNAVAILABLE';
+
+export interface CoordinatorEnv {
+  /** Base URL of the ChittyCommand worker, e.g. https://command.chitty.cc */
+  COORDINATOR_URL?: string;
+  /** Bearer token for the coordinator routes. Broker-provided; never inlined. */
+  COORDINATOR_TOKEN?: string;
+}
+
+/** Identical to NodeLease in daemon/leader.ts. */
+export interface NodeLease {
+  role: string;
+  nodeId: string;
+  nodeDescriptor: string | null;
+  sessionId: string | null;
+  claimedAt: Date;
+  heartbeatAt: Date;
+  leaseExpiresAt: Date;
+  metadata: Record<string, unknown>;
+}
+
+export interface ClaimOptions {
+  nodeId: string;
+  nodeDescriptor?: string;
+  sessionId?: string;
+  leaseSeconds?: number;
+  role?: string;
+  metadata?: Record<string, unknown>;
+}
+
+/**
+ * Both the URL and the token are required. A configured URL with no token
+ * yields a 401 on every call, which loop.ts logs as a claim error and retries
+ * forever — a permanently dead daemon whose logs read like a transient auth
+ * blip. Fail closed on the config error instead, with a distinguishable code.
+ */
+function requireConfig(env: CoordinatorEnv): { url: string; token: string } {
+  const url = env.COORDINATOR_URL?.replace(/\/+$/, '');
+  if (!url) throw new Error(POLICY_BLOCKED_COORDINATOR_UNAVAILABLE);
+  if (!env.COORDINATOR_TOKEN) throw new Error(POLICY_BLOCKED_COORDINATOR_UNAVAILABLE);
+  return { url, token: env.COORDINATOR_TOKEN };
+}
+
+async function call<T>(
+  env: CoordinatorEnv,
+  method: 'GET' | 'POST',
+  path: string,
+  body?: unknown,
+): Promise<T> {
+  const { url, token } = requireConfig(env);
+  const headers: Record<string, string> = {
+    'content-type': 'application/json',
+    authorization: `Bearer ${token}`,
+  };
+
+  const res = await fetch(`${url}/api/meta/coordinator${path}`, {
+    method,
+    headers,
+    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
+  });
+
+  if (!res.ok) {
+    throw new Error(
+      `[daemon/coordinator-lease] ${method} ${path} failed: ${res.status} ${await res.text()}`,
+    );
+  }
+  return (await res.json()) as T;
+}
+
+/** Rehydrate ISO strings into Dates. Returns null for an unheld lease. */
+function toLease(stored: StoredLease | null): NodeLease | null {
+  if (!stored?.nodeId || !stored.claimedAt || !stored.heartbeatAt || !stored.leaseExpiresAt) {
+    return null;
+  }
+  return {
+    role: stored.role,
+    nodeId: stored.nodeId,
+    nodeDescriptor: stored.nodeDescriptor,
+    sessionId: stored.sessionId,
+    claimedAt: new Date(stored.claimedAt),
+    heartbeatAt: new Date(stored.heartbeatAt),
+    leaseExpiresAt: new Date(stored.leaseExpiresAt),
+    metadata: stored.metadata ?? {},
+  };
+}
+
+export async function claimLeadership(
+  env: CoordinatorEnv,
+  options: ClaimOptions,
+): Promise<NodeLease | null> {
+  if (!options?.nodeId) throw new Error('[daemon/coordinator-lease] nodeId is required');
+  return toLease(
+    await call<StoredLease | null>(env, 'POST', '/claim', {
+      nodeId: options.nodeId,
+      nodeDescriptor: options.nodeDescriptor ?? null,
+      sessionId: options.sessionId ?? null,
+      leaseSeconds: options.leaseSeconds,
+      role: options.role,
+      metadata: options.metadata ?? {},
+    }),
+  );
+}
+
+export async function heartbeat(
+  env: CoordinatorEnv,
+  nodeId: string,
+  options: { role?: string; leaseSeconds?: number; sessionId?: string | null } = {},
+): Promise<NodeLease | null> {
+  if (!nodeId) throw new Error('[daemon/coordinator-lease] nodeId is required for heartbeat');
+  return toLease(
+    await call<StoredLease | null>(env, 'POST', '/heartbeat', {
+      nodeId,
+      role: options.role,
+      leaseSeconds: options.leaseSeconds,
+      sessionId: options.sessionId ?? null,
+    }),
+  );
+}
+
+export async function releaseLeadership(
+  env: CoordinatorEnv,
+  nodeId: string,
+  options: { role?: string; sessionId?: string | null } = {},
+): Promise<boolean> {
+  if (!nodeId) throw new Error('[daemon/coordinator-lease] nodeId is required for release');
+  const res = await call<{ released: boolean }>(env, 'POST', '/release', {
+    nodeId,
+    role: options.role,
+    sessionId: options.sessionId ?? null,
+  });
+  return res.released === true;
+}
+
+export async function describeLease(
+  env: CoordinatorEnv,
+  options: { role?: string } = {},
+): Promise<NodeLease | null> {
+  const role = options.role ?? META_LEADER_ROLE;
+  return toLease(
+    await call<StoredLease | null>(
+      env,
+      'GET',
+      `/describe?role=${encodeURIComponent(role)}`,
+    ),
+  );
+}
@@ -0,0 +1,221 @@
+/**
+ * CommandCoordinator — Durable Object arbiter for meta-orchestrator role leases.
+ *
+ * Replaces the Neon `cc_node_leases` table (daemon/leader.ts) as the *arbiter*
+ * of who holds a role. A Durable Object is already a strongly-consistent,
+ * single-threaded singleton, so the properties the Neon lease bought us come
+ * free:
+ *
+ *   - mutual exclusion: the DO serializes requests; no two claimers race
+ *   - durability: DO storage survives eviction and restart
+ *   - availability: runs on the Cloudflare edge, not on operator hardware
+ *
+ * ADR-001 elected a leader across `chittymini-01..06` via Neon leases so the
+ * coordinator would never be down. That mechanism is unnecessary here — there
+ * is no fleet to elect across, because the DO *is* the leader. Nodes remain
+ * meaningful as executors that pull work (they have local filesystem and repo
+ * access, which is the real reason to run on hardware); they are no longer
+ * candidates for leadership.
+ *
+ * Wire semantics are a deliberate 1:1 port of the SQL in daemon/leader.ts,
+ * including two behaviours established by prior review:
+ *   - heartbeat requires (role, nodeId, sessionId) to match the holder
+ *     (codex-p2 PR#101 finding-5)
+ *   - release requires the same triple (codex-p2 PR#101 finding-2)
+ *
+ * @canonical-uri chittycanon://docs/architecture/chittycommand/ADR-001
+ * @canon chittycanon://gov/governance#core-types — a node is a Location (L);
+ *        a lease claim is an Event (E).
+ */
+
+import { DurableObject } from 'cloudflare:workers';
+
+import {
+  DEFAULT_LEASE_SECONDS,
+  MAX_LEASE_SECONDS,
+  META_LEADER_ROLE,
+  MIN_LEASE_SECONDS,
+  normalizeLeaseSeconds,
+  type ClaimBody,
+  type StoredLease,
+} from './lease-types';
+
+// Re-exported so existing importers of this module keep working. The
+// definitions live in lease-types.ts because the daemon runs on Node and
+// cannot evaluate `cloudflare:workers`.
+export {
+  DEFAULT_LEASE_SECONDS,
+  MAX_LEASE_SECONDS,
+  META_LEADER_ROLE,
+  MIN_LEASE_SECONDS,
+  normalizeLeaseSeconds,
+};
+export type { ClaimBody, StoredLease };
+
+/** Storage key for a role's lease. */
+const keyFor = (role: string) => `lease:${role}`;
+
+export class CommandCoordinator extends DurableObject {
+  async fetch(request: Request): Promise<Response> {
+    const url = new URL(request.url);
+    // Anchor on the FIRST '/coordinator' segment. A greedy match here let
+    // `/api/meta/coordinator/a/coordinator/release` dispatch `release`.
+    const marker = '/coordinator';
+    const at = url.pathname.indexOf(marker);
+    const path = at === -1 ? '/' : url.pathname.slice(at + marker.length) || '/';
+
+    try {
+      switch (`${request.method} ${path}`) {
+        case 'POST /claim':
+          return json(await this.claim(await readJson<ClaimBody>(request)));
+        case 'POST /heartbeat':
+          return json(await this.heartbeat(await readJson(request)));
+        case 'POST /release':
+          return json({ released: await this.release(await readJson(request)) });
+        case 'GET /describe':
+          return json(await this.describe(url.searchParams.get('role') ?? META_LEADER_ROLE));
+        default:
+          return json({ error: 'not_found', path }, 404);
+      }
+    } catch (err) {
+      const message = err instanceof Error ? err.message : String(err);
+      return json({ error: 'coordinator_error', message }, 400);
+    }
+  }
+
+  private async read(role: string): Promise<StoredLease | undefined> {
+    return this.ctx.storage.get<StoredLease>(keyFor(role));
+  }
+
+  /**
+   * Claim `role`. Succeeds when the role is unheld, already held by this same
+   * node, or the incumbent's lease has expired.
+   *
+   * Parity note: `claimedAt` is preserved across takeover, mirroring
+   * `COALESCE(claimed_at, NOW())` in the SQL. That means a takeover from an
+   * expired holder reports the *previous* holder's claim time. This is
+   * deliberate and is not a defect: `claimedAt` reads as "when this role was
+   * first continuously held", which is a coherent semantic.
+   *
+   * It is also load-bearing, not diagnostic — daemon/coordinator-lease.ts
+   * returns null when `claimedAt` is falsy, so it participates in the
+   * lease/no-lease decision. Any future change has a second consumer.
+   */
+  async claim(body: ClaimBody): Promise<StoredLease | null> {
+    if (!body?.nodeId) throw new Error('[meta/coordinator] nodeId is required');
+
+    const role = body.role ?? META_LEADER_ROLE;
+    const leaseSeconds = normalizeLeaseSeconds(body.leaseSeconds);
+    const now = Date.now();
+    const current = await this.read(role);
+
+    // A corrupted timestamp parses to NaN, and every NaN comparison is false —
+    // which would leave the role permanently unclaimable. Postgres typed this
+    // column so the SQL had no such failure mode; treat unparseable as expired.
+    const expiresAt = current?.leaseExpiresAt ? Date.parse(current.leaseExpiresAt) : NaN;
+    const expired = !Number.isFinite(expiresAt) || expiresAt < now;
+    const claimable = !current?.nodeId || current.nodeId === body.nodeId || expired;
+    if (!claimable) return null;
+
+    const nowIso = new Date(now).toISOString();
+    const lease: StoredLease = {
+      role,
+      nodeId: body.nodeId,
+      nodeDescriptor: body.nodeDescriptor ?? null,
+      sessionId: body.sessionId ?? null,
+      claimedAt: current?.claimedAt ?? nowIso,
+      heartbeatAt: nowIso,
+      leaseExpiresAt: new Date(now + leaseSeconds * 1000).toISOString(),
+      metadata: body.metadata ?? {},
+    };
+
+    await this.ctx.storage.put(keyFor(role), lease);
+    return lease;
+  }
+
+  /**
+   * Extend the lease. Returns null when this node is no longer the holder, or
+   * when `sessionId` does not match the session recorded on the lease — a
+   * restarted process reusing a nodeId cannot heartbeat over a fresh leader.
+   */
+  async heartbeat(body: {
+    nodeId: string;
+    role?: string;
+    leaseSeconds?: number;
+    sessionId?: string | null;
+  }): Promise<StoredLease | null> {
+    if (!body?.nodeId) throw new Error('[meta/coordinator] nodeId is required for heartbeat');
+
+    const role = body.role ?? META_LEADER_ROLE;
+    const current = await this.read(role);
+    if (!current || current.nodeId !== body.nodeId) return null;
+    if (current.sessionId !== (body.sessionId ?? null)) return null;
+
+    const now = Date.now();
+    const leaseSeconds = normalizeLeaseSeconds(body.leaseSeconds);
+    const lease: StoredLease = {
+      ...current,
+      heartbeatAt: new Date(now).toISOString(),
+      leaseExpiresAt: new Date(now + leaseSeconds * 1000).toISOString(),
+    };
+
+    await this.ctx.storage.put(keyFor(role), lease);
+    return lease;
+  }
+
+  /**
+   * Release the role. Only the holding (nodeId, sessionId) pair may release;
+   * a different node or a newer session of the same node is a no-op.
+   */
+  async release(body: {
+    nodeId: string;
+    role?: string;
+    sessionId?: string | null;
+  }): Promise<boolean> {
+    if (!body?.nodeId) throw new Error('[meta/coordinator] nodeId is required for release');
+
+    const role = body.role ?? META_LEADER_ROLE;
+    const current = await this.read(role);
+    if (!current || current.nodeId !== body.nodeId) return false;
+    if (current.sessionId !== (body.sessionId ?? null)) return false;
+
+    await this.ctx.storage.put(keyFor(role), {
+      role,
+      nodeId: null,
+      nodeDescriptor: null,
+      sessionId: null,
+      claimedAt: null,
+      heartbeatAt: null,
+      leaseExpiresAt: null,
+      metadata: current.metadata,
+    } satisfies StoredLease);
+    return true;
+  }
+
+  /**
+   * Inspect the lease without mutating it. Returns null when unheld.
+   *
+   * Parity note: an *expired but unreleased* lease is still returned, matching
+   * `describeLease()` in daemon/leader.ts, which filters only on `node_id`.
+   * Callers must not treat a non-null result as proof of live leadership.
+   */
+  async describe(role: string = META_LEADER_ROLE): Promise<StoredLease | null> {
+    const current = await this.read(role);
+    return current?.nodeId ? current : null;
+  }
+}
+
+async function readJson<T>(request: Request): Promise<T> {
+  try {
+    return (await request.json()) as T;
+  } catch {
+    throw new Error('invalid JSON body');
+  }
+}
+
+function json(data: unknown, status = 200): Response {
+  return new Response(JSON.stringify(data), {
+    status,
+    headers: { 'content-type': 'application/json' },
+  });
+}