Merge canonical for TASK-pilot-cli-suite-timeout

src/adapters/codex.mjs

@@ -31,15 +31,28 @@ function commandRecord(command, args, cwd) {
   return { command, args, cwd, mutatesRuntime: true };
 }
 
-function mcpArgs(mcp) {
+function mcpRuntimeEnvironmentNames(env, platform) {
+  const runtimeDirectory = env?.XDG_RUNTIME_DIR;
+  if (platform !== 'linux' || typeof runtimeDirectory !== 'string'
+    || !/^\/run\/user\/[1-9]\d*$/.test(runtimeDirectory)) {
+    return [];
+  }
+  return ['XDG_RUNTIME_DIR'];
+}
+
+function mcpArgs(mcp, { env, platform } = {}) {
   if (!mcp) return [];
   if (typeof mcp.name !== 'string' || !/^[a-z0-9_-]+$/.test(mcp.name)
     || typeof mcp.command !== 'string' || !Array.isArray(mcp.args)) {
     throw new TorchError('Codex MCP launch configuration is invalid', { code: 'INVALID_RUNTIME_INPUT' });
   }
+  const environmentNames = mcpRuntimeEnvironmentNames(env, platform);
   return [
     '-c', `mcp_servers.${mcp.name}.command=${JSON.stringify(mcp.command)}`,
     '-c', `mcp_servers.${mcp.name}.args=${JSON.stringify(mcp.args)}`,
+    ...(environmentNames.length
+      ? ['-c', `mcp_servers.${mcp.name}.env_vars=${JSON.stringify(environmentNames)}`]
+      : []),
   ];
 }
 
@@ -169,12 +182,14 @@ export function diagnoseCodexStartupFailure({ stdout, stderr } = {}) {
 export class CodexRuntimeAdapter {
   constructor({
     env = process.env,
+    platform = process.platform,
     runner = null,
     executable = 'codex',
     nodeExecutable = process.execPath,
   } = {}) {
     this.name = 'codex';
     this.env = env;
+    this.platform = platform;
     this.runner = runner;
     this.executable = executable;
     this.nodeExecutable = nodeExecutable;
@@ -241,7 +256,7 @@ export class CodexRuntimeAdapter {
     const cwd = text(worktree, 'worktree');
     const args = [
       '--cd', cwd, ...launchPolicyArgs(launchPolicy ?? this.configuration.launchPolicy),
-      ...mcpArgs(mcp),
+      ...mcpArgs(mcp, { env: this.env, platform: this.platform }),
       ...((reasoning ?? this.configuration.reasoning) ? ['-c', `model_reasoning_effort=${JSON.stringify(text(reasoning ?? this.configuration.reasoning, 'reasoning'))}`] : []),
       'exec', '--json',
       ...optionalModelArgs(model ?? this.configuration.model),
@@ -266,7 +281,7 @@ export class CodexRuntimeAdapter {
       : '';
     const args = [
       '--cd', cwd, ...launchPolicyArgs(launchPolicy ?? this.configuration.launchPolicy),
-      ...mcpArgs(mcp),
+      ...mcpArgs(mcp, { env: this.env, platform: this.platform }),
       ...((reasoning ?? this.configuration.reasoning) ? ['-c', `model_reasoning_effort=${JSON.stringify(text(reasoning ?? this.configuration.reasoning, 'reasoning'))}`] : []),
       'exec', 'resume', '--json',
       ...optionalModelArgs(model ?? this.configuration.model), sessionId, `${text(message, 'message')}${currentInstructions}`,

src/cli.mjs

@@ -14,7 +14,7 @@ import { startConsole } from './console/server.mjs';
 import { classifyRecoverability, createLocalCanonical, planLocalCanonical } from './canonical/local.mjs';
 import { openControlPlane, openControlPlaneReadOnly } from './control-plane/service.mjs';
 import { fetchCanonicalObjects, planCanonicalFetch } from './canonical/fetch.mjs';
-import { spawnSync } from 'node:child_process';
+import { spawn as spawnProcess, spawnSync } from 'node:child_process';
 import { randomUUID } from 'node:crypto';
 import {
   mkdirSync, mkdtempSync, renameSync, rmSync, statSync, writeFileSync,
@@ -70,6 +70,8 @@ import {
 } from './design/architect.mjs';
 import { FleetEvolutionService } from './evolution/service.mjs';
 import { HierarchyEvolutionService } from './evolution/hierarchy.mjs';
+import { issueStoppedExecutorEvidence } from './runtime/stopped-executor-evidence.mjs';
+import { configureContinuation, planContinuation, runContinuation } from './runtime/continuation.mjs';
 import { ConvergenceService } from './convergence/service.mjs';
 import {
   attachForge, detachForge, forgeStatus, planForgeAttach, planForgeDetach,
@@ -95,6 +97,8 @@ Usage:
   torch install --restore [--dry-run] [--yes] [--json]
   torch setup [--parent <path>] [--dry-run] [--yes] [--json]
   torch up [--fresh] [--only <id,id>] [--dry-run] [--yes] [--json]
+  torch recover-runtime --area <id> --unit <service> --invocation <id> [--expected-session <id> --expected-updated-at <time> --yes] [--json]
+  torch continuation <status|enable|pause|tick> [--max-turns-per-day <1-48>] [--today-max-turns <1-96>] [--concurrency <1-7>] [--limit <1-7>] [--yes] [--json]
   torch down [--dry-run] [--yes] [--json]
   torch capture [--json]
   torch detach [--dry-run] [--yes] [--json]
@@ -585,6 +589,56 @@ export async function runCli(argv = process.argv.slice(2), {
       process.stdout.write(HELP);
       return 0;
     }
+    if (command === 'continuation') {
+      const control = openControlPlane({ repositoryRoot: optionValue(argv, '--repo') ?? cwd, env });
+      try {
+        const operation = argv[1] ?? 'status';
+        const backlog = new BacklogService({ repositoryRoot: control.repositoryRoot, controlPlane: control });
+        if (operation === 'status' || !argv.includes('--yes') || argv.includes('--dry-run')) {
+          print(planContinuation(control, backlog), { json }); return 0;
+        }
+        if (operation === 'enable' || operation === 'pause') {
+          print(configureContinuation(control, { actorId: 'owner', enabled: operation === 'enable',
+            maxTurnsPerDay: optionValue(argv, '--max-turns-per-day') === undefined ? undefined : Number(optionValue(argv, '--max-turns-per-day')),
+            todayMaxTurns: optionValue(argv, '--today-max-turns') === undefined ? undefined : Number(optionValue(argv, '--today-max-turns')),
+            maxConcurrency: optionValue(argv, '--concurrency') === undefined ? undefined : Number(optionValue(argv, '--concurrency')) }), { json }); return 0;
+        }
+        if (operation !== 'tick') throw new TorchError('Unknown continuation operation.', { code: 'UNKNOWN_COMMAND' });
+        print(await runContinuation(control, backlog, { actorId: 'owner', limit: optionValue(argv, '--limit') === undefined ? undefined : Number(optionValue(argv, '--limit')),
+          launch: areaId => new Promise(resolveTurn => {
+            const child = spawnProcess(process.execPath, [fileURLToPath(new URL('../bin/torch.mjs', import.meta.url)),
+              'up', '--only', areaId, '--yes', '--json'], { cwd: control.repositoryRoot, env,
+              stdio: ['ignore', 'pipe', 'pipe'] });
+            child.stdout.resume(); child.stderr.resume();
+            let failed = false;
+            child.on('error', () => { failed = true; });
+            child.on('close', (code, signal) => resolveTurn(failed || signal ? null : code));
+          }) }), { json });
+        return 0;
+      } finally { control.close(); }
+    }
+    if (command === 'recover-runtime') {
+      const control = openControlPlane({ repositoryRoot: optionValue(argv, '--repo') ?? cwd, env });
+      try {
+        const areaId = optionValue(argv, '--area');
+        const issued = issueStoppedExecutorEvidence({ controlPlane: control, actorId: 'owner', areaId,
+          unit: optionValue(argv, '--unit'), invocationId: optionValue(argv, '--invocation') });
+        if (!argv.includes('--yes') || argv.includes('--dry-run')) {
+          print({ ...issued.plan, requiresApproval: true }, { json });
+          return 0;
+        }
+        const expected = issued.plan.expectedIdentity;
+        if (optionValue(argv, '--expected-session') !== expected.runtimeSessionId
+          || optionValue(argv, '--expected-updated-at') !== expected.updatedAt) {
+          throw new TorchError('Apply requires the exact session and updated timestamp from the recovery preview.', {
+            code: 'RUNTIME_RECOVERY_SNAPSHOT_CONFLICT',
+          });
+        }
+        print(control.recoverOwnerStoppedUnknownIdentity({ actorId: 'owner', areaId,
+          expectedIdentity: expected, evidence: issued.evidence }), { json });
+        return 0;
+      } finally { control.close(); }
+    }
     if (command === 'runtimes') {
       const operation = argv[1] ?? 'list';
       if (operation === 'update') {

src/control-plane/service.mjs

@@ -5,8 +5,9 @@ import { DatabaseSync } from 'node:sqlite';
 import { TorchError } from '../kernel/errors.mjs';
 import { readInstallManifest } from '../kernel/install.mjs';
 import { projectStatePath } from '../kernel/paths.mjs';
-import { loadProjectConfig } from '../kernel/config.mjs';
+import { loadProjectConfig, validateProjectConfig } from '../kernel/config.mjs';
 import { organizationGraphFromConfig } from '../kernel/organization.mjs';
+import { consumeStoppedExecutorEvidence } from '../runtime/stopped-executor-evidence.mjs';
 
 export const PRESENCE_STATES = Object.freeze([
   'starting', 'working', 'waiting', 'idle', 'stale', 'stopping', 'offline',
@@ -14,9 +15,23 @@ export const PRESENCE_STATES = Object.freeze([
 
 const MESSAGE_KINDS = new Set([
   'message', 'coordination-request', 'handoff-request', 'completion', 'blocker', 'manager-check-in', 'manager-stale-work-review',
-  'approval-request', 'approval-decision',
+  'approval-request', 'approval-decision', 'owner-reply',
 ]);
 
+const CONTROL_PLANE_V2_SCHEMA_VERSION = 2;
+
+// This reader intentionally recognizes only the canonical v2 state shape. A
+// newer state must be rejected rather than interpreted with a partial guess.
+const CONTROL_PLANE_V2_COLUMNS = Object.freeze({
+  identities: ['area_id', 'runtime', 'runtime_session_id', 'state', 'summary', 'current_task', 'heartbeat_at', 'updated_at'],
+  messages: ['id', 'project_id', 'sender_id', 'recipient_id', 'kind', 'created_at', 'body', 'task_ref', 'path_ref', 'commit_ref', 'handoff_ref', 'delivered_at'],
+  message_acks: ['message_id', 'area_id', 'acknowledged_at'],
+  reports: ['id', 'project_id', 'area_id', 'kind', 'summary', 'task_ref', 'evidence', 'created_at'],
+  handoffs: ['id', 'project_id', 'sender_id', 'recipient_id', 'path_ref', 'task_ref', 'reason', 'status', 'created_at'],
+  audit_events: ['id', 'project_id', 'actor_id', 'operation', 'entity_type', 'entity_id', 'details', 'created_at'],
+  approval_requests: ['id', 'project_id', 'requester_id', 'approver_id', 'task_ref', 'title', 'summary', 'evidence', 'status', 'revision', 'created_at', 'decided_at', 'decided_by', 'decision_note'],
+});
+
 function readJson(path, code) {
   try {
     return JSON.parse(readFileSync(path, 'utf8'));
@@ -25,6 +40,40 @@ function readJson(path, code) {
   }
 }
 
+function readJsonBytes(bytes, path, code) {
+  try {
+    return JSON.parse(bytes.toString('utf8'));
+  } catch (error) {
+    throw new TorchError(`Cannot read TORCH state at ${path}`, { code, details: error.message });
+  }
+}
+
+function readOnlyUnavailable(message, details = null) {
+  return new TorchError(message, { code: 'CONTROL_PLANE_READ_ONLY_UNAVAILABLE', details });
+}
+
+function exactColumns(database, table) {
+  return database.prepare(`PRAGMA table_info(${table})`).all().map((column) => column.name);
+}
+
+function validateV2ReadOnlySchema(database) {
+  const version = database.prepare('PRAGMA user_version').get()?.user_version;
+  if (version !== CONTROL_PLANE_V2_SCHEMA_VERSION) {
+    throw readOnlyUnavailable('The control-plane state is not the supported schema-v2 snapshot.', {
+      expected: CONTROL_PLANE_V2_SCHEMA_VERSION, actual: version ?? null,
+    });
+  }
+  for (const [table, expected] of Object.entries(CONTROL_PLANE_V2_COLUMNS)) {
+    const present = database.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?").get(table);
+    const actual = present ? exactColumns(database, table) : null;
+    if (!actual || actual.length !== expected.length || actual.some((name, index) => name !== expected[index])) {
+      throw readOnlyUnavailable('The control-plane state does not match the required schema-v2 tables.', {
+        table, expected, actual,
+      });
+    }
+  }
+}
+
 function requiredText(value, name) {
   if (typeof value !== 'string' || !value.trim()) {
     throw new TorchError(`${name} must be a non-empty string`, {
@@ -39,6 +88,35 @@ function optionalText(value, name) {
   return requiredText(value, name);
 }
 
+function messageObservationSelection(value) {
+  if (value !== 'unread' && value !== 'history') {
+    throw new TorchError('Message observation selection must be unread or history', {
+      code: 'INVALID_MESSAGE_SELECTION', details: { selection: value, allowed: ['unread', 'history'] },
+    });
+  }
+  return value;
+}
+
+function messageObservationLimit(value) {
+  if (!Number.isInteger(value) || value < 1 || value > 1000) {
+    throw new TorchError('Message observation limit must be an integer from 1 through 1000', {
+      code: 'INVALID_MESSAGE_LIMIT', details: { limit: value, minimum: 1, maximum: 1000 },
+    });
+  }
+  return value;
+}
+
+function ownerConversationCursor(value) {
+  if (value === undefined || value === null) return null;
+  if (!value || typeof value !== 'object' || Array.isArray(value)
+    || Object.keys(value).length !== 2 || !Object.hasOwn(value, 'createdAt') || !Object.hasOwn(value, 'id')) {
+    throw new TorchError('Owner conversation cursor must contain only createdAt and id', {
+      code: 'INVALID_OWNER_CONVERSATION_CURSOR',
+    });
+  }
+  return { createdAt: requiredText(value.createdAt, 'cursor.createdAt'), id: requiredText(value.id, 'cursor.id') };
+}
+
 function normalizeReferences(references = {}) {
   return {
     task: optionalText(references.task, 'references.task'),
@@ -112,6 +190,13 @@ function effectiveRoster(roster, config) {
 }
 
 function rowToMessage(row) {
+  const references = {
+    task: row.task_ref ?? null,
+    path: row.path_ref ?? null,
+    commit: row.commit_ref ?? null,
+    handoff: row.handoff_ref ?? null,
+  };
+  if (row.kind === 'owner-reply') references.replyTo = row.handoff_ref ?? null;
   return {
     id: row.id,
     projectId: row.project_id,
@@ -122,12 +207,7 @@ function rowToMessage(row) {
     body: row.body,
     acknowledgedAt: row.acknowledged_at ?? null,
     deliveredAt: row.delivered_at ?? null,
-    references: {
-      task: row.task_ref ?? null,
-      path: row.path_ref ?? null,
-      commit: row.commit_ref ?? null,
-      handoff: row.handoff_ref ?? null,
-    },
+    references,
   };
 }
 
@@ -224,10 +304,15 @@ function initializeSchema(database) {
 }
 
 export class ControlPlane {
-  constructor({ repositoryRoot, env = process.env, clock = () => new Date(), idFactory = randomUUID } = {}) {
+  constructor({ repositoryRoot, env = process.env, clock = () => new Date(), idFactory = randomUUID, readOnly = false } = {}) {
     this.repositoryRoot = resolve(repositoryRoot ?? process.cwd());
     this.clock = clock;
     this.idFactory = idFactory;
+    this.readOnly = readOnly === true;
+    if (this.readOnly) {
+      this.initializeReadOnly(env);
+      return;
+    }
     this.manifest = readInstallManifest(this.repositoryRoot);
     this.config = loadProjectConfig(this.repositoryRoot);
     this.roster = readJson(join(this.repositoryRoot, '.torch', 'roster.yaml'), 'INVALID_TORCH_ROSTER');
@@ -267,7 +352,104 @@ export class ControlPlane {
     for (const areaId of this.agents.keys()) insert.run(areaId, 'offline');
   }
 
+  initializeReadOnly(env) {
+    const trackedRoot = join(this.repositoryRoot, '.torch');
+    const manifestPath = join(trackedRoot, 'install-manifest.json');
+    const configPath = join(trackedRoot, 'torch.yaml');
+    const rosterPath = join(trackedRoot, 'roster.yaml');
+    try {
+      this.readOnlyInputs = [
+        { path: manifestPath, bytes: readFileSync(manifestPath) },
+        { path: configPath, bytes: readFileSync(configPath) },
+        { path: rosterPath, bytes: readFileSync(rosterPath) },
+      ];
+      this.manifest = readJsonBytes(this.readOnlyInputs[0].bytes, manifestPath, 'INVALID_INSTALL_MANIFEST');
+      this.config = validateProjectConfig(readJsonBytes(this.readOnlyInputs[1].bytes, configPath, 'CONFIG_INVALID'));
+      this.roster = readJsonBytes(this.readOnlyInputs[2].bytes, rosterPath, 'INVALID_TORCH_ROSTER');
+    } catch (error) {
+      if (error instanceof TorchError) throw error;
+      throw readOnlyUnavailable('The tracked control-plane inputs could not be read atomically.', { cause: error.message });
+    }
+    if (this.config.project?.id !== this.manifest.projectId) {
+      throw new TorchError('Tracked configuration and installation manifest disagree on project identity', {
+        code: 'PROJECT_ID_MISMATCH',
+      });
+    }
+    const localState = (this.manifest.external ?? []).find((entry) => entry.type === 'local-state');
+    if (!localState?.path || !existsSync(localState.path)) {
+      throw new TorchError('TORCH local project state is unavailable', { code: 'LOCAL_STATE_MISSING' });
+    }
+    const expectedStateRoot = projectStatePath(this.manifest.projectId, env);
+    if (resolve(localState.path) !== resolve(expectedStateRoot)) {
+      throw new TorchError('Installation manifest local state path does not match this environment', {
+        code: 'LOCAL_STATE_PATH_MISMATCH',
+        details: { manifest: localState.path, expected: expectedStateRoot },
+      });
+    }
+    const metadataPath = join(expectedStateRoot, 'project.json');
+    try {
+      this.readOnlyInputs.push({ path: metadataPath, bytes: readFileSync(metadataPath) });
+    } catch (error) {
+      throw readOnlyUnavailable('The local control-plane metadata could not be read.', { cause: error.message });
+    }
+    const metadata = readJsonBytes(this.readOnlyInputs[3].bytes, metadataPath, 'LOCAL_STATE_METADATA_INVALID');
+    if (metadata.projectId !== this.manifest.projectId || resolve(metadata.root) !== this.repositoryRoot) {
+      throw new TorchError('Local state metadata does not belong to this installation root', {
+        code: 'LOCAL_STATE_METADATA_MISMATCH',
+      });
+    }
+    this.stateRoot = expectedStateRoot;
+    this.projectId = this.manifest.projectId;
+    this.agents = new Map(effectiveRoster(this.roster, this.config).map((area) => [area.id, area]));
+    if (!this.agents.has('session-manager')) {
+      throw new TorchError('Roster does not contain the required session-manager identity', {
+        code: 'INVALID_TORCH_ROSTER',
+      });
+    }
+    try {
+      this.database = new DatabaseSync(join(this.stateRoot, 'state.db'), { readOnly: true });
+      // A normal read transaction preserves SQLite's locking and WAL semantics.
+      this.database.exec('BEGIN');
+      this.readTransactionOpen = true;
+      validateV2ReadOnlySchema(this.database);
+      for (const areaId of this.agents.keys()) {
+        if (!this.database.prepare('SELECT area_id FROM identities WHERE area_id = ?').get(areaId)) {
+          throw readOnlyUnavailable('The schema-v2 snapshot does not contain every registered identity.', { areaId });
+        }
+      }
+      this.assertReadOnlyInputsStable();
+    } catch (error) {
+      try { if (this.readTransactionOpen) this.database.exec('ROLLBACK'); } catch { /* close only */ }
+      try { this.database?.close(); } catch { /* close only */ }
+      if (error instanceof TorchError) throw error;
+      throw readOnlyUnavailable('A coherent read-only control-plane snapshot is unavailable.', { cause: error.message });
+    }
+  }
+
+  assertReadOnlyInputsStable() {
+    if (!this.readOnly) return;
+    for (const input of this.readOnlyInputs) {
+      let current;
+      try { current = readFileSync(input.path); } catch (error) {
+        throw readOnlyUnavailable('A control-plane input disappeared while it was being read.', { path: input.path, cause: error.message });
+      }
+      if (!current.equals(input.bytes)) {
+        throw readOnlyUnavailable('A control-plane input changed while the read-only snapshot was open.', { path: input.path });
+      }
+    }
+  }
+
+  assertWritable() {
+    if (this.readOnly) {
+      throw new TorchError('This control-plane handle is read-only.', { code: 'CONTROL_PLANE_READ_ONLY' });
+    }
+  }
+
   refreshRoster() {
+    if (this.readOnly) {
+      this.assertReadOnlyInputsStable();
+      return { schema: this.roster.schema, areaIds: [...this.agents.keys()] };
+    }
     const roster = readJson(join(this.repositoryRoot, '.torch', 'roster.yaml'), 'INVALID_TORCH_ROSTER');
     const config = loadProjectConfig(this.repositoryRoot);
     const agents = new Map(effectiveRoster(roster, config).map((area) => [area.id, area]));
@@ -284,11 +466,18 @@ export class ControlPlane {
   }
 
   close() {
+    if (this.readTransactionOpen) {
+      try { this.database.exec('COMMIT'); } finally { this.readTransactionOpen = false; }
+    }
     this.database.close();
   }
 
   assertIdentity(areaId) {
     this.refreshRoster();
+    return this.assertLoadedIdentity(areaId);
+  }
+
+  assertLoadedIdentity(areaId) {
     const normalized = requiredText(areaId, 'areaId');
     if (!this.agents.has(normalized)) {
       throw new TorchError(`Unknown Fleet identity: ${normalized}`, {
@@ -307,6 +496,9 @@ export class ControlPlane {
   identity(areaId) {
     const id = this.assertIdentity(areaId);
     const row = this.database.prepare('SELECT * FROM identities WHERE area_id = ?').get(id);
+    if (this.readOnly && !row) {
+      throw readOnlyUnavailable('The read-only snapshot is missing a registered identity.', { areaId: id });
+    }
     return rowToAgent(row, this.agents.get(id));
   }
 
@@ -366,6 +558,7 @@ export class ControlPlane {
   }
 
   sendMessage({ sender, recipient, body, kind = 'message', references = {} } = {}) {
+    this.assertWritable();
     const senderId = this.assertIdentity(sender);
     const recipientId = this.assertRecipient(recipient);
     const normalizedKind = requiredText(kind, 'kind');
@@ -395,6 +588,7 @@ export class ControlPlane {
   }
 
   sendOwnerRequest({ actorId, recipient, body, references = {} } = {}) {
+    this.assertWritable();
     const ownerId = this.assertOwnerActor(actorId);
     const recipientId = this.assertIdentity(recipient);
     const normalizedBody = requiredText(body, 'body');
@@ -425,9 +619,43 @@ export class ControlPlane {
     return record;
   }
 
+  sendOwnerReply({ sender, requestId, body } = {}) {
+    this.assertWritable();
+    const senderId = this.assertIdentity(sender);
+    const parentId = requiredText(requestId, 'requestId');
+    const parent = this.database.prepare(`
+      SELECT * FROM messages
+      WHERE id = ? AND project_id = ? AND sender_id = 'owner' AND recipient_id = ? AND kind = 'message'
+    `).get(parentId, this.projectId, senderId);
+    if (!parent) {
+      throw new TorchError('Owner reply must reference an owner request addressed to the sender', {
+        code: 'OWNER_REPLY_REFERENCE_INVALID', details: { requestId: parentId, sender: senderId },
+      });
+    }
+    const normalizedBody = requiredText(body, 'body');
+    const record = {
+      id: this.idFactory(), projectId: this.projectId, sender: senderId, recipient: 'owner',
+      kind: 'owner-reply', createdAt: this.clock().toISOString(), body: normalizedBody,
+      acknowledgedAt: null, deliveredAt: null,
+      references: { task: null, path: null, commit: null, handoff: parentId, replyTo: parentId },
+    };
+    this.database.prepare(`
+      INSERT INTO messages (
+        id, project_id, sender_id, recipient_id, kind, created_at, body,
+        task_ref, path_ref, commit_ref, handoff_ref
+      ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    `).run(
+      record.id, record.projectId, record.sender, record.recipient, record.kind,
+      record.createdAt, record.body, null, null, null, parentId,
+    );
+    this.audit({ actorId: senderId, operation: 'message.reply-owner', entityType: 'message', entityId: record.id,
+      details: { requestId: parentId } });
+    return record;
+  }
+
   readMessages({ recipient, unacknowledgedOnly = false, limit = 100 } = {}) {
     const recipientId = this.assertIdentity(recipient);
-    const normalizedLimit = Number.isInteger(limit) && limit > 0 && limit <= 1000 ? limit : 100;
+    const normalizedLimit = messageObservationLimit(limit);
     const query = `
       SELECT m.*, a.acknowledged_at
       FROM messages m
@@ -440,7 +668,96 @@ export class ControlPlane {
     return this.database.prepare(query).all(recipientId, recipientId, normalizedLimit).map(rowToMessage);
   }
 
+  observeMessages({ recipient, selection, limit } = {}) {
+    // Unlike legacy readMessages, this observation must not refresh the roster:
+    // refreshRoster maintains identity rows and is not a read-only observation.
+    const recipientId = this.assertLoadedIdentity(recipient);
+    const normalizedSelection = messageObservationSelection(selection);
+    const normalizedLimit = messageObservationLimit(limit);
+    const rows = this.database.prepare(`
+      WITH visible_messages AS MATERIALIZED (
+        SELECT m.*, a.acknowledged_at
+        FROM messages m
+        LEFT JOIN message_acks a ON a.message_id = m.id AND a.area_id = ?
+        WHERE m.recipient_id = ? OR m.recipient_id = 'all'
+      ),
+      selected_messages AS MATERIALIZED (
+        SELECT * FROM visible_messages
+        WHERE ? = 'history' OR acknowledged_at IS NULL
+      ),
+      page AS MATERIALIZED (
+        SELECT * FROM selected_messages
+        ORDER BY created_at ASC, id ASC
+        LIMIT ?
+      ),
+      metadata AS (
+        SELECT
+          (SELECT COUNT(*) FROM selected_messages) AS selected_count,
+          (SELECT COUNT(*) FROM visible_messages WHERE acknowledged_at IS NULL) AS pending_unread_count
+      )
+      SELECT page.*, metadata.selected_count, metadata.pending_unread_count
+      FROM metadata
+      LEFT JOIN page ON TRUE
+      ORDER BY page.created_at ASC, page.id ASC
+    `).all(recipientId, recipientId, normalizedSelection, normalizedLimit);
+    const messages = rows.filter((row) => row.id !== null).map(rowToMessage);
+    const selectedCount = Number(rows[0]?.selected_count ?? 0);
+    const complete = selectedCount <= normalizedLimit;
+    return {
+      recipient: recipientId,
+      messages,
+      observation: {
+        selection: normalizedSelection,
+        requestedLimit: normalizedLimit,
+        returnedCount: messages.length,
+        complete,
+        truncated: !complete,
+        pendingUnreadCount: Number(rows[0]?.pending_unread_count ?? 0),
+      },
+    };
+  }
+
+  readOwnerConversation({ actorId, requestId, cursor, limit = 100 } = {}) {
+    this.assertOwnerActor(actorId);
+    const parentId = requiredText(requestId, 'requestId');
+    const pageCursor = ownerConversationCursor(cursor);
+    const normalizedLimit = messageObservationLimit(limit);
+    const request = this.database.prepare(`
+      SELECT * FROM messages
+      WHERE id = ? AND project_id = ? AND sender_id = 'owner' AND kind = 'message'
+    `).get(parentId, this.projectId);
+    if (!request) {
+      throw new TorchError('Owner conversation request is unavailable in this project', {
+        code: 'OWNER_CONVERSATION_NOT_FOUND', details: { requestId: parentId },
+      });
+    }
+    const rows = this.database.prepare(`
+      SELECT * FROM messages
+      WHERE project_id = ? AND recipient_id = 'owner' AND kind = 'owner-reply' AND handoff_ref = ?
+        AND (? IS NULL OR created_at > ? OR (created_at = ? AND id > ?))
+      ORDER BY created_at ASC, id ASC
+      LIMIT ?
+    `).all(
+      this.projectId, parentId,
+      pageCursor?.createdAt ?? null, pageCursor?.createdAt ?? null,
+      pageCursor?.createdAt ?? null, pageCursor?.id ?? null,
+      normalizedLimit + 1,
+    );
+    const hasMore = rows.length > normalizedLimit;
+    const replies = rows.slice(0, normalizedLimit).map(rowToMessage);
+    const last = replies.at(-1);
+    return {
+      request: rowToMessage(request), replies,
+      page: {
+        requestedLimit: normalizedLimit, returnedCount: replies.length,
+        complete: !hasMore, truncated: hasMore,
+        nextCursor: hasMore && last ? { createdAt: last.createdAt, id: last.id } : null,
+      },
+    };
+  }
+
   ackMessage({ recipient, messageId } = {}) {
+    this.assertWritable();
     const recipientId = this.assertIdentity(recipient);
     const id = requiredText(messageId, 'messageId');
     const message = this.database.prepare('SELECT * FROM messages WHERE id = ?').get(id);
@@ -459,7 +776,39 @@ export class ControlPlane {
     return { messageId: id, recipient: recipientId, acknowledgedAt };
   }
 
+  recoverOwnerStoppedUnknownIdentity({ actorId, areaId, expectedIdentity, evidence } = {}) {
+    this.assertWritable();
+    const owner = this.assertOwnerActor(actorId);
+    const id = this.assertIdentity(areaId);
+    if (!expectedIdentity || ['state', 'runtime', 'runtimeSessionId', 'updatedAt'].some(key => !expectedIdentity[key])) {
+      throw new TorchError('Recovery requires the complete previewed identity snapshot.', { code: 'RUNTIME_RECOVERY_SNAPSHOT_REQUIRED' });
+    }
+    this.database.exec('BEGIN IMMEDIATE');
+    try {
+      const current = this.identity(id);
+      if (current.state !== 'working' || !current.summary?.includes('Runtime executor completion is unknown')
+        || ['state', 'runtime', 'runtimeSessionId', 'updatedAt'].some(key => current[key] !== expectedIdentity[key])) {
+        throw new TorchError('Identity changed since recovery was reviewed.', { code: 'RUNTIME_RECOVERY_SNAPSHOT_CONFLICT' });
+      }
+      const facts = consumeStoppedExecutorEvidence(evidence, { projectId: this.projectId, repositoryRoot: this.repositoryRoot, identity: current });
+      const now = this.clock().toISOString();
+      const result = this.database.prepare(`UPDATE identities SET state = 'offline', summary = ?, heartbeat_at = ?, updated_at = ?
+        WHERE area_id = ? AND state = ? AND runtime = ? AND runtime_session_id = ? AND updated_at = ?`)
+        .run('Owner verified stopped legacy executor; prior outcome remains unknown.', now, now,
+          id, current.state, current.runtime, current.runtimeSessionId, current.updatedAt);
+      if (result.changes !== 1) throw new TorchError('Recovery lost its identity snapshot.', { code: 'RUNTIME_RECOVERY_SNAPSHOT_CONFLICT' });
+      const audit = this.auditOwnerAction({ actorId: owner, operation: 'runtime.legacy-unknown.recover-stopped',
+        entityType: 'identity', entityId: id, details: { previousIdentity: expectedIdentity, ...facts } });
+      this.database.exec('COMMIT');
+      return { identity: this.identity(id), auditId: audit.id, priorOutcome: 'unknown', sessionsStarted: false, mutationPerformed: true };
+    } catch (error) {
+      this.database.exec('ROLLBACK');
+      throw error;
+    }
+  }
+
   reportStatus({ areaId, state, summary, runtime, runtimeSessionId, task } = {}) {
+    this.assertWritable();
     const id = this.assertIdentity(areaId);
     const normalizedState = requiredText(state, 'state');
     if (!PRESENCE_STATES.includes(normalizedState)) {
@@ -486,6 +835,7 @@ export class ControlPlane {
   }
 
   writeReport({ areaId, kind, summary, task, evidence } = {}) {
+    this.assertWritable();
     const id = this.assertIdentity(areaId);
     const createdAt = this.clock().toISOString();
     const report = {
@@ -535,6 +885,7 @@ export class ControlPlane {
   }
 
   requestHandoff({ sender, recipient = 'session-manager', path, task, reason } = {}) {
+    this.assertWritable();
     const senderId = this.assertIdentity(sender);
     const recipientId = this.assertIdentity(recipient);
     const pathRef = path ? normalizedProjectPath(path) : null;
@@ -576,6 +927,7 @@ export class ControlPlane {
   }
 
   requestApproval({ requester, approver, task, title, summary, evidence } = {}) {
+    this.assertWritable();
     const requesterId = this.assertIdentity(requester);
     const approverId = requiredText(approver, 'approver');
     if (approverId === 'owner') this.assertOwnerActor(approverId);
@@ -651,6 +1003,7 @@ export class ControlPlane {
   }
 
   decideApproval({ approvalId, decidedBy, decision, note, expectedRevision } = {}) {
+    this.assertWritable();
     const id = requiredText(approvalId, 'approvalId');
     const actor = requiredText(decidedBy, 'decidedBy');
     const owner = actor === 'owner';
@@ -708,6 +1061,7 @@ export class ControlPlane {
   }
 
   insertOwnerApprovalDecisionMessage({ sender, recipient, approvalId, task, decision, note }) {
+    this.assertWritable();
     const id = this.idFactory();
     const createdAt = this.clock().toISOString();
     const body = `Approval ${decision}: ${note ?? 'No additional note.'}`;
@@ -732,6 +1086,7 @@ export class ControlPlane {
   }
 
   audit({ actorId, operation, entityType, entityId = null, details = null } = {}) {
+    this.assertWritable();
     const actor = this.assertIdentity(actorId);
     const event = {
       id: this.idFactory(), projectId: this.projectId, actorId: actor,
@@ -750,8 +1105,9 @@ export class ControlPlane {
   }
 
   auditOwnerAction({ actorId, operation, entityType, entityId = null, details = null } = {}) {
+    this.assertWritable();
     const actor = this.assertOwnerActor(actorId);
-    if (!['artifact.feedback', 'backlog.priority', 'backlog.create', 'agent.request', 'approval.decide', 'approval.notify', 'forge.sync', 'canonical.fetch', 'profile.change', 'schedule.launcher.install', 'schedule.launcher.reconcile', 'schedule.wake.recover', 'hierarchy.conclude-pilot', 'check.recover-prepared'].includes(operation)) {
+    if (!['artifact.feedback', 'backlog.priority', 'backlog.create', 'agent.request', 'approval.decide', 'approval.notify', 'forge.sync', 'canonical.fetch', 'profile.change', 'schedule.launcher.install', 'schedule.launcher.reconcile', 'schedule.wake.recover', 'hierarchy.conclude-pilot', 'check.recover-prepared', 'runtime.legacy-unknown.recover-stopped', 'runtime.continuation.configure'].includes(operation)) {
       throw new TorchError('This owner audit surface does not authorize the requested operation', {
         code: 'OWNER_OPERATION_UNSUPPORTED', details: { operation },
       });
@@ -803,3 +1159,7 @@ export class ControlPlane {
 export function openControlPlane(options) {
   return new ControlPlane(options);
 }
+
+export function openControlPlaneReadOnly(options) {
+  return new ControlPlane({ ...options, readOnly: true });
+}

src/integration/adoption.mjs

@@ -0,0 +1,384 @@
+import { createHash, randomUUID } from 'node:crypto';
+import { spawnSync } from 'node:child_process';
+import { resolve } from 'node:path';
+import { TorchError } from '../kernel/errors.mjs';
+import { observeRuntimeTurn } from '../runtime/turn-guard.mjs';
+
+const ADOPTION_SCHEMA = 'torch.dev/integration-adoption/v1alpha1';
+const AUTHORITY_SCHEMA = 'torch.dev/integration-adoption-authority/v1alpha1';
+const ADAPTER_NAME = 'git-private-ref-cas-v1';
+const ACTIONS = new Set(['plan', 'prepare', 'finalize', 'reconcile', 'rollback', 'request']);
+const MUTATING_ACTIONS = new Set(['prepare', 'finalize', 'rollback', 'request']);
+const SHA = /^[0-9a-f]{40,64}$/;
+const SAFE_ID = /^[a-zA-Z0-9][a-zA-Z0-9._-]{0,119}$/;
+const ZERO_SHA = '0'.repeat(40);
+
+function git(root, args, { allowFailure = false } = {}) {
+  const result = spawnSync('git', ['-C', root, ...args], { encoding: 'utf8' });
+  if (result.status !== 0 && !allowFailure) {
+    throw new TorchError(`Git command failed: git ${args.join(' ')}`, {
+      code: 'GIT_COMMAND_FAILED', details: result.stderr?.trim() || null,
+    });
+  }
+  return { status: result.status, stdout: result.stdout?.trim() ?? '', stderr: result.stderr?.trim() ?? '' };
+}
+
+function tableExists(database, name) {
+  return Boolean(database.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(name));
+}
+
+function text(value, name) {
+  if (typeof value !== 'string' || !value.trim()) {
+    throw new TorchError(`${name} is required`, { code: 'ADOPTION_INPUT_INVALID', details: { name } });
+  }
+  return value.trim();
+}
+
+function sha(value, name) {
+  const normalized = text(value, name);
+  if (!SHA.test(normalized)) throw new TorchError(`${name} must be a full Git object id`, {
+    code: 'ADOPTION_INPUT_INVALID', details: { name },
+  });
+  return normalized;
+}
+
+function digest(value) {
+  return createHash('sha256').update(JSON.stringify(value)).digest('hex');
+}
+
+function durableRecord(event) {
+  if (!event?.details) return null;
+  try {
+    const record = JSON.parse(event.details);
+    return record?.schema === ADOPTION_SCHEMA ? record : null;
+  } catch { return null; }
+}
+
+function listWorktreePaths(root) {
+  const records = [];
+  for (const line of git(root, ['worktree', 'list', '--porcelain']).stdout.split('\n')) {
+    if (line.startsWith('worktree ')) records.push(line.slice('worktree '.length));
+  }
+  return records;
+}
+
+function authorityFrom(approval) {
+  if (!approval || approval.status !== 'approved' || approval.approver_id !== 'owner' || approval.decided_by !== 'owner') return null;
+  try {
+    const authority = JSON.parse(approval.decision_note ?? '');
+    return authority?.schema === AUTHORITY_SCHEMA ? authority : null;
+  } catch { return null; }
+}
+
+function exactAuthority(authority, binding) {
+  if (!authority || authority.action !== binding.action) return false;
+  const fields = ['adoptionId', 'candidateSha', 'baseTargetSha', 'provenanceManifestDigest',
+    'contributorArea', 'destinationArea', 'targetBranch', 'candidateRef', 'expectedOldRef'];
+  return fields.every((field) => authority[field] === binding[field])
+    && authority.adapter?.name === binding.adapter?.name
+    && authority.adapter?.id === binding.adapter?.id;
+}
+
+function exactInput(record, input) {
+  const fields = ['adoptionId', 'candidateSha', 'baseTargetSha', 'provenanceManifestDigest',
+    'contributorArea', 'destinationArea', 'targetBranch', 'candidateRef', 'expectedOldRef'];
+  const mismatches = fields.filter((field) => input[field] !== undefined
+    && input[field] !== (field === 'adoptionId' ? record.id : record[field]));
+  if (input.provenance !== undefined && JSON.stringify(input.provenance) !== JSON.stringify(record.provenance)) mismatches.push('provenance');
+  if (mismatches.length) throw new TorchError('Lifecycle input cannot override immutable prepared adoption fields', {
+    code: 'ADOPTION_IMMUTABLE_BINDING_STALE', details: { adoptionId: record.id, fields: mismatches },
+  });
+  if (!Number.isInteger(input.expectedRecordRevision) || input.expectedRecordRevision !== record.revision) {
+    throw new TorchError('Lifecycle input must bind the current adoption record revision', {
+      code: 'ADOPTION_RECORD_REVISION_STALE',
+      details: { adoptionId: record.id, expectedRevision: input.expectedRecordRevision ?? null, actual: record.revision },
+    });
+  }
+}
+
+/**
+ * Real Git private-ref transaction adapter. It is source code only: callers
+ * decide when it is invoked, while the adapter itself gives temp-repository
+ * scenarios the same atomic update-ref semantics as a real repository.
+ */
+export class GitPrivateRefAdapter {
+  constructor({ repositoryRoot } = {}) {
+    this.repositoryRoot = resolve(text(repositoryRoot, 'repositoryRoot'));
+    this.name = ADAPTER_NAME;
+    this.id = digest({ adapter: this.name, repositoryRoot: this.repositoryRoot });
+  }
+
+  readRefs(refs) {
+    const values = {};
+    for (const ref of refs) {
+      const result = git(this.repositoryRoot, ['rev-parse', '--verify', '--quiet', ref], { allowFailure: true });
+      values[ref] = result.status === 0 ? result.stdout : null;
+    }
+    return values;
+  }
+
+  compareAndSwapRefs({ expected = [], writes = [] } = {}) {
+    const commands = ['start'];
+    const writtenRefs = new Set(writes.map((entry) => entry.ref));
+    for (const { ref, sha: expectedSha } of expected) {
+      if (writtenRefs.has(ref)) continue;
+      commands.push(`verify ${text(ref, 'ref')} ${expectedSha ?? ZERO_SHA}`);
+    }
+    for (const { ref, sha: nextSha } of writes) {
+      const expectedEntry = expected.find((entry) => entry.ref === ref);
+      if (!expectedEntry) throw new TorchError('Every private-ref write requires an exact expected value', {
+        code: 'ADOPTION_REF_CAS_EXPECTED_REQUIRED', details: { ref },
+      });
+      const current = expectedEntry.sha ?? null;
+      if (nextSha === null) commands.push(`delete ${text(ref, 'ref')} ${current ?? ZERO_SHA}`);
+      else commands.push(`update ${text(ref, 'ref')} ${sha(nextSha, 'next ref sha')} ${current ?? ZERO_SHA}`);
+    }
+    commands.push('prepare', 'commit', '');
+    const result = spawnSync('git', ['-C', this.repositoryRoot, 'update-ref', '--stdin'], {
+      input: commands.join('\n'), encoding: 'utf8',
+    });
+    if (result.status !== 0) throw new TorchError('Private ref compare-and-swap refused', {
+      code: 'ADOPTION_REF_CAS_FAILED', details: result.stderr?.trim() || null,
+    });
+    return { adapter: { name: this.name, id: this.id }, expected, writes };
+  }
+}
+
+export class AdoptionService {
+  constructor({ repositoryRoot, controlPlane, clock = () => new Date(), idFactory = randomUUID, policy, refAdapter } = {}) {
+    this.repositoryRoot = repositoryRoot;
+    this.controlPlane = controlPlane;
+    this.clock = clock;
+    this.idFactory = idFactory;
+    this.policy = policy ?? {};
+    this.refAdapter = refAdapter ?? new GitPrivateRefAdapter({ repositoryRoot });
+  }
+
+  plan({ action = 'plan', actorId, approvalId, approvalRevision, adoptionId,
+    candidateSha, baseTargetSha, provenance, provenanceManifestDigest, contributorArea,
+    destinationArea, targetBranch, candidateRef, expectedOldRef, adapter } = {}) {
+    const blockers = [];
+    const id = adoptionId ? text(adoptionId, 'adoptionId') : null;
+    if (id && !SAFE_ID.test(id)) blockers.push({ code: 'ADOPTION_ID_INVALID' });
+    if (!ACTIONS.has(action)) blockers.push({ code: 'ADOPTION_ACTION_INVALID' });
+    if (actorId !== 'owner') blockers.push({ code: 'OWNER_ACTOR_REQUIRED' });
+    if (!Number.isInteger(approvalRevision) || approvalRevision < 1) blockers.push({ code: 'ADOPTION_APPROVAL_REVISION_REQUIRED' });
+    const adapterBinding = adapter?.name && adapter?.id ? { name: adapter.name, id: adapter.id } : null;
+    if (MUTATING_ACTIONS.has(action) && !adapterBinding) blockers.push({ code: 'ADOPTION_ADAPTER_BINDING_REQUIRED' });
+    if (adapterBinding && (adapterBinding.name !== this.refAdapter.name || adapterBinding.id !== this.refAdapter.id)) {
+      blockers.push({ code: 'ADOPTION_ADAPTER_PROVENANCE_INVALID' });
+    }
+
+    let candidate; let base; let provenanceDigest;
+    try {
+      candidate = sha(candidateSha, 'candidateSha');
+      base = sha(baseTargetSha, 'baseTargetSha');
+      if (!Array.isArray(provenance) || !provenance.length || provenance.some((entry) => typeof entry !== 'string' || !entry)) {
+        throw new TorchError('provenance must be a non-empty ordered string list', { code: 'ADOPTION_INPUT_INVALID' });
+      }
+      provenanceDigest = digest(provenance);
+      if (sha(provenanceManifestDigest, 'provenanceManifestDigest') !== provenanceDigest) blockers.push({ code: 'ADOPTION_PROVENANCE_DIGEST_MISMATCH', expected: provenanceDigest });
+    } catch (error) {
+      blockers.push({ code: error.code ?? 'ADOPTION_INPUT_INVALID', details: error.details ?? null });
+    }
+
+    const target = targetBranch ?? this.policy.target;
+    if (typeof target !== 'string' || !target.trim()) blockers.push({ code: 'ADOPTION_TARGET_REQUIRED' });
+    const candidateReference = candidateRef ?? (id ? `refs/torch/integration-candidates/${id}` : null);
+    const expectedReference = expectedOldRef ?? (target ? `refs/heads/${target}` : null);
+    if (!candidateReference || !/^refs\/torch\/integration-candidates\/[A-Za-z0-9._-]+$/.test(candidateReference)) blockers.push({ code: 'ADOPTION_CANDIDATE_REF_INVALID' });
+    if (!expectedReference || !/^refs\/heads\/[A-Za-z0-9._/-]+$/.test(expectedReference)) blockers.push({ code: 'ADOPTION_EXPECTED_REF_INVALID' });
+
+    let contributor; let destination;
+    try { contributor = this.controlPlane.assertIdentity(text(contributorArea, 'contributorArea')); }
+    catch (error) { blockers.push({ code: 'ADOPTION_CONTRIBUTOR_INVALID', details: error.code ?? null }); }
+    try { destination = this.controlPlane.assertIdentity(text(destinationArea, 'destinationArea')); }
+    catch (error) { blockers.push({ code: 'ADOPTION_DESTINATION_INVALID', details: error.code ?? null }); }
+    if (contributor && destination && contributor === destination) blockers.push({ code: 'ADOPTION_IDENTITIES_MUST_DIFFER' });
+
+    const binding = { action, adoptionId: id, candidateSha: candidate, baseTargetSha: base, provenanceManifestDigest: provenanceDigest,
+      contributorArea: contributor, destinationArea: destination, targetBranch: target, candidateRef: candidateReference,
+      expectedOldRef: expectedReference, adapter: adapterBinding };
+    const approval = approvalId ? this.controlPlane.database.prepare(`SELECT * FROM approval_requests WHERE id = ? AND project_id = ?`)
+      .get(approvalId, this.controlPlane.projectId) : null;
+    if (!approval) blockers.push({ code: 'ADOPTION_APPROVAL_MISSING' });
+    else if (approval.revision !== approvalRevision) blockers.push({ code: 'ADOPTION_APPROVAL_REVISION_STALE', actual: approval.revision });
+    else if (!exactAuthority(authorityFrom(approval), binding)) blockers.push({ code: 'ADOPTION_APPROVAL_BINDING_INVALID' });
+
+    if (candidate && git(this.repositoryRoot, ['cat-file', '-e', `${candidate}^{commit}`], { allowFailure: true }).status !== 0) blockers.push({ code: 'ADOPTION_CANDIDATE_MISSING' });
+    if (base && git(this.repositoryRoot, ['cat-file', '-e', `${base}^{commit}`], { allowFailure: true }).status !== 0) blockers.push({ code: 'ADOPTION_BASE_MISSING' });
+    if (candidate && base && git(this.repositoryRoot, ['merge-base', '--is-ancestor', base, candidate], { allowFailure: true }).status !== 0) blockers.push({ code: 'ADOPTION_CANDIDATE_NOT_DESCENDED_FROM_BASE' });
+    if (expectedReference && base) {
+      const current = this.refAdapter.readRefs([expectedReference])[expectedReference];
+      if (!current) blockers.push({ code: 'ADOPTION_EXPECTED_REF_MISSING' });
+      else if (current !== base) blockers.push({ code: 'ADOPTION_EXPECTED_REF_STALE', expected: base, actual: current });
+    }
+    const candidateExists = candidateReference && this.refAdapter.readRefs([candidateReference])[candidateReference] != null;
+    if (['plan', 'prepare', 'finalize'].includes(action) && candidateExists) blockers.push({ code: 'ADOPTION_CANDIDATE_REF_ALREADY_EXISTS' });
+    if (action === 'request' && !candidateExists) blockers.push({ code: 'ADOPTION_CANDIDATE_REF_MISSING' });
+
+    for (const worktree of listWorktreePaths(this.repositoryRoot)) {
+      const entries = git(worktree, ['status', '--porcelain=v1', '--untracked-files=all']).stdout.split('\n').filter(Boolean);
+      if (entries.length) blockers.push({ code: 'ADOPTION_WORKTREE_DIRTY', worktree, entries });
+    }
+    if (tableExists(this.controlPlane.database, 'worktree_guards')) {
+      for (const guard of this.controlPlane.database.prepare(`SELECT area_id, guard_type, reason FROM worktree_guards WHERE project_id = ? AND released_at IS NULL`).all(this.controlPlane.projectId)) blockers.push({ code: 'ADOPTION_WORKTREE_GUARD_ACTIVE', guard });
+    }
+    if (tableExists(this.controlPlane.database, 'resource_leases')) {
+      for (const lease of this.controlPlane.database.prepare(`SELECT id, resource_id, area_id FROM resource_leases WHERE released_at IS NULL`).all()) blockers.push({ code: 'ADOPTION_RESOURCE_LEASE_ACTIVE', lease });
+    }
+    if (tableExists(this.controlPlane.database, 'prepared_checks')) {
+      for (const check of this.controlPlane.database.prepare(`SELECT id, area_id, state FROM prepared_checks WHERE project_id = ? AND state IN ('prepared', 'running')`).all(this.controlPlane.projectId)) blockers.push({ code: 'ADOPTION_PREPARED_CHECK_ACTIVE', check });
+    }
+    for (const area of [contributor, destination].filter(Boolean)) {
+      const executor = observeRuntimeTurn({ stateRoot: this.controlPlane.stateRoot, repositoryRoot: this.repositoryRoot,
+        projectId: this.controlPlane.projectId, areaId: area, now: this.clock });
+      if (executor.state !== 'inactive') blockers.push({ code: 'ADOPTION_EXECUTOR_NOT_INACTIVE', areaId: area, executor });
+    }
+    return { action: `plan-adoption-${action}`, adoptionId: id, approval: approval ? { id: approval.id, revision: approval.revision, status: approval.status, approver: approval.approver_id } : null,
+      authority: binding, candidateSha: candidate ?? null, baseTargetSha: base ?? null, provenanceManifestDigest: provenanceDigest ?? null,
+      contributorArea: contributor ?? null, destinationArea: destination ?? null, targetBranch: target ?? null, candidateRef: candidateReference,
+      expectedOldRef: expectedReference, blockers, canProceed: blockers.length === 0, mutationPerformed: false };
+  }
+
+  #append(adoptionId, expectedRevision, record, operation) {
+    this.controlPlane.database.exec('BEGIN IMMEDIATE');
+    try {
+      this.#appendInTransaction(adoptionId, expectedRevision, record, operation);
+      this.controlPlane.database.exec('COMMIT');
+    } catch (error) { this.controlPlane.database.exec('ROLLBACK'); throw error; }
+  }
+
+  #appendInTransaction(adoptionId, expectedRevision, record, operation) {
+    const current = this.#find(adoptionId);
+    if (expectedRevision === null ? current : !current || current.revision !== expectedRevision) {
+      throw new TorchError('Adoption journal changed before this transition could be recorded', {
+        code: 'ADOPTION_RECORD_REVISION_STALE', details: { adoptionId, expectedRevision, actual: current?.revision ?? null },
+      });
+    }
+    this.controlPlane.database.prepare(`INSERT INTO audit_events
+      (id, project_id, actor_id, operation, entity_type, entity_id, details, created_at)
+      VALUES (?, ?, 'owner', ?, 'integration-adoption', ?, ?, ?)`)
+      .run(this.idFactory(), this.controlPlane.projectId, operation, adoptionId, JSON.stringify(record), this.clock().toISOString());
+  }
+
+  #fromRecord(record, input, action) {
+    exactInput(record, input);
+    return this.plan({ action, actorId: input.actorId, approvalId: input.approvalId, approvalRevision: input.approvalRevision,
+      adoptionId: record.id, candidateSha: record.candidateSha, baseTargetSha: record.baseTargetSha, provenance: record.provenance,
+      provenanceManifestDigest: record.provenanceManifestDigest, contributorArea: record.contributorArea, destinationArea: record.destinationArea,
+      targetBranch: record.targetBranch, candidateRef: record.candidateRef, expectedOldRef: record.expectedOldRef, adapter: input.adapter });
+  }
+
+  prepare(input = {}) {
+    const adoptionId = input.adoptionId ?? this.idFactory();
+    const plan = this.plan({ ...input, action: 'prepare', adoptionId });
+    if (!plan.canProceed) throw new TorchError('Adoption inputs are not eligible for preparation', { code: 'ADOPTION_PREPARE_BLOCKED', details: plan.blockers });
+    const now = this.clock().toISOString();
+    const record = { schema: ADOPTION_SCHEMA, id: adoptionId, revision: 1, state: 'prepared', approvalId: text(input.approvalId, 'approvalId'),
+      approvalRevision: input.approvalRevision, authority: plan.authority, provenance: [...input.provenance], candidateSha: plan.candidateSha,
+      baseTargetSha: plan.baseTargetSha, provenanceManifestDigest: plan.provenanceManifestDigest, candidateRef: plan.candidateRef,
+      expectedOldRef: plan.expectedOldRef, contributorArea: plan.contributorArea, destinationArea: plan.destinationArea, targetBranch: plan.targetBranch,
+      archiveRef: `refs/torch/integration-archives/${adoptionId}`, preparedAt: now, finalizedAt: null, rolledBackAt: null };
+    this.#append(adoptionId, null, record, 'integration.adoption.prepare');
+    return { ...record, mutationPerformed: true };
+  }
+
+  finalize(input = {}) {
+    const record = this.getPrepared(text(input.adoptionId, 'adoptionId'));
+    if (record.state !== 'prepared') throw new TorchError('Only a prepared adoption can be finalized', { code: 'ADOPTION_FINALIZE_STATE_INVALID', details: { state: record.state } });
+    const plan = this.#fromRecord(record, input, 'finalize');
+    if (!plan.canProceed) throw new TorchError('Adoption finalize is blocked', { code: 'ADOPTION_FINALIZE_BLOCKED', details: plan.blockers });
+    const receipt = this.refAdapter.compareAndSwapRefs({ expected: [{ ref: record.expectedOldRef, sha: record.baseTargetSha }, { ref: record.candidateRef, sha: null }, { ref: record.archiveRef, sha: null }],
+      writes: [{ ref: record.archiveRef, sha: record.baseTargetSha }, { ref: record.candidateRef, sha: record.candidateSha }] });
+    const finalized = { ...record, revision: record.revision + 1, state: 'finalized', finalizeApprovalId: text(input.approvalId, 'approvalId'),
+      finalizeApprovalRevision: input.approvalRevision, finalizedAt: this.clock().toISOString(), refReceiptDigest: digest(receipt) };
+    this.#append(record.id, record.revision, finalized, 'integration.adoption.finalize');
+    return { ...finalized, mutationPerformed: true };
+  }
+
+  reconcile(input = {}) {
+    const record = this.getPrepared(text(input.adoptionId, 'adoptionId'));
+    const plan = this.#fromRecord(record, input, 'reconcile');
+    if (!plan.canProceed) return { record, plan, state: 'blocked', mutationPerformed: false };
+    const refs = this.refAdapter.readRefs([record.expectedOldRef, record.candidateRef, record.archiveRef]);
+    const finalizedRefs = refs[record.expectedOldRef] === record.baseTargetSha && refs[record.candidateRef] === record.candidateSha && refs[record.archiveRef] === record.baseTargetSha;
+    if (record.state === 'prepared' && finalizedRefs) {
+      const recovered = { ...record, revision: record.revision + 1, state: 'finalized', finalizedAt: this.clock().toISOString(),
+        recoveredAt: this.clock().toISOString(), recovery: 'private-ref-cas-completed-before-journal' };
+      this.#append(record.id, record.revision, recovered, 'integration.adoption.reconcile-finalized');
+      return { record: recovered, refs, state: 'recovered-finalized', mutationPerformed: true };
+    }
+    if (record.state === 'finalized' && finalizedRefs) return { record, refs, state: 'consistent', mutationPerformed: false };
+    if (record.state === 'prepared' && refs[record.candidateRef] === null && refs[record.archiveRef] === null) return { record, refs, state: 'prepared', mutationPerformed: false };
+    return { record, refs, state: 'diverged', mutationPerformed: false };
+  }
+
+  rollback(input = {}) {
+    const record = this.getPrepared(text(input.adoptionId, 'adoptionId'));
+    if (record.state !== 'finalized') throw new TorchError('Only a finalized adoption can be rolled back', { code: 'ADOPTION_ROLLBACK_STATE_INVALID', details: { state: record.state } });
+    const plan = this.#fromRecord(record, input, 'rollback');
+    if (!plan.canProceed) throw new TorchError('Adoption rollback is blocked', { code: 'ADOPTION_ROLLBACK_BLOCKED', details: plan.blockers });
+    const refs = this.refAdapter.readRefs([record.expectedOldRef, record.candidateRef, record.archiveRef]);
+    const recovered = refs[record.expectedOldRef] === record.baseTargetSha
+      && refs[record.candidateRef] === null && refs[record.archiveRef] === record.baseTargetSha;
+    if (!recovered && refs[record.candidateRef] !== record.candidateSha) {
+      throw new TorchError('Candidate ref changed before rollback', { code: 'ADOPTION_ROLLBACK_REF_DIVERGED', details: refs });
+    }
+    const receipt = recovered ? { adapter: { name: this.refAdapter.name, id: this.refAdapter.id }, recovered: true, refs }
+      : this.refAdapter.compareAndSwapRefs({ expected: [{ ref: record.candidateRef, sha: record.candidateSha }], writes: [{ ref: record.candidateRef, sha: null }] });
+    const rolledBack = { ...record, revision: record.revision + 1, state: 'rolled-back', rollbackApprovalId: text(input.approvalId, 'approvalId'),
+      rollbackApprovalRevision: input.approvalRevision, rolledBackAt: this.clock().toISOString(), rollbackReceiptDigest: digest(receipt),
+      ...(recovered ? { recoveredAt: this.clock().toISOString(), recovery: 'private-ref-rollback-completed-before-journal' } : {}) };
+    this.#append(record.id, record.revision, rolledBack, 'integration.adoption.rollback');
+    return { ...rolledBack, mutationPerformed: true };
+  }
+
+  requestAdoptedCandidate(input = {}, { requestIntegration, findIntegrationRequest } = {}) {
+    if (typeof requestIntegration !== 'function') throw new TorchError('Native integration request adapter is unavailable', { code: 'ADOPTION_NATIVE_REQUEST_UNAVAILABLE' });
+    const adoptionId = text(input.adoptionId, 'adoptionId');
+    this.controlPlane.database.exec('BEGIN IMMEDIATE');
+    try {
+      const record = this.getPrepared(adoptionId);
+      exactInput(record, input);
+      if (!['finalized', 'requesting'].includes(record.state)) {
+        throw new TorchError('Only a finalized adoption can request native qualification', {
+          code: 'ADOPTION_REQUEST_STATE_INVALID', details: { state: record.state },
+        });
+      }
+      const plan = this.#fromRecord(record, input, 'request');
+      if (!plan.canProceed) throw new TorchError('Adopted candidate request is blocked', { code: 'ADOPTION_REQUEST_BLOCKED', details: plan.blockers });
+      const requesting = record.state === 'requesting' ? record : { ...record, revision: record.revision + 1, state: 'requesting',
+        requestApprovalId: text(input.approvalId, 'approvalId'), requestApprovalRevision: input.approvalRevision, requestStartedAt: this.clock().toISOString() };
+      if (record.state !== 'requesting') this.#appendInTransaction(record.id, record.revision, requesting, 'integration.adoption.request-started');
+      const existing = findIntegrationRequest?.({ candidateRef: requesting.candidateRef, commit: requesting.candidateSha });
+      if (existing && (existing.sourceArea !== requesting.contributorArea || existing.targetBranch !== requesting.targetBranch)) {
+        throw new TorchError('Existing private-ref integration request does not match this adoption', {
+          code: 'ADOPTION_INTEGRATION_REQUEST_CONFLICT', details: { adoptionId: requesting.id, integrationRequestId: existing.id },
+        });
+      }
+      const request = existing ?? requestIntegration({ adoptionId: requesting.id, areaId: requesting.contributorArea,
+        commit: requesting.candidateSha, candidateRef: requesting.candidateRef });
+      const requested = { ...requesting, revision: requesting.revision + 1, state: 'requested', integrationRequestId: request.id, requestedAt: this.clock().toISOString() };
+      this.#appendInTransaction(requesting.id, requesting.revision, requested, 'integration.adoption.request');
+      this.controlPlane.database.exec('COMMIT');
+      return { ...requested, integrationRequest: request, mutationPerformed: true };
+    } catch (error) {
+      this.controlPlane.database.exec('ROLLBACK');
+      throw error;
+    }
+  }
+
+  #find(adoptionId) {
+    const row = this.controlPlane.database.prepare(`SELECT details FROM audit_events WHERE project_id = ?
+      AND entity_type = 'integration-adoption' AND entity_id = ? ORDER BY rowid DESC LIMIT 1`).get(this.controlPlane.projectId, adoptionId);
+    return durableRecord(row);
+  }
+
+  getPrepared(adoptionId) {
+    const id = text(adoptionId, 'adoptionId');
+    const record = this.#find(id);
+    if (!record) throw new TorchError('Unknown adoption journal record', { code: 'ADOPTION_NOT_FOUND', details: { adoptionId: id } });
+    return record;
+  }
+}

src/integration/service.mjs

@@ -4,6 +4,7 @@ import { TorchError } from '../kernel/errors.mjs';
 import { readInstallManifest } from '../kernel/install.mjs';
 import { loadFleetDefinition } from '../kernel/worktrees.mjs';
 import { BacklogService } from '../backlog/service.mjs';
+import { AdoptionService } from './adoption.mjs';
 
 function git(root, args, { allowFailure = false } = {}) {
   const result = spawnSync('git', ['-C', root, ...args], { encoding: 'utf8' });
@@ -15,6 +16,21 @@ function git(root, args, { allowFailure = false } = {}) {
   return { status: result.status, stdout: result.stdout?.trim() ?? '', stderr: result.stderr?.trim() ?? '' };
 }
 
+function text(value, name) {
+  if (typeof value !== 'string' || !value.trim()) {
+    throw new TorchError(`${name} is required`, { code: 'INTEGRATION_INPUT_INVALID', details: { name } });
+  }
+  return value.trim();
+}
+
+function sha(value, name) {
+  const normalized = text(value, name);
+  if (!/^[0-9a-f]{40,64}$/.test(normalized)) {
+    throw new TorchError(`${name} must be a full Git object id`, { code: 'INTEGRATION_INPUT_INVALID', details: { name } });
+  }
+  return normalized;
+}
+
 function integrationRow(row) {
   if (!row) return null;
   return {
@@ -42,6 +58,8 @@ export class IntegrationService {
     this.worktrees = new Map((manifest.external ?? [])
       .filter((entry) => entry.type === 'worktree')
       .map((entry) => [entry.area, entry]));
+    this.adoption = new AdoptionService({ repositoryRoot, controlPlane, clock, idFactory,
+      policy: { target: this.policy.target } });
     this.controlPlane.database.exec(`
       CREATE TABLE IF NOT EXISTS integration_requests (
         id TEXT PRIMARY KEY,
@@ -84,12 +102,56 @@ export class IntegrationService {
         code: 'INTEGRATION_SOURCE_MOVED', details: { branchTip, sourceCommit },
       });
     }
+    return this.#registerRequest({ area, sourceBranch: worktree.branch, sourceCommit });
+  }
+
+  requestCandidate({ areaId, commit, candidateRef } = {}) {
+    throw new TorchError('Private candidate intake requires an authenticated finalized adoption', {
+      code: 'INTEGRATION_ADOPTED_INTAKE_INTERNAL_ONLY', details: { areaId, commit, candidateRef },
+    });
+  }
+
+  #requestFinalizedCandidate({ adoptionId, areaId, commit, candidateRef } = {}) {
+    const area = this.controlPlane.assertIdentity(areaId);
+    if (area === 'session-manager') {
+      throw new TorchError('Session Manager does not own a feature worktree by default', {
+        code: 'INTEGRATION_SOURCE_INVALID',
+      });
+    }
+    const adoption = this.adoption.getPrepared(text(adoptionId, 'adoptionId'));
+    if (adoption.state !== 'requesting' || adoption.contributorArea !== area
+      || adoption.candidateSha !== commit || adoption.candidateRef !== candidateRef) {
+      throw new TorchError('Private candidate does not match an authenticated finalized adoption', {
+        code: 'INTEGRATION_ADOPTED_BINDING_INVALID', details: { adoptionId, state: adoption.state },
+      });
+    }
+    const ref = text(candidateRef, 'candidateRef');
+    if (!/^refs\/torch\/integration-candidates\/[A-Za-z0-9._-]+$/.test(ref)) {
+      throw new TorchError('Adopted candidates must use a private integration candidate ref', {
+        code: 'INTEGRATION_ADOPTED_REF_INVALID', details: { candidateRef: ref },
+      });
+    }
+    const sourceCommit = sha(commit, 'commit');
+    const refTip = git(this.repositoryRoot, ['rev-parse', '--verify', ref], { allowFailure: true });
+    if (refTip.status !== 0 || refTip.stdout !== sourceCommit) {
+      throw new TorchError('Adopted candidate ref no longer names the exact requested commit', {
+        code: 'INTEGRATION_ADOPTED_REF_STALE', details: { candidateRef: ref, sourceCommit, actual: refTip.stdout || null },
+      });
+    }
+    const existing = this.controlPlane.database.prepare(`SELECT id FROM integration_requests
+      WHERE project_id = ? AND source_branch = ? AND source_commit = ? ORDER BY rowid LIMIT 1`)
+      .get(this.controlPlane.projectId, ref, sourceCommit);
+    if (existing) return this.get(existing.id);
+    return this.#registerRequest({ area, sourceBranch: ref, sourceCommit });
+  }
+
+  #registerRequest({ area, sourceBranch, sourceCommit }) {
     const targetBranch = this.policy.target;
     const baseTargetCommit = git(this.repositoryRoot, ['rev-parse', targetBranch]).stdout;
     const now = this.clock().toISOString();
     const record = {
       id: this.idFactory(), projectId: this.controlPlane.projectId, sourceArea: area,
-      sourceBranch: worktree.branch, sourceCommit, targetBranch, baseTargetCommit,
+      sourceBranch, sourceCommit, targetBranch, baseTargetCommit,
       state: 'testing', reason: null, requiredChecks: this.policy.required_checks ?? [],
       authorizedBy: null, authorizedAt: null, createdAt: now, updatedAt: now, landedAt: null,
     };
@@ -110,6 +172,44 @@ export class IntegrationService {
     return this.evaluate(record.id);
   }
 
+  planAdoption(input = {}) {
+    return this.adoption.plan(input);
+  }
+
+  prepareAdoption(input = {}) {
+    return this.adoption.prepare(input);
+  }
+
+  getPreparedAdoption(adoptionId) {
+    return this.adoption.getPrepared(adoptionId);
+  }
+
+  finalizeAdoption(input = {}) {
+    return this.adoption.finalize(input);
+  }
+
+  reconcileAdoption(input = {}) {
+    return this.adoption.reconcile(input);
+  }
+
+  rollbackAdoption(input = {}) {
+    return this.adoption.rollback(input);
+  }
+
+  requestAdoptedCandidate(input = {}) {
+    return this.adoption.requestAdoptedCandidate(input, {
+      requestIntegration: ({ adoptionId, areaId, commit, candidateRef }) => this.#requestFinalizedCandidate({ adoptionId, areaId, commit, candidateRef }),
+      findIntegrationRequest: ({ candidateRef, commit }) => this.findCandidateRequest({ candidateRef, commit }),
+    });
+  }
+
+  findCandidateRequest({ candidateRef, commit } = {}) {
+    const row = this.controlPlane.database.prepare(`SELECT id FROM integration_requests
+      WHERE project_id = ? AND source_branch = ? AND source_commit = ? ORDER BY rowid LIMIT 1`)
+      .get(this.controlPlane.projectId, candidateRef, commit);
+    return row ? this.get(row.id) : null;
+  }
+
   get(requestId) {
     const row = this.controlPlane.database.prepare('SELECT * FROM integration_requests WHERE id = ?').get(requestId);
     if (!row) throw new TorchError(`Unknown integration request: ${requestId}`, { code: 'INTEGRATION_NOT_FOUND' });

src/mcp/tools.mjs

@@ -19,6 +19,7 @@ export const TORCH_CORE_MCP_TOOL_NAMES = Object.freeze([
   'torch_who_owns',
   'torch_neighbours',
   'torch_send_message',
+  'torch_reply_to_owner',
   'torch_read_messages',
   'torch_ack_message',
   'torch_report_status',
@@ -100,6 +101,19 @@ function claimedIdentity(controlPlane, actorId, claim, field) {
   });
 }
 
+function boundActorIdentity(controlPlane, actorId) {
+  if (typeof actorId !== 'string' || actorId.trim().length === 0) {
+    throw new TorchError('torch_reply_to_owner requires an actual bound Fleet identity', {
+      code: 'FLEET_IDENTITY_REQUIRED', details: { field: 'bound_actor' },
+    });
+  }
+  return controlPlane.assertIdentity(actorId);
+}
+
+function inputSchema(tool) {
+  return typeof tool.schema?.safeParse === 'function' ? tool.schema : z.object(tool.schema);
+}
+
 export function createTorchToolset(controlPlane, {
   actorId, backlogService, checkService, resourceService, integrationService, contextTelemetryService, scheduleService,
   evolutionService, hierarchyService, repositoryRoot,
@@ -169,6 +183,15 @@ export function createTorchToolset(controlPlane, {
         sender: claimedIdentity(controlPlane, actorId, sender, 'sender'), recipient, body, references,
       }),
     }],
+    ['torch_reply_to_owner', {
+      description: 'Reply to one owner-originated request as the bound Fleet identity. This tool cannot select a sender, recipient, owner, or conversation history.',
+      schema: z.object({
+        request_id: z.string().min(1), body: z.string().min(1),
+      }).strict(),
+      invoke: ({ request_id: requestId, body }) => controlPlane.sendOwnerReply({
+        sender: boundActorIdentity(controlPlane, actorId), requestId, body,
+      }),
+    }],
     ['torch_read_messages', {
       description: 'Read durable direct and group messages visible to one Fleet identity.',
       schema: {
@@ -736,14 +759,14 @@ export function describeTorchTools() {
   }).entries()].map(([name, tool]) => ({
     name,
     description: tool.description,
-    inputSchema: z.toJSONSchema(z.object(tool.schema)),
+    inputSchema: z.toJSONSchema(inputSchema(tool)),
   }));
 }
 
 export function callTorchTool(controlPlane, name, input = {}, options = {}) {
   const tool = createTorchToolset(controlPlane, options).get(name);
   if (!tool) throw new Error(`Unknown TORCH MCP tool: ${name}`);
-  const parsed = z.object(tool.schema).parse(input);
+  const parsed = inputSchema(tool).parse(input);
   return tool.invoke(parsed);
 }
 
@@ -762,7 +785,7 @@ export function createTorchMcpServer(controlPlane, {
     actorId, backlogService, checkService, resourceService, integrationService, contextTelemetryService, scheduleService,
     evolutionService, hierarchyService, repositoryRoot, artifactService,
   })) {
-    server.registerTool(name, { description: tool.description, inputSchema: tool.schema }, async (input) => {
+    server.registerTool(name, { description: tool.description, inputSchema: inputSchema(tool) }, async (input) => {
       try {
         const result = await tool.invoke(input);
         return {

src/observability/snapshot.mjs

@@ -16,6 +16,8 @@ import { assessBacklogHealth, backlogObservedCommitDistance } from '../backlog/s
 import { observeTaskActivity } from '../backlog/activity.mjs';
 import { ScheduleLauncherService } from '../schedules/launcher.mjs';
 import { hierarchyOrder } from '../runtime/hierarchy-order.mjs';
+import { observeRuntimeTurn } from '../runtime/turn-guard.mjs';
+import { observeContinuation } from '../runtime/continuation.mjs';
 
 function json(path) {
   return JSON.parse(readFileSync(path, 'utf8'));
@@ -403,6 +405,7 @@ export function observeProject({ repositoryRoot, env = process.env, now = () =>
       metrics: ['cachedInput', 'uncachedInput', 'cacheCreation', 'cacheRead', 'compactions', 'resumedPromptBytes', 'costPerVerifiedItem'],
     },
   };
+  result.continuation = observeContinuation({ stateRoot, tasks: result.backlog, now });
   result.backlogHealth = backlogHealth(repository.root, result.backlog, result.agents, now);
   result.backlogActivity = observeBacklogActivity({ repositoryRoot: repository.root,
     tasks: result.backlog, config, manifest, now: now() });
@@ -417,11 +420,21 @@ export function observeProject({ repositoryRoot, env = process.env, now = () =>
       `).all().map((row) => [row.area_id, row]));
       result.agents = result.agents.map((agent) => {
         const row = live.get(agent.areaId);
-        return row ? {
+        const reported = row ? {
           ...agent, state: row.state, runtime: row.runtime ?? agent.runtime,
           runtimeSessionId: row.runtime_session_id, summary: row.summary,
           task: row.task, heartbeatAt: row.heartbeat_at,
         } : agent;
+        const executor = observeRuntimeTurn({ stateRoot, repositoryRoot: repository.root,
+          projectId: manifest.projectId, areaId: agent.areaId, now });
+        const staleIdle = ['idle', 'offline'].includes(reported.state);
+        return { ...reported, reportedState: reported.state, reportedSummary: reported.summary, executor,
+          ...(executor.state === 'active' && staleIdle ? {
+            state: executor.phase,
+            summary: `Live runtime executor is ${executor.phase} (last reported state: ${reported.state}).`,
+          } : executor.state === 'unknown' && staleIdle ? {
+            state: 'unknown', summary: 'Runtime turn guard exists, but its executor could not be verified.',
+          } : {}) };
       });
     }
     if (hasTable(database, 'schedule_wake_reservations')) {

src/runtime/continuation.mjs

@@ -0,0 +1,178 @@
+import { createHash, randomUUID } from 'node:crypto';
+import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmdirSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { TorchError } from '../kernel/errors.mjs';
+
+function statePath(control) { return join(control.stateRoot, 'continuation.json'); }
+function occupiedAreas(control) {
+  const path = join(control.stateRoot, 'sessions', 'turn-guards');
+  return existsSync(path) ? readdirSync(path, { withFileTypes: true }).filter(d => d.isDirectory()).map(d => d.name) : [];
+}
+function taskKey(tasks) { return JSON.stringify(tasks.map(t => [t.id, t.revision]).sort((a, b) => a[0].localeCompare(b[0]))); }
+function dailyCap(state, day) {
+  return state.dailyOverride?.day === day ? Math.max(state.maxTurnsPerDay, state.dailyOverride.maxTurns) : state.maxTurnsPerDay;
+}
+function read(control) {
+  const policyPath = join(control.stateRoot, 'continuation-policy.json');
+  const state = { enabled: false, maxTurnsPerDay: 12, maxConcurrency: 3, areas: [], day: null, attempts: 0, handled: {}, progress: {},
+    ...(existsSync(statePath(control)) ? JSON.parse(readFileSync(statePath(control), 'utf8')) : {}),
+    ...(existsSync(policyPath) ? JSON.parse(readFileSync(policyPath, 'utf8')) : {}) };
+  if (typeof state.enabled !== 'boolean' || !Number.isInteger(state.maxTurnsPerDay)
+    || state.maxTurnsPerDay < 1 || state.maxTurnsPerDay > 48 || !Array.isArray(state.areas)
+    || !Number.isInteger(state.maxConcurrency) || state.maxConcurrency < 1 || state.maxConcurrency > 7
+    || !Number.isInteger(state.attempts) || state.attempts < 0 || !state.handled || typeof state.handled !== 'object'
+    || !state.progress || typeof state.progress !== 'object'
+    || (state.dailyOverride != null && (!/^\d{4}-\d{2}-\d{2}$/.test(state.dailyOverride.day)
+      || !Number.isInteger(state.dailyOverride.maxTurns) || state.dailyOverride.maxTurns < 1 || state.dailyOverride.maxTurns > 96))) {
+    throw new TorchError('Continuation policy or attempt ledger is invalid; launch refused.', { code: 'CONTINUATION_POLICY_INVALID' });
+  }
+  return state;
+}
+function save(control, state, policy = false) {
+  const path = policy ? join(control.stateRoot, 'continuation-policy.json') : statePath(control);
+  const temporary = `${path}.${randomUUID()}.tmp`;
+  const value = policy ? { enabled: state.enabled, maxTurnsPerDay: state.maxTurnsPerDay, maxConcurrency: state.maxConcurrency, areas: state.areas, dailyOverride: state.dailyOverride ?? null }
+    : { day: state.day, attempts: state.attempts, handled: state.handled, progress: state.progress };
+  writeFileSync(temporary, JSON.stringify(value), { flag: 'wx', mode: 0o600 });
+  renameSync(temporary, path);
+}
+
+export function configureContinuation(control, { actorId, enabled, maxTurnsPerDay, maxConcurrency, todayMaxTurns } = {}) {
+  control.assertOwnerActor(actorId);
+  const previous = read(control);
+  maxTurnsPerDay ??= previous.maxTurnsPerDay;
+  maxConcurrency ??= previous.maxConcurrency;
+  if (typeof enabled !== 'boolean' || !Number.isInteger(maxTurnsPerDay) || maxTurnsPerDay < 1 || maxTurnsPerDay > 48
+    || !Number.isInteger(maxConcurrency) || maxConcurrency < 1 || maxConcurrency > 7
+    || (todayMaxTurns !== undefined && (!Number.isInteger(todayMaxTurns) || todayMaxTurns < maxTurnsPerDay || todayMaxTurns > 96))) {
+    throw new TorchError('Continuation requires explicit enable/pause, 1–48 daily turns, 1–7 concurrent turns and an optional same-day total between the daily cap and 96.', { code: 'CONTINUATION_POLICY_INVALID' });
+  }
+  const state = { ...previous, enabled, maxTurnsPerDay, maxConcurrency, areas: control.listAgents().map(a => a.areaId) };
+  if (todayMaxTurns !== undefined) state.dailyOverride = { day: control.clock().toISOString().slice(0, 10), maxTurns: todayMaxTurns };
+  save(control, state, true);
+  control.auditOwnerAction({ actorId, operation: 'runtime.continuation.configure', entityType: 'project', entityId: control.projectId,
+    details: { enabled, maxTurnsPerDay, maxConcurrency, dailyOverride: state.dailyOverride ?? null, areas: state.areas } });
+  return { enabled, maxTurnsPerDay, maxConcurrency, dailyOverride: state.dailyOverride ?? null, areas: state.areas, sessionsStarted: false };
+}
+
+export function observeContinuation({ stateRoot, tasks = [], now = () => new Date() } = {}) {
+  try {
+    const state = read({ stateRoot });
+    const day = now().toISOString().slice(0, 10);
+    const attempts = state.day === day ? state.attempts : 0;
+    const cap = dailyCap(state, day);
+    const remainingTurns = Math.max(0, cap - attempts);
+    const held = Object.entries(state.progress).filter(([area, progress]) => {
+      const active = tasks.filter(t => t.owner === area && ['assigned', 'in_progress'].includes(t.state));
+      return active.length && progress?.status === 0 && progress.noProgress >= 2 && progress.taskKey === taskKey(active);
+    }).map(([areaId]) => ({ areaId, reason: 'no-progress-needs-coordination' }));
+    return { available: existsSync(join(stateRoot, 'continuation-policy.json')), enabled: state.enabled,
+      day, dayTimezone: 'UTC', attempts, maxTurnsPerDay: cap, configuredMaxTurnsPerDay: state.maxTurnsPerDay,
+      dailyOverride: state.dailyOverride ?? null, maxConcurrency: state.maxConcurrency, remainingTurns,
+      stopReason: !state.enabled ? 'paused' : remainingTurns === 0 ? 'daily-turn-cap' : null,
+      capacityHeldBy: occupiedAreas({ stateRoot }), held,
+      budgetKind: 'automatic-turn-cap', manualTurnsIncluded: false, tokenOrCostGovernance: 'unknown',
+      timerInstallationVerified: false, mutationPerformed: false };
+  } catch {
+    return { available: false, stopReason: 'policy-or-ledger-unreadable', mutationPerformed: false };
+  }
+}
+
+export function planContinuation(control, backlog, { at = new Date(), limit } = {}) {
+  const state = read(control);
+  const day = at.toISOString().slice(0, 10);
+  const attempts = state.day === day ? state.attempts : 0;
+  const cap = dailyCap(state, day);
+  if (limit !== undefined && (!Number.isInteger(limit) || limit < 1 || limit > 7)) throw new TorchError('Continuation limit must be 1–7.', { code: 'CONTINUATION_POLICY_INVALID' });
+  const capacity = Math.min(limit ?? state.maxConcurrency, state.maxConcurrency);
+  const metadata = JSON.parse(readFileSync(join(control.stateRoot, 'project.json'), 'utf8'));
+  if (!state.enabled || metadata.detachedAt || attempts >= cap) {
+    return { enabled: state.enabled, maxConcurrency: capacity, reason: metadata.detachedAt ? 'detached' : !state.enabled ? 'paused' : 'daily-turn-cap', candidates: [], attempts };
+  }
+  const tasks = backlog.list();
+  const candidates = [];
+  const held = [];
+  for (const areaId of state.areas) {
+    const agent = control.getAgent(areaId);
+    if (!['idle', 'offline', 'waiting'].includes(agent.state)
+      || existsSync(join(control.stateRoot, 'sessions', 'turn-guards', areaId))) continue;
+    const messages = control.readMessages({ recipient: areaId, unacknowledgedOnly: true, limit: 1000 });
+    const active = tasks.filter(t => t.owner === areaId && ['assigned', 'in_progress'].includes(t.state)
+      && (t.dependencies ?? []).every(id => tasks.find(t => t.id === id)?.state === 'completed'));
+    const previous = state.handled[areaId] ?? {};
+    const freshMessages = messages.filter(m => !previous.messageIds?.includes(m.id));
+    const freshTasks = active.filter(t => previous.taskRevisions?.[t.id] !== t.revision);
+    const progress = state.progress[areaId];
+    const continueWork = active.length > 0 && progress?.status === 0
+      && progress.taskKey === taskKey(active) && progress.noProgress < 2;
+    if (!freshMessages.length && !freshTasks.length && !continueWork) {
+      if (active.length && progress?.noProgress >= 2) held.push({ areaId, reason: 'no-progress-needs-coordination', taskIds: active.map(t => t.id) });
+      continue;
+    }
+    const signal = { messageIds: messages.map(m => m.id), taskRevisions: Object.fromEntries(active.map(t => [t.id, t.revision])) };
+    const fingerprint = createHash('sha256').update(JSON.stringify(signal)).digest('hex');
+    candidates.push({ areaId, fingerprint, signal, workKey: taskKey(active),
+      messageCount: freshMessages.length, taskIds: active.map(t => t.id), continueWork });
+  }
+  const ordinal = area => state.handled[area]?.dispatchedDay === day ? state.handled[area].dispatchedOrdinal ?? 0 : 0;
+  candidates.sort((a, b) => ordinal(a.areaId) - ordinal(b.areaId)
+    || Number(b.areaId === 'session-manager') - Number(a.areaId === 'session-manager'));
+  return { enabled: true, maxConcurrency: capacity, day, attempts, remainingTurns: cap - attempts,
+    occupiedAreas: occupiedAreas(control), held,
+    candidates: candidates.slice(0, Math.min(capacity, cap - attempts)), mutationPerformed: false };
+}
+
+export async function runContinuation(control, backlog, { actorId, launch, at, clock = () => at ?? new Date(), limit } = {}) {
+  control.assertOwnerActor(actorId);
+  if (typeof launch !== 'function') throw new TorchError('Continuation needs an authorized launcher.', { code: 'RUNTIME_EXECUTION_NOT_AUTHORIZED' });
+  const lock = join(control.stateRoot, 'continuation.lock');
+  try { mkdirSync(lock); } catch (error) {
+    if (error.code === 'EEXIST') return { reason: 'controller-active-or-unreconciled', launched: [] };
+    throw error;
+  }
+  try {
+    const pending = new Map();
+    const launched = [];
+    let reason = null;
+    while (true) {
+      // Refill immediately after EACH completion. Pending promises and physical
+      // guards both count, including manual executors outside this controller.
+      let current = planContinuation(control, backlog, { at: clock(), limit });
+      let occupied = new Set([...occupiedAreas(control), ...pending.keys()]);
+      while (occupied.size < current.maxConcurrency && current.enabled && !current.reason) {
+        const candidate = current.candidates.find(c => !occupied.has(c.areaId));
+        if (!candidate) break;
+        const state = read(control);
+        if (state.day !== current.day) { state.day = current.day; state.attempts = 0; }
+        state.attempts++;
+        state.handled[candidate.areaId] = { ...candidate.signal,
+          dispatchedDay: current.day, dispatchedOrdinal: state.attempts };
+        save(control, state); // Reserve before launch; unknown attempts are never refunded.
+        let result;
+        try { result = launch(candidate.areaId); } catch { result = null; }
+        pending.set(candidate.areaId, Promise.resolve(result).catch(() => null)
+          .then(status => ({ areaId: candidate.areaId, status, candidate })));
+        current = planContinuation(control, backlog, { at: clock(), limit });
+        occupied = new Set([...occupiedAreas(control), ...pending.keys()]);
+      }
+      reason = current.reason ?? (occupied.size >= current.maxConcurrency ? 'capacity-full' : null);
+      if (!pending.size) break;
+      const completed = await Promise.race(pending.values());
+      pending.delete(completed.areaId);
+      launched.push({ areaId: completed.areaId, status: completed.status });
+      const state = read(control);
+      const active = backlog.list().filter(t => t.owner === completed.areaId && ['assigned', 'in_progress'].includes(t.state));
+      const key = taskKey(active);
+      const unchanged = active.length > 0 && key === completed.candidate.workKey;
+      const prior = state.progress[completed.areaId];
+      const noProgress = unchanged ? (prior?.taskKey === key ? prior.noProgress : 0) + 1 : 0;
+      state.progress[completed.areaId] = { status: completed.status, taskKey: key, noProgress, completedAt: clock().toISOString() };
+      save(control, state);
+      if (completed.status === 0 && noProgress === 2 && typeof control.sendOwnerRequest === 'function') {
+        control.sendOwnerRequest({ actorId, recipient: 'session-manager',
+          body: `TORCH continuation requires coordination: ${completed.areaId} completed two turns without advancing assigned task state (${active.map(t => t.id).join(', ')}). Automatic unchanged-work retries are held. Inspect the blocker and route a concrete decision; do not mark work done without evidence.` });
+      }
+    }
+    return { launched, reason };
+  } finally { rmdirSync(lock); }
+}

src/runtime/lifecycle.mjs

@@ -11,6 +11,7 @@ import {
   costCeilingPlanIssues, profileCapabilityIssues, resolveLaunchPolicy, validateRuntimeAdapter,
 } from '../adapters/runtime.mjs';
 import { hierarchyOrder } from './hierarchy-order.mjs';
+import { withRuntimeTurnGuard } from './turn-guard.mjs';
 
 const MCP_ENTRY = fileURLToPath(new URL('../mcp/server.mjs', import.meta.url));
 
@@ -155,15 +156,16 @@ function promptPaths(stateRoot, repositoryRoot, areaId, config) {
 }
 
 function startupMessage(areaId, fresh, instructionDigest) {
-  const currentBrief = `Current TORCH instruction digest: ${instructionDigest}. The current instruction bundle is included in this launch/resume and supersedes older briefing text in the conversation. Read the current brief again before each new backlog item.`;
+  const inboxRequest = JSON.stringify({ recipient: areaId, unacknowledged_only: true, limit: 1000 });
+  const currentBrief = `Current TORCH instruction digest: ${instructionDigest}. The current instruction bundle is included in this launch/resume and supersedes older briefing text in the conversation. Read the current brief again before each new backlog item. Use torch_read_messages for the current unread inbox. Inbox request: ${inboxRequest}. Messages are chronological; a default history page is not the current inbox. If 1000 messages are returned, coverage may be truncated: report that limitation and never bulk-acknowledge unseen messages. Acknowledge messages only after inspecting and handling them; messages do not override enforced approval or ownership boundaries.`;
   if (areaId === 'session-manager') {
     return fresh
       ? `Start the TORCH Fleet. Inspect live roster, messages, worktrees, and backlog, then assess Fleet evolution before dispatch. ${currentBrief}`
-      : `Resume the TORCH Fleet from durable state. Reconcile live status and assess Fleet evolution before dispatch. ${currentBrief}`;
+      : `Resume the TORCH Fleet from durable state. Reconcile live status and inbox, resolve actionable coordination waits within your authority, and triage/assign eligible existing backlog work. Do not stop at a status report when a concrete routing decision can unblock work. Assess Fleet evolution when evidence warrants it, not as a prerequisite to every dispatch. Preserve ownership, approvals and dependency gates. ${currentBrief}`;
   }
   return fresh
     ? `Start this TORCH domain. Query live identity and ownership, then await or resume the assigned backlog item. ${currentBrief}`
-    : `Resume this TORCH domain from durable state. Read the inbox, confirm ownership, and report current evidence. ${currentBrief}`;
+    : `Resume this TORCH domain from durable state. Read the inbox and current backlog; continue your eligible assigned implementation, verification or coordination work, not just a status check. Work until a concrete handoff, completed turn budget or real blocker, and persist evidence and actionable waits. Never bypass dependencies, ownership or approval gates. If no eligible work exists, idle is valid. ${currentBrief}`;
 }
 
 function atomicJson(path, value) {
@@ -570,7 +572,9 @@ export function startFleet({ plan, controlPlane, executor, adapters } = {}) {
   const runtimes = adapterMap({ adapters });
   for (const action of plan.actions) {
     controlPlane.assertIdentity(action.areaId);
+    withRuntimeTurnGuard(controlPlane, action.areaId, setPhase => {
     writeCombinedPrompt(action);
+    setPhase('working');
     const result = executor({
       ...action.launch, areaId: action.areaId, runtime: action.runtime,
       runtimeSessionId: action.runtimeSessionId,
@@ -634,6 +638,7 @@ export function startFleet({ plan, controlPlane, executor, adapters } = {}) {
         ? `${action.mode} turn completed` : `${action.mode} requested`,
     });
     started.push({ areaId: action.areaId, runtimeSessionId: identity.runtimeSessionId, mode: action.mode });
+    });
   }
   if (plan.stateRoot) {
     const metadata = projectMetadata(plan.stateRoot);

src/runtime/stopped-executor-evidence.mjs

@@ -0,0 +1,106 @@
+import { spawnSync } from 'node:child_process';
+import { createHash } from 'node:crypto';
+import { readFileSync, readdirSync, readlinkSync, statSync } from 'node:fs';
+import { basename, join, resolve } from 'node:path';
+import { TorchError } from '../kernel/errors.mjs';
+
+const issued = new WeakMap();
+const TTL_MS = 60_000;
+function refuse(message) {
+  throw new TorchError(message, { code: 'RUNTIME_RECOVERY_UNVERIFIED' });
+}
+
+export function readStoppedUnit(unit) {
+  const result = spawnSync('systemctl', ['--user', 'show', unit,
+    '--property=Id,ActiveState,SubState,MainPID,ControlPID,InvocationID,ExecStart,WorkingDirectory,ControlGroup,ExecMainExitTimestamp'],
+  { encoding: 'utf8', timeout: 5000, maxBuffer: 64 * 1024 });
+  if (result.error || result.signal || result.status !== 0) refuse('Cannot verify the owner-selected service invocation.');
+  return Object.fromEntries(result.stdout.trim().split('\n').map(line => {
+    const index = line.indexOf('=');
+    return [line.slice(0, index), line.slice(index + 1)];
+  }));
+}
+
+export function readNativeProcesses() {
+  if (process.platform !== 'linux') refuse('Legacy process inspection is supported only on Linux.');
+  const processes = [];
+  for (const name of readdirSync('/proc').filter(value => /^\d+$/.test(value))) {
+    const directory = join('/proc', name);
+    try {
+      if (statSync(directory).uid !== process.getuid()) continue;
+      const argv = readFileSync(join(directory, 'cmdline'), 'utf8').split('\0').filter(Boolean);
+      if (!argv.length) continue;
+      const native = argv.some(value => /^(?:codex|claude|pi)(?:\.js)?$/.test(basename(value)));
+      if (native) processes.push({ pid: Number(name), argv, cwd: readlinkSync(join(directory, 'cwd')) });
+    } catch (error) {
+      if (error.code !== 'ENOENT' && error.code !== 'ESRCH') refuse('Native process visibility is incomplete; recovery refused.');
+    }
+  }
+  return processes;
+}
+
+function verify(record) {
+  const unit = record.readUnit(record.unit);
+  const prefix = `torch-${record.projectId}-`;
+  if (typeof record.unit !== 'string' || !record.unit.startsWith(prefix) || !/^[A-Za-z0-9_.-]+\.service$/.test(record.unit)
+    || unit.Id !== record.unit || !/^[a-f0-9]{32}$/.test(unit.InvocationID ?? '')
+    || unit.InvocationID !== record.invocationId
+    || !['failed', 'inactive'].includes(unit.ActiveState)
+    || !['failed', 'dead'].includes(unit.SubState)
+    || unit.MainPID !== '0' || unit.ControlPID !== '0' || unit.ControlGroup !== ''
+    || resolve(unit.WorkingDirectory || '/') !== record.repositoryRoot) {
+    refuse('The selected project service is not a verified stopped invocation.');
+  }
+  // Explicit legacy operator binding, not a claim of cryptographic session provenance.
+  // Exact area token and terminal timestamp must agree with the retained unknown row.
+  if (!(unit.ExecStart ?? '').split(/[\s;]+/).includes(record.identity.areaId)
+    || !Number.isFinite(Date.parse(unit.ExecMainExitTimestamp))
+    || Math.abs(Date.parse(unit.ExecMainExitTimestamp) - Date.parse(record.identity.updatedAt)) > 5000) {
+    refuse('Service area or terminal timestamp does not match the retained unknown identity.');
+  }
+  for (const processRecord of record.readProcesses()) {
+    if (processRecord.argv.includes(record.identity.runtimeSessionId)
+      || processRecord.argv.includes(record.worktree)
+      || resolve(processRecord.cwd) === record.worktree) {
+      refuse('A native process still references the selected session or workspace.');
+    }
+  }
+  return { unit: record.unit, invocationId: unit.InvocationID,
+    terminalAt: new Date(Date.parse(unit.ExecMainExitTimestamp)).toISOString(),
+    outcome: 'unknown', verification: 'owner-selected-legacy-invocation' };
+}
+
+export function issueStoppedExecutorEvidence({ controlPlane, actorId, areaId, unit, invocationId,
+  readUnit = readStoppedUnit, readProcesses = readNativeProcesses, now = () => Date.now() } = {}) {
+  controlPlane.assertOwnerActor(actorId);
+  const identity = controlPlane.getAgent(areaId);
+  if (identity.state !== 'working' || !identity.runtimeSessionId || !identity.updatedAt
+    || !identity.summary?.includes('Runtime executor completion is unknown')) {
+    refuse('Only an exact retained working/unknown identity can use legacy recovery.');
+  }
+  const worktree = controlPlane.manifest.external?.find(entry => entry.type === 'worktree' && entry.area === areaId)?.path;
+  if (!worktree) refuse('No registered workspace exists for this identity.');
+  const record = { identity: Object.freeze({ areaId, state: identity.state, runtime: identity.runtime,
+    runtimeSessionId: identity.runtimeSessionId, updatedAt: identity.updatedAt }),
+  unit, invocationId, projectId: controlPlane.projectId, repositoryRoot: resolve(controlPlane.repositoryRoot),
+  worktree: resolve(worktree), readUnit, readProcesses, now, issuedAt: now() };
+  const facts = verify(record);
+  const evidence = Object.freeze({});
+  issued.set(evidence, record);
+  return { evidence, plan: { areaId, expectedIdentity: record.identity, ...facts,
+    mutationPerformed: false, effect: 'Recover presence to offline; preserve task, session and unknown outcome. No provider launch.' } };
+}
+
+export function consumeStoppedExecutorEvidence(evidence, { projectId, repositoryRoot, identity } = {}) {
+  const record = issued.get(evidence);
+  if (!record) refuse('Recovery requires fresh Runtime-issued evidence, not caller text or a Boolean.');
+  issued.delete(evidence);
+  if (record.projectId !== projectId || record.repositoryRoot !== resolve(repositoryRoot)
+    || record.now() < record.issuedAt
+    || record.now() - record.issuedAt > TTL_MS
+    || ['areaId', 'state', 'runtime', 'runtimeSessionId', 'updatedAt'].some(key => record.identity[key] !== identity[key])) {
+    refuse('Recovery evidence is stale, expired or belongs to another identity.');
+  }
+  const facts = verify(record); // Recheck physical state immediately before the transactional mutation.
+  return { ...facts, digest: createHash('sha256').update(JSON.stringify({ identity: record.identity, ...facts })).digest('hex') };
+}

src/runtime/turn-guard.mjs

@@ -0,0 +1,82 @@
+import { randomUUID } from 'node:crypto';
+import { mkdirSync, readFileSync, readlinkSync, renameSync, rmdirSync, statSync, unlinkSync, writeFileSync } from 'node:fs';
+import { basename, join, resolve } from 'node:path';
+import { TorchError } from '../kernel/errors.mjs';
+
+function nativeProcess(pid) {
+  if (process.platform !== 'linux') throw new Error('Process observation unavailable');
+  const directory = `/proc/${pid}`;
+  const stat = readFileSync(join(directory, 'stat'), 'utf8');
+  const fields = stat.slice(stat.lastIndexOf(')') + 2).split(' ');
+  return { uid: statSync(directory).uid, state: fields[0], startTicks: fields[19],
+    cwd: readlinkSync(join(directory, 'cwd')),
+    argv: readFileSync(join(directory, 'cmdline'), 'utf8').split('\0').filter(Boolean) };
+}
+
+// Observational only: never changes assignments, heartbeat, authority or launch eligibility.
+export function observeRuntimeTurn({ stateRoot, repositoryRoot, projectId, areaId,
+  inspectProcess = nativeProcess, uid = process.getuid?.(), now = () => new Date() } = {}) {
+  if (!/^[a-z0-9][a-z0-9-]*$/.test(areaId)) return { state: 'unknown', reason: 'invalid-identity' };
+  let record;
+  try { record = JSON.parse(readFileSync(join(stateRoot, 'sessions', 'turn-guards', areaId, 'owner.json'), 'utf8')); }
+  catch (error) { return { state: error.code === 'ENOENT' ? 'inactive' : 'unknown', reason: 'guard-unavailable' }; }
+  if (record.projectId !== projectId || record.areaId !== areaId || !Number.isSafeInteger(record.pid)
+    || record.pid < 1 || typeof record.nonce !== 'string' || !record.nonce) {
+    return { state: 'unknown', reason: 'guard-invalid' };
+  }
+  try {
+    const live = inspectProcess(record.pid);
+    if (live.uid !== uid || ['Z', 'X'].includes(live.state)) return { state: 'unknown', reason: 'executor-not-live' };
+    if (record.startTicks) {
+      if (record.startTicks !== live.startTicks) return { state: 'unknown', reason: 'executor-replaced' };
+    } else {
+      // Existing pre-observation guards: require an exact identity-bound TORCH up process.
+      const only = live.argv.indexOf('--only');
+      if (resolve(live.cwd) !== resolve(repositoryRoot) || !live.argv.includes('up')
+        || only < 0 || live.argv[only + 1] !== areaId
+        || !live.argv.some(arg => ['torch.mjs', 'torch'].includes(basename(arg)))) {
+        return { state: 'unknown', reason: 'legacy-executor-unverified' };
+      }
+    }
+    return { state: 'active', phase: record.phase === 'starting' ? 'starting' : 'working',
+      observedAt: now().toISOString(), source: 'live-process-and-turn-guard' };
+  } catch { return { state: 'unknown', reason: 'executor-unobservable' }; }
+}
+
+// One physical executor per identity, including the interval after an agent
+// reports idle but before its harness actually exits. Crashed guards never expire.
+export function withRuntimeTurnGuard(controlPlane, areaId, run) {
+  controlPlane.assertIdentity(areaId);
+  if (!/^[a-z0-9][a-z0-9-]*$/.test(areaId)) throw new TorchError('Invalid runtime guard identity.', { code: 'INVALID_RUNTIME_INPUT' });
+  const parent = join(controlPlane.stateRoot, 'sessions', 'turn-guards');
+  mkdirSync(parent, { recursive: true });
+  const directory = join(parent, areaId);
+  try { mkdirSync(directory); } catch (error) {
+    if (error.code === 'EEXIST') throw new TorchError('An executor already holds this identity turn guard.', {
+      code: 'RUNTIME_TURN_ACTIVE', details: { areaId },
+    });
+    throw error;
+  }
+  const path = join(directory, 'owner.json');
+  const nonce = randomUUID();
+  let startTicks;
+  try { startTicks = nativeProcess(process.pid).startTicks; } catch { /* Unavailable hosts remain observationally unknown. */ }
+  const record = { nonce, pid: process.pid, areaId, projectId: controlPlane.projectId, startTicks, phase: 'starting' };
+  writeFileSync(path, JSON.stringify(record), { flag: 'wx', mode: 0o600 });
+  const setPhase = phase => {
+    if (!['starting', 'working'].includes(phase)) throw new TorchError('Invalid runtime phase.', { code: 'INVALID_RUNTIME_INPUT' });
+    const current = JSON.parse(readFileSync(path, 'utf8'));
+    if (current.nonce !== nonce || current.pid !== process.pid) throw new TorchError('Runtime guard changed.', { code: 'RUNTIME_TURN_ACTIVE' });
+    const temporary = `${path}.${nonce}.tmp`;
+    writeFileSync(temporary, JSON.stringify({ ...record, phase }), { flag: 'wx', mode: 0o600 });
+    renameSync(temporary, path);
+  };
+  try { return run(setPhase); } finally {
+    // Delete only the exact guard created by this invocation; preserve foreign/changed guards.
+    const record = JSON.parse(readFileSync(path, 'utf8'));
+    if (record.nonce === nonce && record.pid === process.pid) {
+      unlinkSync(path);
+      rmdirSync(directory);
+    }
+  }
+}