fix: resolve 8 Codex P2 findings on PR #101

daemon/leader.ts

@@ -120,40 +120,50 @@ export async function claimLeadership(
 
 /**
  * Extend the lease. Returns null if this node is no longer the holder
- * (another node took over).
+ * (another node took over) OR if the caller's sessionId does not match the
+ * session currently recorded on the lease — a restarted process with the same
+ * nodeId cannot heartbeat over a fresh leader.
+ *
+ * fixes codex-p2 PR#101 finding-5 — session ownership required on heartbeat.
  */
 export async function heartbeat(
   env: LeaderEnv,
   nodeId: string,
-  options: { role?: string; leaseSeconds?: number } = {},
+  options: { role?: string; leaseSeconds?: number; sessionId?: string | null } = {},
 ): Promise<NodeLease | null> {
   if (!nodeId) throw new Error('[daemon/leader] nodeId is required for heartbeat');
   const sql = getSql(env);
   const role = options.role ?? META_LEADER_ROLE;
   const leaseSeconds = normalizeLeaseSeconds(options.leaseSeconds);
+  const sessionId = options.sessionId ?? null;
 
   const rows = await sql`
     UPDATE cc_node_leases
     SET heartbeat_at = NOW(),
         lease_expires_at = NOW() + (${leaseSeconds} * INTERVAL '1 second'),
         updated_at = NOW()
-    WHERE role = ${role} AND node_id = ${nodeId}
+    WHERE role = ${role}
+      AND node_id = ${nodeId}
+      AND session_id IS NOT DISTINCT FROM ${sessionId}
     RETURNING *`;
   return rows[0] ? rowToLease(rows[0]) : null;
 }
 
 /**
- * Release leadership. Only this node can release — if a different node
- * holds the role, this is a no-op (returns false).
+ * Release leadership. Only this node + session can release — if a different
+ * node or a newer session of the same node holds the role, this is a no-op.
+ *
+ * fixes codex-p2 PR#101 finding-2 — session ownership required on release.
  */
 export async function releaseLeadership(
   env: LeaderEnv,
   nodeId: string,
-  options: { role?: string } = {},
+  options: { role?: string; sessionId?: string | null } = {},
 ): Promise<boolean> {
   if (!nodeId) throw new Error('[daemon/leader] nodeId is required for release');
   const sql = getSql(env);
   const role = options.role ?? META_LEADER_ROLE;
+  const sessionId = options.sessionId ?? null;
   const rows = await sql`
     UPDATE cc_node_leases
     SET node_id = NULL,
@@ -163,7 +173,9 @@ export async function releaseLeadership(
         heartbeat_at = NULL,
         lease_expires_at = NULL,
         updated_at = NOW()
-    WHERE role = ${role} AND node_id = ${nodeId}
+    WHERE role = ${role}
+      AND node_id = ${nodeId}
+      AND session_id IS NOT DISTINCT FROM ${sessionId}
     RETURNING role`;
   return rows.length > 0;
 }

daemon/loop.ts

@@ -30,6 +30,7 @@ import {
   completeIntent,
   failIntent,
   markIntentDispatched,
+  reclaimStuckIntents,
   type Intent,
   type IntentEnv,
 } from '../meta/intent';
@@ -122,9 +123,10 @@ export async function runLeaderLoop(
     intentsProcessed = innerResult.intentsProcessed;
 
     if (innerResult.reason === 'aborted' || innerResult.reason === 'maxIntents') {
-      // Best-effort release on clean exit.
+      // Best-effort release on clean exit — pass sessionId so the release
+      // refuses to clear a newer leader's lease (fixes codex-p2 PR#101 finding-2).
       try {
-        await releaseLeadership(env, options.nodeId, { role });
+        await releaseLeadership(env, options.nodeId, { role, sessionId: options.sessionId });
       } catch (err) {
         log('release_error', { error: err instanceof Error ? err.message : String(err) });
       }
@@ -159,11 +161,22 @@ async function innerLoop(
   let intentsProcessed = startCount;
   let lastHeartbeat = Date.now();
 
+  // Heartbeat cadence inside executor.execute() — half the lease so a slow
+  // executor can't let the lease lapse mid-flight.
+  // fixes codex-p2 PR#101 finding-3
+  const innerHeartbeatMs = Math.max(1_000, Math.floor((leaseSeconds * 1000) / 2));
+
   while (!signal?.aborted) {
-    // Heartbeat if due.
+    // Heartbeat if due. Session-scoped so a restarted process can't extend
+    // a lease that already belongs to a newer leader.
+    // fixes codex-p2 PR#101 finding-5
     if (Date.now() - lastHeartbeat >= heartbeatMs) {
       try {
-        const renewed = await heartbeat(env, options.nodeId, { role, leaseSeconds });
+        const renewed = await heartbeat(env, options.nodeId, {
+          role,
+          leaseSeconds,
+          sessionId: options.sessionId,
+        });
         if (!renewed) {
           return { intentsProcessed, reason: 'leaseLost' };
         }
@@ -178,6 +191,17 @@ async function innerLoop(
       }
     }
 
+    // Reclaim intents stuck in running/claimed past 2x the lease window before
+    // we ask for new work. Idempotent and cheap; if nothing is stuck this is
+    // a single UPDATE returning 0 rows.
+    // fixes codex-p2 PR#101 finding-1
+    try {
+      const reclaimed = await reclaimStuckIntents(env, leaseSeconds * 2);
+      if (reclaimed > 0) log('intents_reclaimed', { count: reclaimed });
+    } catch (err) {
+      log('reclaim_error', { error: err instanceof Error ? err.message : String(err) });
+    }
+
     // Claim and dispatch one intent.
     let intent: Intent | null;
     try {
@@ -199,18 +223,80 @@ async function innerLoop(
 
     log('intent_claimed', { intentId: intent.id, intentType: intent.intentType });
 
+    // Background heartbeat ticker covering the executor.execute() span.
+    // Uses the current session token so the heartbeat is rejected if a newer
+    // leader has taken over.
+    // fixes codex-p2 PR#101 finding-3, finding-5
+    const executorHeartbeat = setInterval(() => {
+      heartbeat(env, options.nodeId, {
+        role,
+        leaseSeconds,
+        sessionId: options.sessionId,
+      })
+        .then((renewed) => {
+          if (renewed) {
+            lastHeartbeat = Date.now();
+            log('exec_heartbeat_ok', { expiresAt: renewed.leaseExpiresAt });
+          } else {
+            log('exec_heartbeat_lost', { intentId: intent!.id });
+          }
+        })
+        .catch((err) => {
+          log('exec_heartbeat_error', {
+            error: err instanceof Error ? err.message : String(err),
+          });
+        });
+    }, innerHeartbeatMs);
+
+    // fixes codex-p2 PR#103 P1-B — capture the dispatched_task_id from
+    // markIntentDispatched as an execution token. completeIntent / failIntent
+    // gate on it so a stale leader returning from executor() after a fresher
+    // leader has reclaimed + redispatched the intent cannot mark the fresher
+    // execution done / failed. The token is set on dispatch and cleared by
+    // reclaimStuckIntents, so it's monotonic-per-execution.
+    let dispatchedTaskId: string | null = null;
     try {
       const result = await options.executor(intent);
-      await markIntentDispatched(env, intent.id, result.dispatchedTaskId);
-      await completeIntent(env, intent.id);
-      intentsProcessed += 1;
-      log('intent_completed', { intentId: intent.id, dispatchedTaskId: result.dispatchedTaskId });
+      const dispatched = await markIntentDispatched(env, intent.id, result.dispatchedTaskId);
+      if (!dispatched) {
+        // Status was no longer 'claimed' — another leader reclaimed and is
+        // (re)driving this intent. Do not touch completion.
+        log('intent_dispatch_lost', {
+          intentId: intent.id,
+          dispatchedTaskId: result.dispatchedTaskId,
+        });
+      } else {
+        dispatchedTaskId = result.dispatchedTaskId;
+        const completed = await completeIntent(env, intent.id, dispatchedTaskId);
+        if (!completed) {
+          log('intent_completion_ignored_stale', {
+            intentId: intent.id,
+            dispatchedTaskId,
+          });
+        } else {
+          intentsProcessed += 1;
+          log('intent_completed', {
+            intentId: intent.id,
+            dispatchedTaskId,
+          });
+        }
+      }
     } catch (err) {
       const msg = err instanceof Error ? err.message : String(err);
-      await failIntent(env, intent.id, msg).catch(() => {
-        /* surface only the original error */
-      });
-      log('intent_failed', { intentId: intent.id, error: msg });
+      // If we already captured a dispatch token, gate failure on it so we
+      // can't fail a fresher leader's running execution. Otherwise we failed
+      // before dispatch (intent still 'claimed') and the legacy unguarded
+      // status-only WHERE applies.
+      const failed = dispatchedTaskId
+        ? await failIntent(env, intent.id, msg, dispatchedTaskId).catch(() => null)
+        : await failIntent(env, intent.id, msg).catch(() => null);
+      if (!failed) {
+        log('intent_failure_ignored_stale', { intentId: intent.id, error: msg });
+      } else {
+        log('intent_failed', { intentId: intent.id, error: msg });
+      }
+    } finally {
+      clearInterval(executorHeartbeat);
     }
 
     if (options.maxIntents && intentsProcessed >= options.maxIntents) {

meta/context.ts

@@ -13,13 +13,32 @@
 
 export interface ContextEnv {
   CHITTYCONNECT_URL?: string;
+  /**
+   * Canonical ChittyConnect bearer token binding used elsewhere in this
+   * worker (see src/lib/cron.ts, src/routes/bridge/*). Preferred name.
+   * fixes codex-p2 PR#101 finding-8
+   */
+  CHITTY_CONNECT_TOKEN?: string;
+  /** Legacy name kept for backwards compatibility with earlier ContextEnv shape. */
   CHITTYCONNECT_TOKEN?: string;
   /** Optional service binding for ChittyConnect, used when primary fails. */
   AGENT_CONNECT?: Fetcher;
   /** Caller identity; defaults to "chittycommand-meta". */
   SERVICE_NAME?: string;
 }
 
+/**
+ * Resolve the ChittyConnect bearer token, preferring the canonical
+ * CHITTY_CONNECT_TOKEN binding used by the rest of the worker
+ * (src/middleware/auth.ts, src/lib/cron.ts, src/routes/bridge/*) and
+ * falling back to the legacy CHITTYCONNECT_TOKEN name.
+ *
+ * fixes codex-p2 PR#101 finding-8
+ */
+function resolveConnectToken(env: ContextEnv): string | undefined {
+  return env.CHITTY_CONNECT_TOKEN ?? env.CHITTYCONNECT_TOKEN;
+}
+
 export type ContextPath = 'primary' | 'fallback';
 
 export interface ContextResult<T = unknown> {
@@ -59,16 +78,17 @@ async function request<T>(
   body?: unknown,
   timeoutMs = 8000,
 ): Promise<{ ok: boolean; data?: T; error?: string; status?: number }> {
-  if (!env.CHITTYCONNECT_URL || !env.CHITTYCONNECT_TOKEN) {
-    return { ok: false, error: 'CHITTYCONNECT_URL or CHITTYCONNECT_TOKEN not set' };
+  const token = resolveConnectToken(env);
+  if (!env.CHITTYCONNECT_URL || !token) {
+    return { ok: false, error: 'CHITTYCONNECT_URL or CHITTY_CONNECT_TOKEN not set' };
   }
   const url = `${env.CHITTYCONNECT_URL.replace(/\/$/, '')}${path}`;
   try {
     const res = await fetch(url, {
       method,
       headers: {
         'Content-Type': 'application/json',
-        Authorization: `Bearer ${env.CHITTYCONNECT_TOKEN}`,
+        Authorization: `Bearer ${token}`,
         'X-ChittyOS-Caller': env.SERVICE_NAME ?? CALLER_DEFAULT,
       },
       body: body ? JSON.stringify(body) : undefined,
@@ -128,7 +148,21 @@ export async function getEcosystemAwareness(env: ContextEnv): Promise<EcosystemA
     5000,
   );
   if (primary.ok && primary.data) return primary.data;
-  return { success: false, error: primary.error ?? 'Awareness check failed' };
+
+  // fixes codex-p2 PR#101 finding-7 — when the HTTPS path has no URL/token or
+  // the upstream fetch failed, route the same request through the AGENT_CONNECT
+  // service binding if the worker has one. Matches the persist/recall pattern.
+  const fb = await fallback<EcosystemAwareness>(
+    env,
+    'GET',
+    '/api/intelligence/consciousness/awareness',
+  );
+  if (fb.ok && fb.data) return fb.data;
+
+  return {
+    success: false,
+    error: primary.error ?? fb.error ?? 'Awareness check failed',
+  };
 }
 
 // ── MemoryCloude: Persist ───────────────────────────────────