TASK-33d49b5b: compose schema-v2 owner reply boundary

docs/TEST_INTEGRITY.md

@@ -3063,3 +3063,40 @@ test_integrity_note:
     did_relax_any_assertion: false
     if_true_explain_spec_basis: ""
 ```
+
+## 2026-10-03 — Schema-v2 read-only owner-reply boundary
+
+```yaml
+test_integrity_note:
+  change_type: new_tests
+  scenarios:
+    - id: SCN-read-only-v2-control-plane
+      given: "A canonical schema-v2 fixture, including a write-refusing state directory"
+      when: "A reader opens a normal SQLite read transaction and a mutation is attempted"
+      then: "It returns one coherent snapshot or a typed refusal; no database or sidecar bytes change"
+    - id: SCN-read-only-pinned-correlated-snapshot
+      given: "Two v2 identities in one observable generation"
+      when: "A writer atomically commits their next generation while the reader remains open"
+      then: "The reader observes one generation or refuses, never a mixed pair"
+    - id: SCN-owner-addressed-reply-durability
+      given: "A registered agent has an owner-originated request addressed to it"
+      when: "It sends a dedicated reply and state is reopened"
+      then: "The reply linkage, bounded owner cursor, and actor audit persist"
+    - id: SCN-owner-addressed-reply-refusal
+      given: "Spoofed senders, generic owner recipients, foreign/forged references, or a non-owner reader"
+      when: "They invoke the owner-reply boundary"
+      then: "They refuse without message or audit writes; direct and broadcast delivery remain intact"
+  observable_outcomes:
+    - "Schema-v2 compatibility refusal, stable snapshot rows, byte/sidecar preservation, durable reply linkage, cursor page, and audit event"
+  determinism_controls:
+    - "Hermetic Git/SQLite fixtures, fixed clocks and IDs, explicit transactions, no network, provider, sleep, or retry"
+  anti_cheat_rationale:
+    prevents:
+      - "Migrating or initializing a read-only or unknown schema"
+      - "Returning stale or mixed identity generations"
+      - "Treating owner as a fleet identity or accepting generic recipient bypasses"
+      - "Accepting foreign or forged owner-reply linkage"
+  relaxation:
+    did_relax_any_assertion: false
+    if_true_explain_spec_basis: ""
+```

src/control-plane/service.mjs

@@ -5,7 +5,7 @@ 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';
 
@@ -15,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'));
@@ -26,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`, {
@@ -40,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'),
@@ -113,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,
@@ -123,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,
   };
 }
 
@@ -225,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');
@@ -268,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]));
@@ -285,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}`, {
@@ -308,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));
   }
 
@@ -367,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');
@@ -396,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');
@@ -426,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
@@ -441,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);
@@ -461,6 +777,7 @@ export class ControlPlane {
   }
 
   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])) {
@@ -491,6 +808,7 @@ export class ControlPlane {
   }
 
   reportStatus({ areaId, state, summary, runtime, runtimeSessionId, task } = {}) {
+    this.assertWritable();
     const id = this.assertIdentity(areaId);
     const normalizedState = requiredText(state, 'state');
     if (!PRESENCE_STATES.includes(normalizedState)) {
@@ -517,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 = {
@@ -566,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;
@@ -607,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);
@@ -682,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';
@@ -739,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.'}`;
@@ -763,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,
@@ -781,6 +1105,7 @@ 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', 'runtime.legacy-unknown.recover-stopped', 'runtime.continuation.configure'].includes(operation)) {
       throw new TorchError('This owner audit surface does not authorize the requested operation', {
@@ -834,3 +1159,7 @@ export class ControlPlane {
 export function openControlPlane(options) {
   return new ControlPlane(options);
 }
+
+export function openControlPlaneReadOnly(options) {
+  return new ControlPlane({ ...options, readOnly: true });
+}

test/fleet/owner-addressed-message-replies.test.mjs

@@ -0,0 +1,90 @@
+import assert from 'node:assert/strict';
+import { execFileSync } from 'node:child_process';
+import { mkdtempSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import test from 'node:test';
+import { openControlPlane } from '../../src/control-plane/service.mjs';
+import { analyzeRepository } from '../../src/kernel/analyze.mjs';
+import { proposeDomains } from '../../src/kernel/domains.mjs';
+import { inspectRepository } from '../../src/kernel/git.mjs';
+import { installProject } from '../../src/kernel/install.mjs';
+
+function installedFixture({ projectId = 'owner-reply-fixture', sequencePrefix = 'main' } = {}) {
+  const root = mkdtempSync(join(tmpdir(), 'torch-owner-reply-'));
+  execFileSync('git', ['init', '-b', 'main', root]);
+  execFileSync('git', ['-C', root, 'config', 'user.email', '[email protected]']);
+  execFileSync('git', ['-C', root, 'config', 'user.name', 'TORCH Test']);
+  writeFileSync(join(root, 'index.js'), 'export const fixture = true;\n');
+  execFileSync('git', ['-C', root, 'add', '.']);
+  execFileSync('git', ['-C', root, 'commit', '-m', 'fixture']);
+  const repository = inspectRepository(root);
+  const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) });
+  proposal.review = { status: 'approved', reviewedAt: '2026-10-03T00:00:00Z', reviewedBy: 'fixture-owner', notes: [] };
+  const env = { ...process.env, XDG_DATA_HOME: join(root, '.data') };
+  installProject({ repository, proposal, env, projectId });
+  let sequence = 0;
+  let seconds = 0;
+  return {
+    root, env,
+    worker: proposal.domains.find((domain) => domain.id !== 'qa')?.id ?? proposal.domains[0].id,
+    options: {
+      repositoryRoot: root, env,
+      idFactory: () => `${sequencePrefix}-${++sequence}`,
+      clock: () => new Date(Date.UTC(2026, 9, 3, 12, 0, seconds++)),
+    },
+  };
+}
+
+test('SCN-owner-addressed-reply-durability: a registered agent replies to its owner request with durable attribution', () => {
+  const fixture = installedFixture();
+  let control = openControlPlane(fixture.options);
+  const request = control.sendOwnerRequest({ actorId: 'owner', recipient: fixture.worker, body: 'Report the persisted outcome.' });
+  const first = control.sendOwnerReply({ sender: fixture.worker, requestId: request.id, body: 'The durable outcome is ready.' });
+  const second = control.sendOwnerReply({ sender: fixture.worker, requestId: request.id, body: 'The evidence remains bounded.' });
+  control.close();
+
+  control = openControlPlane(fixture.options);
+  const firstPage = control.readOwnerConversation({ actorId: 'owner', requestId: request.id, limit: 1 });
+  assert.equal(firstPage.request.id, request.id);
+  assert.deepEqual(firstPage.replies.map((message) => message.id), [first.id]);
+  assert.equal(firstPage.replies[0].recipient, 'owner');
+  assert.equal(firstPage.replies[0].references.replyTo, request.id);
+  assert.equal(firstPage.page.truncated, true);
+  const secondPage = control.readOwnerConversation({ actorId: 'owner', requestId: request.id, cursor: firstPage.page.nextCursor, limit: 1 });
+  assert.deepEqual(secondPage.replies.map((message) => message.id), [second.id]);
+  assert.equal(secondPage.page.complete, true);
+  assert.equal(secondPage.page.nextCursor, null);
+  const audit = control.readAudit().find((event) => event.operation === 'message.reply-owner' && event.entityId === first.id);
+  assert.deepEqual(audit && { actorId: audit.actorId, details: audit.details }, { actorId: fixture.worker, details: { requestId: request.id } });
+  control.close();
+});
+
+test('SCN-owner-addressed-reply-refusal: forged owner recipients, senders, and references leave durable state untouched', () => {
+  const fixture = installedFixture();
+  const foreign = installedFixture({ projectId: 'foreign-owner-reply-fixture', sequencePrefix: 'foreign' });
+  const control = openControlPlane(fixture.options);
+  const request = control.sendOwnerRequest({ actorId: 'owner', recipient: fixture.worker, body: 'Respond only from your registered identity.' });
+  const foreignControl = openControlPlane(foreign.options);
+  const foreignRequest = foreignControl.sendOwnerRequest({ actorId: 'owner', recipient: foreign.worker, body: 'Foreign request.' });
+  foreignControl.close();
+  const beforeAudit = control.readAudit().length;
+  const beforeMessages = control.readMessages({ recipient: fixture.worker }).length;
+  const beforeConversation = control.readOwnerConversation({ actorId: 'owner', requestId: request.id, limit: 10 });
+  assert.deepEqual(beforeConversation.replies, []);
+
+  assert.throws(() => control.sendMessage({ sender: fixture.worker, recipient: 'owner', body: 'Bypass the reply boundary.' }), (error) => error.code === 'UNKNOWN_FLEET_IDENTITY');
+  assert.throws(() => control.sendOwnerReply({ sender: 'owner', requestId: request.id, body: 'Spoofed sender.' }), (error) => error.code === 'UNKNOWN_FLEET_IDENTITY');
+  assert.throws(() => control.sendOwnerReply({ sender: fixture.worker, requestId: foreignRequest.id, body: 'Foreign reference.' }), (error) => error.code === 'OWNER_REPLY_REFERENCE_INVALID');
+  assert.throws(() => control.sendOwnerReply({ sender: fixture.worker, requestId: 'forged-message-id', body: 'Forged reference.' }), (error) => error.code === 'OWNER_REPLY_REFERENCE_INVALID');
+  assert.throws(() => control.readOwnerConversation({ actorId: fixture.worker, requestId: request.id, limit: 1 }), (error) => error.code === 'OWNER_AUTHORITY_REQUIRED');
+  assert.equal(control.readAudit().length, beforeAudit);
+  assert.equal(control.readMessages({ recipient: fixture.worker }).length, beforeMessages);
+  const afterConversation = control.readOwnerConversation({ actorId: 'owner', requestId: request.id, limit: 10 });
+  assert.deepEqual(afterConversation.replies, []);
+  assert.deepEqual(afterConversation.page, beforeConversation.page);
+  const direct = control.sendMessage({ sender: 'session-manager', recipient: fixture.worker, body: 'Existing direct delivery.' });
+  const broadcast = control.sendMessage({ sender: fixture.worker, recipient: 'all', body: 'Existing broadcast delivery.' });
+  assert.deepEqual(control.readMessages({ recipient: fixture.worker }).slice(-2).map((message) => message.id), [direct.id, broadcast.id]);
+  control.close();
+});

test/fleet/read-only-control-plane.test.mjs

@@ -0,0 +1,163 @@
+import assert from 'node:assert/strict';
+import { execFileSync } from 'node:child_process';
+import { chmodSync, existsSync, mkdtempSync, readFileSync, readdirSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { basename, join } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
+import test from 'node:test';
+import { openControlPlane, openControlPlaneReadOnly } from '../../src/control-plane/service.mjs';
+import { analyzeRepository } from '../../src/kernel/analyze.mjs';
+import { proposeDomains } from '../../src/kernel/domains.mjs';
+import { inspectRepository } from '../../src/kernel/git.mjs';
+import { installProject } from '../../src/kernel/install.mjs';
+import { projectStatePath } from '../../src/kernel/paths.mjs';
+import { setupProject } from '../../src/kernel/setup.mjs';
+
+const now = () => new Date('2026-10-03T10:00:00Z');
+
+function fixture() {
+  const root = mkdtempSync(join(tmpdir(), 'torch-read-only-control-plane-'));
+  const git = (args) => execFileSync('git', ['-C', root, ...args], {
+    env: { ...process.env, GIT_AUTHOR_DATE: '2026-10-03T10:00:00Z', GIT_COMMITTER_DATE: '2026-10-03T10:00:00Z' },
+  });
+  git(['init', '-b', 'main']);
+  git(['config', 'user.email', '[email protected]']);
+  git(['config', 'user.name', 'TORCH Test']);
+  writeFileSync(join(root, 'app.js'), 'export const value = 1;\n');
+  git(['add', '.']);
+  git(['commit', '-m', 'fixture']);
+  const repository = inspectRepository(root);
+  const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) });
+  proposal.review = { status: 'approved', reviewedAt: now().toISOString(), reviewedBy: 'owner', notes: [] };
+  const env = { ...process.env, XDG_DATA_HOME: mkdtempSync(join(tmpdir(), 'torch-read-only-state-')) };
+  const projectId = 'read-only-fixture';
+  installProject({ repository, proposal, env, projectId });
+  setupProject({ repository: inspectRepository(root), env, parentOverride: join(tmpdir(), `${basename(root)}-trees`), authorized: true });
+  const writer = openControlPlane({ repositoryRoot: root, env, clock: now });
+  const worker = proposal.domains[0].id;
+  writer.reportStatus({ areaId: worker, state: 'idle', summary: 'registered v2 snapshot' });
+  writer.close();
+  const stateRoot = projectStatePath(projectId, env);
+  const stable = new DatabaseSync(join(stateRoot, 'state.db'));
+  stable.exec('PRAGMA journal_mode = DELETE');
+  stable.close();
+  return { root, env, worker, stateRoot };
+}
+
+function sidecars(stateRoot) {
+  return readdirSync(stateRoot).filter((name) => name === 'state.db-wal' || name === 'state.db-shm').sort();
+}
+
+test('SCN-read-only-v2-control-plane: a write-refusing state directory yields a stable v2 snapshot and rejects mutation', (t) => {
+  const context = fixture();
+  const statePath = join(context.stateRoot, 'state.db');
+  const before = readFileSync(statePath);
+  const beforeSidecars = sidecars(context.stateRoot);
+  assert.deepEqual(beforeSidecars, [], 'fixture must prove sidecar-free reader behavior');
+  chmodSync(statePath, 0o444);
+  chmodSync(context.stateRoot, 0o555);
+  t.after(() => {
+    chmodSync(context.stateRoot, 0o755);
+    chmodSync(statePath, 0o644);
+  });
+  const reader = openControlPlaneReadOnly({ repositoryRoot: context.root, env: context.env, clock: now });
+  assert.equal(reader.identity(context.worker).summary, 'registered v2 snapshot');
+  assert.deepEqual(reader.observeMessages({ recipient: context.worker, selection: 'unread', limit: 1 }), {
+    recipient: context.worker,
+    messages: [],
+    observation: { selection: 'unread', requestedLimit: 1, returnedCount: 0, complete: true, truncated: false, pendingUnreadCount: 0 },
+  });
+  assert.throws(
+    () => reader.reportStatus({ areaId: context.worker, state: 'working', summary: 'must not persist' }),
+    (error) => error.code === 'CONTROL_PLANE_READ_ONLY',
+  );
+  reader.close();
+  assert.deepEqual(readFileSync(statePath), before, 'the database bytes remain unchanged');
+  assert.deepEqual(sidecars(context.stateRoot), beforeSidecars, 'the reader creates no WAL or SHM sidecar');
+});
+
+test('SCN-read-only-v2-control-plane-race: a concurrent v2 writer produces one coherent snapshot or a typed refusal', () => {
+  const context = fixture();
+  const statePath = join(context.stateRoot, 'state.db');
+  const wal = new DatabaseSync(statePath);
+  wal.exec('PRAGMA journal_mode = WAL');
+  wal.close();
+  const writer = new DatabaseSync(statePath);
+  writer.exec('BEGIN IMMEDIATE');
+  writer.prepare('UPDATE identities SET summary = ?, updated_at = ? WHERE area_id = ?')
+    .run('writer committed after reader snapshot', '2026-10-03T10:01:00Z', context.worker);
+  let reader;
+  let refusal = null;
+  try { reader = openControlPlaneReadOnly({ repositoryRoot: context.root, env: context.env, clock: now }); } catch (error) { refusal = error; }
+  writer.exec('COMMIT');
+  writer.close();
+  if (refusal) {
+    assert.equal(refusal.code, 'CONTROL_PLANE_READ_ONLY_UNAVAILABLE');
+    return;
+  }
+  const observed = reader.identity(context.worker).summary;
+  assert.ok(observed === 'registered v2 snapshot' || observed === 'writer committed after reader snapshot', `reader must expose one whole SQLite snapshot, received ${observed}`);
+  reader.close();
+});
+
+test('SCN-read-only-pinned-correlated-snapshot: one reader cannot combine generations across an atomic writer commit', () => {
+  const context = fixture();
+  const statePath = join(context.stateRoot, 'state.db');
+  const generationA = '2026-10-03T10:00:00.000Z';
+  const generationB = '2026-10-03T10:01:00.000Z';
+  const correlated = [context.worker, 'session-manager'];
+  const seed = new DatabaseSync(statePath);
+  seed.exec('PRAGMA journal_mode = WAL');
+  const setGeneration = seed.prepare('UPDATE identities SET summary = ?, updated_at = ? WHERE area_id = ?');
+  for (const areaId of correlated) setGeneration.run('generation A', generationA, areaId);
+  seed.close();
+
+  const reader = openControlPlaneReadOnly({ repositoryRoot: context.root, env: context.env, clock: now });
+  const first = reader.identity(context.worker);
+  assert.deepEqual({ summary: first.summary, updatedAt: first.updatedAt }, { summary: 'generation A', updatedAt: generationA });
+  const writer = new DatabaseSync(statePath);
+  writer.exec('BEGIN IMMEDIATE');
+  const update = writer.prepare('UPDATE identities SET summary = ?, updated_at = ? WHERE area_id = ?');
+  for (const areaId of correlated) update.run('generation B', generationB, areaId);
+  writer.exec('COMMIT');
+  writer.close();
+  let second;
+  try { second = reader.identity('session-manager'); } catch (error) { assert.equal(error.code, 'CONTROL_PLANE_READ_ONLY_UNAVAILABLE'); }
+  if (second) {
+    assert.deepEqual({ summary: second.summary, updatedAt: second.updatedAt }, { summary: 'generation A', updatedAt: generationA }, 'the same reader must not combine generations');
+  }
+  reader.close();
+  const fresh = openControlPlaneReadOnly({ repositoryRoot: context.root, env: context.env, clock: now });
+  assert.deepEqual(
+    Object.fromEntries(correlated.map((areaId) => {
+      const identity = fresh.identity(areaId);
+      return [areaId, { summary: identity.summary, updatedAt: identity.updatedAt }];
+    })),
+    Object.fromEntries(correlated.map((areaId) => [areaId, { summary: 'generation B', updatedAt: generationB }])),
+  );
+  fresh.close();
+});
+
+test('SCN-read-only-v2-control-plane-schema: unknown schema revisions are refused instead of guessed', () => {
+  const context = fixture();
+  const statePath = join(context.stateRoot, 'state.db');
+  const writer = new DatabaseSync(statePath);
+  writer.exec('PRAGMA user_version = 3');
+  writer.close();
+  assert.ok(existsSync(statePath));
+  assert.throws(
+    () => openControlPlaneReadOnly({ repositoryRoot: context.root, env: context.env }),
+    (error) => error.code === 'CONTROL_PLANE_READ_ONLY_UNAVAILABLE',
+  );
+});
+
+test('SCN-read-only-v2-control-plane-input-drift: tracked identity inputs changing after snapshot make reads unavailable', () => {
+  const context = fixture();
+  const rosterPath = join(context.root, '.torch', 'roster.yaml');
+  const original = readFileSync(rosterPath, 'utf8');
+  const reader = openControlPlaneReadOnly({ repositoryRoot: context.root, env: context.env });
+  writeFileSync(rosterPath, `${original}\n`);
+  assert.throws(() => reader.identity(context.worker), (error) => error.code === 'CONTROL_PLANE_READ_ONLY_UNAVAILABLE');
+  writeFileSync(rosterPath, original);
+  reader.close();
+});