diff --git a/docs/architecture/rfcs/complete-state-recovery-v0.md b/docs/architecture/rfcs/complete-state-recovery-v0.md
index 4240c01cc7..b3d93fde8e 100644
--- a/docs/architecture/rfcs/complete-state-recovery-v0.md
+++ b/docs/architecture/rfcs/complete-state-recovery-v0.md
@@ -526,7 +526,7 @@ active Goal dependency optional.
 | --- | --- | --- |
 | Registry routing and machine/Goal configuration: [configuration capture adapter](../../../loopx/capabilities/configuration_backup.py), [typed checkpoint envelope](../../../loopx/control_plane/configuration_backup.ts), [machine configuration store](../../../loopx/capabilities/machine_configuration/store.py) | Full backups compare the registry and stored machine-configuration bytes actually decoded by their owners with mapped copied-byte witnesses. Missing/differing copies, A → B → A and a captured absence beside an earlier copied file refuse publication. The private manifest records matching revisions and the configuration capture interval. | Bind authority and every other required owner to their own intervals and dependencies. Configuration-source agreement and a valid envelope do not prove cross-owner consistency. |
 | Goal narrative, vision and authority material references: registry-resolved active state, [vision checkpoint owner](../../../loopx/control_plane/goals/vision_checkpoint.ts) | Discovered active-state paths and declared local `kind=goal_doc` documents are copied through the existing material-root boundary. Hash-only registrations and document links are not recursively resolved. | Preserve the full Goal and revision-bound required artifacts, classify unavailable dependencies, and keep derived Markdown projections subordinate to their owners. |
-| Canonical work, leases and retained operation history: [local provider selection](../../../loopx/control_plane/coordination/local_authority_provider.ts), [authority archive owner](../../../loopx/control_plane/coordination/authority_archive.ts), [legacy writer fence](../../../loopx/control_plane/coordination/legacy_writer_fence.ts) | File head/history share an atomic envelope; SQLite uses the [existing physical snapshot owner](../../../loopx/control_plane/coordination/sqlite_backup.ts). M1 opens existing stores and verifies head/history without selecting a live destination. | Bind provider/store identity, selector, fence, revision and cursor; obtain an owner export or qualified snapshot. Component-local consistency cannot establish cross-owner consistency or make historical leases valid. |
+| Canonical work, leases and retained operation history: [local provider selection](../../../loopx/control_plane/coordination/local_authority_provider.ts), [authority archive owner](../../../loopx/control_plane/coordination/authority_archive.ts), [legacy writer fence](../../../loopx/control_plane/coordination/legacy_writer_fence.ts) | New ordinary full backups reuse the native retained-prefix export, bind configuration digest/runtime route, provider/store identity, revision/cursor and actual interval, and retain selector/fence byte witnesses. Inert inspection reads that export; older v0 archives retain raw File/SQLite readback. | Complete the selected Goal's required-owner closure and establish the declared cross-owner consistency profile. A valid local prefix cannot make historical leases valid or settle other owners. |
 | Quota accounting and Turn settlement: [accounting artifact transaction](../../../loopx/control_plane/quota/accounting_artifact_transaction.ts), [spend owner](../../../loopx/control_plane/quota/spend_commit.ts), [Turn journal commit](../../../loopx/control_plane/turn_driver/turn_journal_effects.ts) | Files under included runtime/project roots are retained; M1 does not reconcile their transaction lineage. | Capture committed receipts, indexes and in-flight journal inputs together with their exact settlement identity; let the original owner resolve replay or ambiguity without a second debit. |
 | Goal lifetime and session/execution binding: [source-session lifetime owner](../../../loopx/control_plane/goals/source_session_lifetime.ts), [Chat session lifecycle](../../../loopx/control_plane/goals/chat_session_lifecycle.ts), [execution identity](../../../loopx/control_plane/runtime/execution_identity.ts) | Existing Goal-scoped registration, recreation and Turn-settlement directories beside the actual registry are retained through their existing path owners. Native Host session stores outside those paths are not implicitly captured or adopted. | Resolve the actual entrypoint's owner, lifetime and generation. Prove stale executor exclusion and a qualified session recovery path before the original entrypoint can continue. |
 | Pending effects, delivery and derived projection outboxes: [source Turn effect gate](../../../loopx/control_plane/goals/source_session_lifetime.ts), [return-delivery classifier](../../../loopx/control_plane/collaboration/return_delivery.ts), [authority shadow outbox](../../../loopx/control_plane/coordination/local_authority_shadow_outbox.ts) | Included journals/outboxes are retained as bytes; neither their external outcome nor their replay eligibility is inferred. | Each owning effect/transport resolves committed, absent or unknown outcomes and exact receipts. Distinguish a derived projection from an external effect; an unknown external result stays held. |
@@ -572,16 +572,27 @@ publication; it does not qualify an `online` profile or the full M2 inventory.
 Retaining the lifecycle journals above does not authorize generic reads of the
 lifecycle-only `source_session_v1` profile, which remains refused.
 
-The next implementation slice belongs to the existing backup/recovery workflow:
-resolve one Goal's required owner closure, capture configuration and canonical
-authority through those owners, and record their source intervals/dependencies
-before any profile can be qualified. Reuse the typed authority export/snapshot
-and configuration envelope; Python remains the filesystem/transport adapter.
-Exercise concurrent writes, a missing required owner and moved dependencies
-through the real producer and inert verifier. Other rows remain explicit holds
-until their owner evidence exists. M2's complete-inventory entry and
-profile-specific exits, M3 admission, and the frozen RPO/RTO criteria stay open;
-this inventory neither adds a parallel coordinator nor changes their criteria.
+The first configuration/canonical pair now has actual owner content: the
+[backup composition](../../../loopx/control_plane/coordination/authority_backup.ts)
+reuses the retained-prefix archive and provider owners. It records each
+canonical capture interval and binds the configuration checkpoint digest,
+captured runtime, selector/fence byte witnesses, store identity, revision and
+cursor. Ordinary concurrent appends preserve the captured prefix; moved
+dependencies or missing authority do not become a successful canonical capture.
+Source-loss File/SQLite readback and standalone native restore/audit exercise
+the saved content, including the original operation receipts. Raw copies remain
+a separate historical safety copy; the extra logical export uses the same
+unchanged capacity bounds. Python owns filesystem/archive transport only.
+
+The next slice must resolve the selected Goal's remaining required-owner
+closure, beginning with quota/Turn settlement and lifetime/effect dependencies
+at the real entrypoint, and obtain their actual content exports with exact
+intervals and reconciliation identities. Do not add another capture-field-only
+stage or infer a common epoch from the configuration/canonical pair. The other
+inventory rows remain explicit holds until their owner evidence exists. M2's
+complete-inventory entry and profile-specific exits, M3 admission, Host
+qualification and the frozen RPO/RTO criteria stay open; this checkpoint neither
+adds a parallel coordinator nor changes their criteria.
 
 ### Implementation ownership and reuse evidence
 
diff --git a/docs/reference/configuration-backup.md b/docs/reference/configuration-backup.md
index 2ecb0e1943..3990eedf3b 100644
--- a/docs/reference/configuration-backup.md
+++ b/docs/reference/configuration-backup.md
@@ -394,6 +394,52 @@ inert inspection by the [recorded maintainer decision](../architecture/rfcs/comp
 Compound worst cases and broader platforms/providers remain unqualified; no
 production SLO or merge authority follows from capacity consent.
 
+### Native authority capture in new full backups
+
+Ordinary full backups now also call the existing authority archive owner for
+each configuration-selected local Goal. The private `execution.authority_captures`
+rows bind the configuration checkpoint digest and captured runtime route to an
+actual capture interval, provider/store identity, revision, cursor and native
+history export. Selector and writer-fence bytes, including confirmed absence,
+travel beside that export with their byte digests. Missing authority is not
+initialized; a dependency or store-lineage change during export leaves that
+Goal's capture unknown. Ordinary concurrent appends may continue: the archive
+owner verifies the complete retained prefix at its captured cursor.
+
+Inspection prefers this owner export when the new rows are present. It verifies
+the full transition/receipt chain and reads the Goal/Todo model through the
+existing typed owners. A damaged or missing native capture stays unknown even
+when the earlier raw store copy is readable. An independent valid native export
+can remain readable when a raw database is unreadable; that database still has
+its own unknown storage finding and is never queried through restored views.
+Older v0 archives without these rows retain their existing raw-store inspection.
+Explicit migration-only backups keep their existing capture contract.
+
+After inert verification, the native artifact is at the row's `archive_path`
+under the new destination's `data/`. Its `archive.archive_sha256` is the native
+archive's chained seal, **not** the file byte checksum. Existing
+`authority-archive verify`, `restore` and `audit` can independently verify,
+restore to a new standalone store, and check every original receipt. Those
+operations grant no Goal, lease, session or effect admission; follow the
+[existing recovery contracts](../architecture/rfcs/complete-state-recovery-v0.md#relationship-to-existing-recovery-contracts)
+and owner cutover rules before live use.
+
+The additional logical export preserves the raw safety copy and consumes the
+same unchanged archive, history and verifier budgets. It is not a shared epoch
+with configuration, quota, session, scheduler, material or external-effect
+owners. Every audit remains `held` / `legacy_best_effort`; M2 profile
+qualification and M3 activation remain open. Rollback is to the earlier producer
+for future backups; already-created artifacts remain private inert data.
+
+新完整备份会通过既有 authority owner 保存各 Goal 的完整历史前缀，并绑定配置摘要、
+来源 runtime、实际捕获区间、provider/store 身份、revision、cursor，以及 selector 和
+writer fence 的原始字节或确认不存在。并发追加可继续；捕获中依赖或存储身份变化、
+来源缺失时保持 unknown，不会初始化来源。新格式优先检查原生导出，损坏时不会用较早
+的 raw copy 冒充成功；旧 v0 和专用迁移备份保留原有规则。导出可由现有
+`authority-archive verify` / `restore` / `audit` 独立核验和恢复到全新隔离 store，
+不会激活 Goal、租约、会话或外部操作。额外历史占用原有容量预算；跨 owner 一致性、
+M2 完整资格和 M3 执行准入仍未完成，报告继续保持 held。
+
 ### Choosing backup inspection or a research checkpoint
 
 Use this Settings entry or `backup-state verify` to read a complete archive when
diff --git a/loopx/control_plane/coordination/authority_backup.ts b/loopx/control_plane/coordination/authority_backup.ts
new file mode 100644
index 0000000000..fa94e66bbb
--- /dev/null
+++ b/loopx/control_plane/coordination/authority_backup.ts
@@ -0,0 +1,166 @@
+/** Full-backup composition of the existing retained-prefix archive owner. */
+import {createHash} from "node:crypto";
+import {constants} from "node:fs";
+import {mkdir, open, rm, writeFile} from "node:fs/promises";
+import {isAbsolute, join} from "node:path";
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject} from "../runtime_decode.ts";
+import {isRecoveryDigest, type RecoveryComponentState} from "../state_recovery_contract.ts";
+import {canonicalAuthoritySha256, requireAuthorityStoreId} from "./authority_store_codec.ts";
+import {exportAuthorityArchive} from "./authority_archive.ts";
+import {withVerifiedAuthorityArchive} from "./authority_archive_read.ts";
+import {canonicalTodoCollection} from "./local_authority_read.ts";
+import {decodeLocalAuthoritySelection, localAuthorityProviderPaths, openRuntimeAuthorityStore,
+  type LocalAuthorityProviderDependencies} from "./local_authority_provider.ts";
+import {decodeLegacyCoordinationWriterFence, legacyCoordinationWriterFencePath} from "./legacy_writer_fence.ts";
+
+const hash = (bytes: Uint8Array) => createHash("sha256").update(bytes).digest("hex");
+export function authorityBackupDirectory(goalId: string): string {
+  return `authority-captures/${hash(Buffer.from(requireAuthorityStoreId(goalId, "goal id")))}`;
+}
+
+/** Small routing dependencies, never live-store discovery or initialization. */
+async function sourceBytes(path: string): Promise<Buffer | null> {
+  let file;
+  try { file = await open(path, constants.O_RDONLY | constants.O_NONBLOCK | (constants.O_NOFOLLOW ?? 0)); }
+  catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; throw error; }
+  try {
+    if (!(await file.stat()).isFile()) throw new Error("authority dependency is not a regular file");
+    const bytes = Buffer.alloc(64 * 1024 + 1);
+    let size = 0;
+    while (size < bytes.length) {
+      const read = await file.read(bytes, size, bytes.length - size, size);
+      if (read.bytesRead === 0) break;
+      size += read.bytesRead;
+    }
+    if (size === bytes.length) throw new Error("authority dependency exceeds 64 KiB");
+    return bytes.subarray(0, size);
+  } finally { await file.close(); }
+}
+const witness = (bytes: Buffer | null) => bytes === null ? null : hash(bytes);
+async function routing(root: string, goalId: string) {
+  const selector = await sourceBytes(localAuthorityProviderPaths(root, goalId).marker);
+  if (selector !== null) decodeLocalAuthoritySelection(JSON.parse(selector.toString("utf8")), goalId);
+  const fence = await sourceBytes(legacyCoordinationWriterFencePath(root, goalId));
+  if (fence !== null && decodeLegacyCoordinationWriterFence(JSON.parse(fence.toString("utf8"))).goal_id !== goalId) {
+    throw new Error("authority fence Goal mismatch");
+  }
+  return {selector, fence, dependencies: {selector_sha256: witness(selector), writer_fence_sha256: witness(fence)}};
+}
+
+/** Retain a coherent native history prefix while ordinary appends may continue.
+ * This is one owner's export, never a cross-owner epoch or admission receipt. */
+export async function captureAuthorityBackup(value: unknown,
+  dependencies: LocalAuthorityProviderDependencies = {}): Promise<JsonObject> {
+  const request = requireJsonObject(value, "authority backup capture");
+  const goalId = requireAuthorityStoreId(request.goal_id, "goal id");
+  if ((request.runtime_root !== null && (typeof request.runtime_root !== "string" || !isAbsolute(request.runtime_root))) ||
+      typeof request.destination !== "string" || !isAbsolute(request.destination)) {
+    throw new Error("authority capture requires absolute source and destination paths");
+  }
+  const started_at = new Date().toISOString();
+  const base = {goal_id: goalId, started_at};
+  const unavailable = () => ({...base, completed_at: new Date().toISOString(), state: "unknown",
+    archive_path: null, reason_code: "authority_capture_unavailable"});
+  if (request.runtime_root === null) return unavailable();
+  const directory = authorityBackupDirectory(goalId);
+  const destination = join(request.destination, directory);
+  // Only remove a directory created by this invocation. Existing destinations
+  // are never adopted, overwritten, or recursively removed on a failed retry.
+  let created = false;
+  try {
+    await mkdir(join(request.destination, "authority-captures"), {mode: 0o700, recursive: true});
+    await mkdir(destination, {mode: 0o700}); created = true;
+    const before = await routing(request.runtime_root, goalId);
+    const selection = before.selector === null ? null
+      : decodeLocalAuthoritySelection(JSON.parse(before.selector.toString("utf8")), goalId);
+    if (selection?.provider === "postgresql") throw new Error("local backup requires the provider's own export");
+    const store = await openRuntimeAuthorityStore(request.runtime_root, goalId, dependencies, {existingOnly: true});
+    const archive = await exportAuthorityArchive(store, goalId, join(destination, "authority.jsonl"));
+    const after = await routing(request.runtime_root, goalId);
+    const identity = await store.storeIdentity();
+    if (canonicalAuthoritySha256(before.dependencies) !== canonicalAuthoritySha256(after.dependencies) ||
+        identity.status !== "available" || identity.store_identity !== archive.source_store_identity ||
+        (selection === null ? archive.source_provider !== "file" :
+          selection.provider !== archive.source_provider || selection.store_identity !== archive.source_store_identity)) {
+      throw new Error("authority routing or lineage changed during capture");
+    }
+    for (const [name, bytes] of [["selector.json", before.selector], ["writer-fence.json", before.fence]] as const) {
+      if (bytes !== null) await writeFile(join(destination, name), bytes, {flag: "wx", mode: 0o600});
+    }
+    return {...base, completed_at: new Date().toISOString(), state: "readable",
+      archive_path: `${directory}/authority.jsonl`, archive: {...archive}, dependencies: before.dependencies};
+  } catch (error) {
+    if (!created) throw error;
+    await rm(destination, {recursive: true, force: true});
+    // The ordinary raw archive remains useful for an unknown/missing owner.
+    // Never silently claim that it is an owner-exported canonical checkpoint.
+    return unavailable();
+  }
+}
+
+/** Validate the optional v0 row without trusting a manifest's observed facts. */
+export function decodeAuthorityBackupCapture(value: unknown): JsonObject {
+  const row = requireJsonObject(value, "authority backup capture");
+  const goalId = requireAuthorityStoreId(row.goal_id, "goal id");
+  const validTime = (time: unknown): time is string => typeof time === "string" && time.length <= 40 && Number.isFinite(Date.parse(time));
+  if (!isRecoveryDigest(row.configuration_sha256) ||
+      !validTime(row.started_at) || !validTime(row.completed_at) || Date.parse(row.started_at) > Date.parse(row.completed_at) ||
+      (row.runtime_archive_path !== null && typeof row.runtime_archive_path !== "string")) {
+    throw new Error("invalid authority capture interval or dependency");
+  }
+  // Reuse the existing recovery component classification; no new state machine.
+  const state = row.state as RecoveryComponentState;
+  if (state === "unknown") {
+    if (row.archive_path !== null || row.reason_code !== "authority_capture_unavailable") throw new Error("invalid unavailable capture");
+    return row;
+  }
+  if (state !== "readable" || row.runtime_archive_path === null || row.archive_path !== `${authorityBackupDirectory(goalId)}/authority.jsonl`) {
+    throw new Error("invalid authority capture artifact");
+  }
+  const archive = requireJsonObject(row.archive, "authority archive summary");
+  const binding = requireJsonObject(row.dependencies, "authority capture dependencies");
+  if (archive.goal_id !== goalId || !isRecoveryDigest(archive.archive_sha256) ||
+      !Object.hasOwn(binding, "selector_sha256") || !Object.hasOwn(binding, "writer_fence_sha256") ||
+      Object.values(binding).some(value => value !== null && !isRecoveryDigest(value))) {
+    throw new Error("invalid authority capture evidence");
+  }
+  return row;
+}
+
+/** Inert consumer of the actual owner export; no live source, store creation,
+ * import, session binding, receipt replay or effect dispatch is involved. */
+export async function inspectAuthorityBackup(dataRoot: string, row: JsonObject): Promise<JsonObject> {
+  const goalId = String(row.goal_id);
+  const directory = join(dataRoot, authorityBackupDirectory(goalId));
+  const dependencies = requireJsonObject(row.dependencies, "authority capture dependencies");
+  for (const [name, key] of [["selector.json", "selector_sha256"], ["writer-fence.json", "writer_fence_sha256"]] as const) {
+    const bytes = await sourceBytes(join(directory, name));
+    if (witness(bytes) !== dependencies[key]) throw new Error("captured authority dependency mismatch");
+    if (bytes !== null) {
+      if (key === "selector_sha256") decodeLocalAuthoritySelection(JSON.parse(bytes.toString("utf8")), goalId);
+      else if (decodeLegacyCoordinationWriterFence(JSON.parse(bytes.toString("utf8"))).goal_id !== goalId) throw new Error("captured fence Goal mismatch");
+    }
+  }
+  const expected = requireJsonObject(row.archive, "authority archive summary");
+  // The archive owner verifies the seal, each ordered transition and receipt,
+  // then supplies its private verified copy to the Goal read-model owner.
+  return withVerifiedAuthorityArchive(join(directory, "authority.jsonl"), String(expected.archive_sha256), async archive => {
+    if (canonicalAuthoritySha256({...archive.summary}) !== canonicalAuthoritySha256(expected)) throw new Error("authority capture summary mismatch");
+    const selector = await sourceBytes(join(directory, "selector.json"));
+    if (selector !== null) {
+      const selection = decodeLocalAuthoritySelection(JSON.parse(selector.toString("utf8")), goalId);
+      if (selection.provider !== archive.summary.source_provider || selection.store_identity !== archive.summary.source_store_identity) {
+        throw new Error("captured selector differs from exported lineage");
+      }
+    } else if (archive.summary.source_provider !== "file") throw new Error("selected authority requires its selector");
+    let head: JsonObject | null = null;
+    for await (const row of archive.transactions()) head = row.projection;
+    if (head === null) throw new Error("authority capture has no head");
+    const collection = canonicalTodoCollection(head, goalId, true);
+    return {provider: archive.summary.source_provider, head_digest: archive.summary.projection_sha256,
+      history_commits: Number(archive.summary.commits), todo_count: collection.projection.todo_ids.length,
+      historical_lease_count: collection.leaseIndex?.lease_todo_ids.length ?? 0,
+      capture_started_at: row.started_at, capture_completed_at: row.completed_at};
+  });
+}
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index dd21cf752f..4697a8e217 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -420,6 +420,7 @@ export function createEffectRuntimeHandlers(
     ["state.recovery.manifest.verify", lazyHandler(() => import("./state_recovery.ts"), ({verifyStateRecoveryManifest}) => verifyStateRecoveryManifest)],
     ["state.recovery.audit.readback", lazyHandler(() => import("./state_recovery.ts"), ({verifySavedStateRecoveryAudit}) => verifySavedStateRecoveryAudit)],
     ["coordination.sqlite_backup.snapshot", lazyHandler(() => import("./coordination/sqlite_backup.ts"), ({snapshotSqliteBackup}) => snapshotSqliteBackup)],
+    ["coordination.authority_backup.capture", lazyHandler(() => import("./coordination/authority_backup.ts"), ({captureAuthorityBackup}) => captureAuthorityBackup)],
     ["coordination.local_authority.new_goal_storage", lazyHandler(() => import("./coordination/local_authority_defaults.ts"), ({manageNewGoalStorage}) => manageNewGoalStorage)],
     ["coordination.local_authority.promotion_review", lazyHandler(() => Promise.all([import("./coordination/local_authority_runtime.ts"), import("./coordination/source_transfer.ts")]), ([{reviewLocalCoordinationAuthorityPromotion}, {withCoordinationSourceTransfer}]) => withCoordinationSourceTransfer("coordination.local_authority.promotion_review", reviewLocalCoordinationAuthorityPromotion))],
     ["coordination.local_authority.promotion_reviewed", lazyHandler(() => import("./coordination/local_authority_runtime.ts"), ({executeReviewedCoordinationPromotion}) => executeReviewedCoordinationPromotion)],
diff --git a/loopx/control_plane/state_recovery.ts b/loopx/control_plane/state_recovery.ts
index 1fa4946657..aaa652d45a 100644
--- a/loopx/control_plane/state_recovery.ts
+++ b/loopx/control_plane/state_recovery.ts
@@ -12,6 +12,7 @@ import {decodeLocalAuthoritySelection, localAuthorityProviderPaths,
 import {sqliteAuthorityRuntime} from "./coordination/sqlite_runtime.ts";
 import {sqliteAuthorityPath} from "./coordination/sqlite_authority_store.ts";
 import {canonicalTodoCollection} from "./coordination/local_authority_read.ts";
+import {decodeAuthorityBackupCapture, inspectAuthorityBackup} from "./coordination/authority_backup.ts";
 import {STATE_RECOVERY_AUDIT_SCHEMA, decodeRecoveryMeasurements, decodeStateRecoveryAudit, isRecoveryDigest, recoveryGoalLabel,
   type RecoveryComponent, type RecoveryComponentState} from "./state_recovery_contract.ts";
 
@@ -60,6 +61,27 @@ function capturedRuntimeRoots(manifest: JsonObject): Map<string, string | null>
   return result;
 }
 
+function authorityCaptures(value: unknown, roots: Map<string, string | null> | null): Map<string, JsonObject> | null {
+  if (value === undefined || value === null) return null;
+  check(Array.isArray(value) && value.length <= MAX_GOALS, "authority_capture_inventory_invalid");
+  const rows = new Map<string, JsonObject>();
+  for (const raw of value) {
+    let row: JsonObject;
+    try { row = decodeAuthorityBackupCapture(raw); }
+    catch { throw new RecoveryInputError("authority_capture_invalid"); }
+    const goalId = String(row.goal_id);
+    check(!rows.has(goalId) && roots !== null && row.runtime_archive_path === (roots.get(goalId) ?? null),
+      "authority_capture_dependency_mismatch");
+    if (row.state === "readable") {
+      const archive = requireJsonObject(row.archive, "authority capture summary");
+      check(typeof archive.commits === "string" && /^[1-9]\d*$/.test(archive.commits) &&
+        BigInt(archive.commits) <= BigInt(MAX_COMMITS), "authority_history_capacity");
+    }
+    rows.set(goalId, row);
+  }
+  return rows;
+}
+
 function externalEvidence(request: JsonObject): {external: JsonObject; execution: JsonObject} {
   const external = requireJsonObject(request.external_manifest, "external backup manifest");
   check(external.schema_version === "loopx_state_backup_v0", "backup_schema_unsupported");
@@ -99,7 +121,7 @@ export function verifyStateRecoveryManifest(value: unknown): JsonObject {
       "backup_schema_unsupported");
     check(canonicalAuthoritySha256(manifestCore(external)) === canonicalAuthoritySha256(manifestCore(embedded)),
       "manifest_disagreement");
-    capturedRuntimeRoots(external);
+    const captures = authorityCaptures(execution.authority_captures, capturedRuntimeRoots(external));
     check(Array.isArray(request.files), "member_inventory_invalid");
     check(Number.isSafeInteger(request.sqlite_member_count) && Number(request.sqlite_member_count) >= 0
       && Number(request.sqlite_member_count) <= MAX_SQLITE_FILES, "sqlite_inventory_capacity");
@@ -111,6 +133,15 @@ export function verifyStateRecoveryManifest(value: unknown): JsonObject {
       "member_inventory_invalid");
       observed.set(file.archive_path, file);
     }
+    for (const row of captures?.values() ?? []) {
+      check(row.configuration_sha256 === (execution.configuration_backup as JsonObject)?.sha256,
+        "authority_capture_dependency_mismatch");
+      if (row.state !== "readable") continue;
+      // The native archive digest is its chained terminal seal, not a byte
+      // checksum. Member witnesses bind bytes; the archive owner checks its seal.
+      check(observed.has(String(row.archive_path)),
+        "authority_capture_member_mismatch");
+    }
     const witnesses = execution.file_members;
     if (witnesses !== undefined) {
       check(Array.isArray(witnesses) && witnesses.length === observed.size, "member_witness_mismatch");
@@ -216,6 +247,7 @@ async function auditOwners(value: unknown): Promise<JsonObject> {
   const started = performance.now();
   const request = requireJsonObject(value, "recovery owner audit");
   const runtimeRoots = capturedRuntimeRoots(request);
+  const captures = authorityCaptures(request.authority_captures, runtimeRoots);
   const measurements = decodeRecoveryMeasurements({
     ...requireJsonObject(request.measurements, "host recovery measurements"), owner_read_ms: 0});
   const basis = requireJsonObject(request.verified, "verified recovery basis");
@@ -282,8 +314,28 @@ async function auditOwners(value: unknown): Promise<JsonObject> {
   if (configuration !== null) {
     const goals = (configuration.data as JsonObject).goals as JsonObject[];
     check(goals.length <= MAX_GOALS, "goal_inventory_capacity");
+    const goalIds = new Set(goals.map(goal => String(goal.goal_id)));
+    check([...captures?.keys() ?? []].every(goalId => goalIds.has(goalId)), "authority_capture_goal_mismatch");
     for (const goal of goals) {
       const goalId = String(goal.goal_id);
+      if (captures !== null) {
+        const id = `goal:${hash(goalId).slice(0, 20)}`;
+        const goalFacts = recoveryGoalLabel(goalId) === null ? {} : {goal_id: goalId};
+        const capture = captures.get(goalId);
+        try {
+          check(capture?.state === "readable", "authority_capture_unavailable");
+          check(capture.configuration_sha256 === configuration.sha256, "authority_capture_dependency_mismatch");
+          const facts = await inspectAuthorityBackup(dataRoot, capture);
+          check(Number(facts.history_commits) <= MAX_COMMITS, "authority_history_capacity");
+          components.push(component(id, "coordination_authority", "readable", "historical_head_and_history_readable",
+            "review_goal_identity_before_adoption", {...goalFacts, ...facts}, ["configuration"]));
+        } catch (error) {
+          if (error instanceof RecoveryInputError && error.code === "authority_history_capacity") throw error;
+          components.push(component(id, "coordination_authority", "unknown", "authority_capture_unavailable",
+            "inspect_captured_authority_with_owner", goalFacts, ["configuration"]));
+        }
+        continue;
+      }
       const root = runtimeRoots === null ? "runtime-root" : runtimeRoots.get(goalId);
       components.push(await inspectGoal(root ? join(dataRoot, root) : null, goalId, ownerReadableSqlite));
     }
diff --git a/loopx/control_plane/state_recovery_io.py b/loopx/control_plane/state_recovery_io.py
index 94ea981898..e859a482b8 100644
--- a/loopx/control_plane/state_recovery_io.py
+++ b/loopx/control_plane/state_recovery_io.py
@@ -379,6 +379,7 @@ def verify_state_backup(
                      "recovery_input_changed")
             audit = _audit_owners({"data_root": str(data), "verified": verified,
                 "registry_discovery": external.get("registry_discovery"), "included": external.get("included"),
+                "authority_captures": external.get("execution", {}).get("authority_captures"),
                 "measurements": {"archive_bytes": archive_size,
                     "regular_file_bytes": copied["logical_bytes"],
                     "archive_processing_ms": (time.monotonic_ns() - started) // 1_000_000},
diff --git a/loopx/state_backup.py b/loopx/state_backup.py
index 09580c7caf..254c2a77c4 100644
--- a/loopx/state_backup.py
+++ b/loopx/state_backup.py
@@ -842,6 +842,26 @@ def execute_state_backup_plan(payload: dict[str, Any]) -> dict[str, Any]:
             info.size, info.mode = len(configuration_bytes), 0o600
             tar.addfile(info, io.BytesIO(configuration_bytes))
             updated["execution"]["configuration_backup"] = verify_configuration_backup(configuration)
+            if tar.record_file_members and "goal_runtime_roots" in payload.get("registry_discovery", {}):
+                # Capture through the authority owner, not by decoding a store
+                # here. Raw roots remain a separate historical safety copy.
+                targets = {item["archive_path"]: item["source_path"] for item in included}
+                roots = payload["registry_discovery"]["goal_runtime_roots"]
+                captures = []
+                for goal in configuration["data"]["goals"]:
+                    goal_id = goal["goal_id"]
+                    root = roots.get(goal_id)
+                    capture = effect_runtime_result("coordination.authority_backup.capture", {
+                        "goal_id": goal_id, "runtime_root": str(_resolved(Path(targets[root]))) if root in targets else None,
+                        "destination": str(staging),
+                    }, timeout=300.0, retry_safe=False)
+                    capture["runtime_archive_path"] = root
+                    capture["configuration_sha256"] = configuration["sha256"]
+                    captures.append(capture)
+                    if capture["state"] == "readable":
+                        directory = str(PurePosixPath(capture["archive_path"]).parent)
+                        tar.add(staging / directory, arcname=directory)
+                updated["execution"]["authority_captures"] = captures
             updated["execution"]["sqlite_snapshots"] = [entry[1] for entry in snapshots.values()]
             # manifest.json is deliberately excluded: it contains this list.
             # The existing external archive checksum witnesses the whole tar.
diff --git a/tests/control_plane/test_backup_authority_capture.py b/tests/control_plane/test_backup_authority_capture.py
new file mode 100644
index 0000000000..8bbf6cb3b2
--- /dev/null
+++ b/tests/control_plane/test_backup_authority_capture.py
@@ -0,0 +1,92 @@
+"""The ordinary producer must retain an owner-exported canonical prefix."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+from loopx.state_backup import build_state_backup_plan, execute_state_backup_plan
+from loopx.control_plane.state_recovery_io import verify_state_backup
+from tests.control_plane.canonical_authority_fixture import promoted_create_fixture
+
+
+pytestmark = pytest.mark.usefixtures("isolated_effect_runtime")
+
+
+def cli(*args: object) -> dict:
+    result = subprocess.run([sys.executable, "-I", "-m", "loopx.cli", "--format", "json", *map(str, args)],
+                            capture_output=True, text=True, check=True, timeout=90)
+    return json.loads(result.stdout)
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_full_backup_retains_native_authority_export_and_reads_it_after_source_loss(tmp_path, provider):
+    registry, runtime, state = promoted_create_fixture(tmp_path, provider=provider)
+    cli("--registry", registry, "--runtime-root", runtime, "todo", "add", "--goal-id", "goal-a",
+        "--role", "agent", "--claimed-by", "agent-a", "--operation-id", "retained-work",
+        "--text", "Preserve unfinished work and its original operation")
+    backup = execute_state_backup_plan(build_state_backup_plan(project=state.parents[3], runtime_root=runtime,
+        registry_path=registry, output_dir=tmp_path / "backups", backup_id="native-capture",
+        include_registry_projects=False, include_automations=False, include_skills=False))
+    [capture] = backup["execution"]["authority_captures"]
+    assert capture["state"] == "readable", capture
+    assert capture["goal_id"] == "goal-a"
+    assert capture["archive"]["source_provider"] == provider
+    assert capture["archive"]["commits"] == "2"
+    assert capture["started_at"] <= capture["completed_at"]
+    assert capture["dependencies"]["writer_fence_sha256"]
+    assert capture["configuration_sha256"] == backup["execution"]["configuration_backup"]["sha256"]
+    assert bool(capture["dependencies"]["selector_sha256"]) is (provider == "sqlite")
+    for source in (runtime, state.parents[3], registry):
+        source.rename(source.with_name(f"lost-{source.name}"))
+    destination = tmp_path / "inert"
+    audit = verify_state_backup(archive=Path(backup["archive_path"]), manifest=Path(backup["manifest_path"]),
+                                destination=destination, execute=True)
+    [authority] = [row for row in audit["components"] if row["owner_kind"] == "coordination_authority"]
+    assert authority["state"] == "readable", authority
+    assert authority["facts"]["todo_count"] == 1
+    assert authority["facts"]["history_commits"] == 2
+    assert authority["facts"]["capture_started_at"] == capture["started_at"]
+    assert authority["facts"]["capture_completed_at"] == capture["completed_at"]
+    assert audit["status"] == "held" and audit["activation_eligible"] is False
+    assert audit["execution_authority_granted"] is False
+    artifact = destination / "data" / capture["archive_path"]
+    verified = cli("authority-archive", "verify", "--archive", artifact)
+    assert verified["archive"] == capture["archive"]
+    restored = cli("authority-archive", "restore", "--goal-id", "goal-a", "--archive", artifact,
+        "--archive-sha256", capture["archive"]["archive_sha256"], "--provider", provider,
+        "--destination", tmp_path / "standalone", "--execute")
+    assert restored["status"] == "restored"
+    checked = cli("authority-archive", "audit", "--goal-id", "goal-a", "--archive", artifact,
+        "--archive-sha256", capture["archive"]["archive_sha256"], "--destination", tmp_path / "standalone")
+    assert checked["audit"]["status"] == "matched"
+    assert checked["audit"]["compared_commits"] == "2"
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_missing_canonical_owner_is_not_replaced_by_the_earlier_raw_copy(tmp_path, monkeypatch, provider):
+    from loopx.capabilities import configuration_backup
+
+    registry, runtime, state = promoted_create_fixture(tmp_path, provider=provider)
+    original_capture = configuration_backup.capture_configuration_backup
+
+    def lose_authority(**kwargs):
+        configuration = original_capture(**kwargs)
+        (runtime / "authority").rename(tmp_path / "lost-authority")
+        return configuration
+
+    monkeypatch.setattr(configuration_backup, "capture_configuration_backup", lose_authority)
+    backup = execute_state_backup_plan(build_state_backup_plan(project=state.parents[3], runtime_root=runtime,
+        registry_path=registry, output_dir=tmp_path / "backups", include_registry_projects=False,
+        include_automations=False, include_skills=False))
+    [capture] = backup["execution"]["authority_captures"]
+    assert capture["state"] == "unknown" and capture["archive_path"] is None
+    assert not (runtime / "authority").exists(), "capture must not initialize a missing authority"
+    audit = verify_state_backup(archive=Path(backup["archive_path"]), manifest=Path(backup["manifest_path"]),
+                                destination=tmp_path / "inert", execute=True)
+    [authority] = [row for row in audit["components"] if row["owner_kind"] == "coordination_authority"]
+    assert authority["state"] == "unknown" and authority["reason_code"] == "authority_capture_unavailable"
+    assert audit["status"] == "held" and audit["execution_authority_granted"] is False
diff --git a/tests/control_plane_ts/authority_backup.test.ts b/tests/control_plane_ts/authority_backup.test.ts
new file mode 100644
index 0000000000..1340325b4f
--- /dev/null
+++ b/tests/control_plane_ts/authority_backup.test.ts
@@ -0,0 +1,104 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {existsSync} from "node:fs";
+import {mkdir, mkdtemp, readFile, rm, writeFile} from "node:fs/promises";
+import {tmpdir} from "node:os";
+import {dirname, join} from "node:path";
+import {authorityBackupDirectory, captureAuthorityBackup} from "../../loopx/control_plane/coordination/authority_backup.ts";
+import {verifyAuthorityArchive} from "../../loopx/control_plane/coordination/authority_archive.ts";
+import {openLocalAuthorityStore, localAuthorityProviderPaths, selectLocalSqliteAuthority} from "../../loopx/control_plane/coordination/local_authority_provider.ts";
+import {legacyCoordinationWriterFencePath, COLD_SOURCE_IMPORT_WRITER_FENCE_SCHEMA} from "../../loopx/control_plane/coordination/legacy_writer_fence.ts";
+import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts";
+
+for (const provider of ["file", "sqlite"] as const) {
+  test(`${provider} owner capture keeps a complete prefix during an actual append`, async () => {
+    const root = await mkdtemp(join(tmpdir(), "loopx-authority-backup-"));
+    try {
+      if (provider === "sqlite") await selectLocalSqliteAuthority(root, "goal-a", true);
+      const store = await openLocalAuthorityStore(root, "goal-a");
+      const first = await store.commitAuthority({expected_provider_revision: null, operation_id: "first",
+        events: [], receipts: [{operation_id: "first", retained: true}], next_projection: {goal_id: "goal-a", value: 1}});
+      assert.equal(first.status, "applied");
+      let appended = false;
+      // Wrap only scheduling of a real provider read. Both commits and the
+      // exporter use the real File/SQLite owner and its retained journal.
+      const interleaved = new Proxy(store, {get(target, key) {
+        if (key === "scanCommitted") return async (...args: Parameters<AuthorityStore["scanCommitted"]>) => {
+          const page = await target.scanCommitted(...args);
+          if (!appended) {
+            appended = true;
+            const second = await target.commitAuthority({expected_provider_revision: first.provider_revision!,
+              operation_id: "second", events: [], receipts: [], next_projection: {goal_id: "goal-a", value: 2}});
+            assert.equal(second.status, "applied");
+          }
+          return page;
+        };
+        const value = Reflect.get(target, key);
+        return typeof value === "function" ? value.bind(target) : value;
+      }});
+      const destination = join(root, "capture");
+      const result = await captureAuthorityBackup({runtime_root: root, goal_id: "goal-a", destination},
+        {createStore: () => interleaved});
+      assert.equal(result.state, "readable");
+      const exported = await verifyAuthorityArchive(join(destination, authorityBackupDirectory("goal-a"), "authority.jsonl"));
+      assert.equal(exported.commits, "1");
+      assert.equal((await store.loadAuthority()).cursor, "2");
+      assert.equal(appended, true);
+    } finally { await rm(root, {recursive: true, force: true}); }
+  });
+}
+
+for (const moved of ["selector", "writer_fence", "identity"] as const) {
+  test(`owner capture refuses a moved ${moved} and keeps the source change`, async () => {
+    const root = await mkdtemp(join(tmpdir(), "loopx-authority-backup-"));
+    try {
+      const store = await openLocalAuthorityStore(root, "goal-a");
+      await store.commitAuthority({expected_provider_revision: null, operation_id: "first",
+        events: [], receipts: [], next_projection: {goal_id: "goal-a"}});
+      const identity = await store.storeIdentity();
+      assert.equal(identity.status, "available");
+      const paths = localAuthorityProviderPaths(root, "goal-a");
+      const changedPath = moved === "selector" ? paths.marker : moved === "writer_fence"
+        ? legacyCoordinationWriterFencePath(root, "goal-a") : join(paths.file, "store-identity");
+      // Parent creation is source-fixture setup, never capture behavior.
+      await mkdir(dirname(changedPath), {recursive: true});
+      let changed = false;
+      const interleaved = new Proxy(store, {get(target, key) {
+        if (key === "scanCommitted") return async (...args: Parameters<AuthorityStore["scanCommitted"]>) => {
+          const page = await target.scanCommitted(...args);
+          if (!changed) {
+            changed = true;
+            await writeFile(changedPath, moved === "identity" ? `file:${"a".repeat(32)}` : JSON.stringify(moved === "selector"
+              ? {schema_version: "loopx_local_authority_provider_v0", goal_id: "goal-a", provider: "file", store_identity: identity.store_identity}
+              : {schema_version: COLD_SOURCE_IMPORT_WRITER_FENCE_SCHEMA, state: "engaged", goal_id: "goal-a",
+                fence_id: "new-fence", import_operation_id: "import", import_plan_sha256: "b".repeat(64)}));
+          }
+          return page;
+        };
+        const value = Reflect.get(target, key);
+        return typeof value === "function" ? value.bind(target) : value;
+      }});
+      const destination = join(root, "capture");
+      const result = await captureAuthorityBackup({runtime_root: root, goal_id: "goal-a", destination},
+        {createStore: () => interleaved});
+      assert.equal(result.state, "unknown");
+      assert.equal(existsSync(join(destination, authorityBackupDirectory("goal-a"))), false);
+      assert.equal(changed, true);
+      assert.ok((await readFile(changedPath)).length);
+    } finally { await rm(root, {recursive: true, force: true}); }
+  });
+}
+
+test("missing authority creates no source store; an existing capture is never removed", async () => {
+  const root = await mkdtemp(join(tmpdir(), "loopx-authority-backup-"));
+  try {
+    const request = {runtime_root: root, goal_id: "goal-a", destination: join(root, "capture")};
+    assert.equal((await captureAuthorityBackup(request)).state, "unknown");
+    assert.equal(existsSync(join(root, "authority")), false);
+    await mkdir(join(request.destination, authorityBackupDirectory("goal-a")), {recursive: true});
+    const existing = join(request.destination, authorityBackupDirectory("goal-a"), "retained");
+    await writeFile(existing, "existing user artifact");
+    await assert.rejects(() => captureAuthorityBackup(request), /EEXIST/);
+    assert.equal(await readFile(existing, "utf8"), "existing user artifact");
+  } finally { await rm(root, {recursive: true, force: true}); }
+});
diff --git a/tests/control_plane_ts/state_recovery.test.ts b/tests/control_plane_ts/state_recovery.test.ts
index af95c97205..bff51941fd 100644
--- a/tests/control_plane_ts/state_recovery.test.ts
+++ b/tests/control_plane_ts/state_recovery.test.ts
@@ -4,6 +4,7 @@ import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/a
 import {createEffectRuntimeHandlers, dispatchEffectRuntimeMethod} from "../../loopx/control_plane/effect_runtime_handlers.ts";
 import {verifyStateRecoveryArchive, verifyStateRecoveryManifest} from "../../loopx/control_plane/state_recovery.ts";
 import {decodeStateRecoveryAudit, STATE_RECOVERY_AUDIT_SCHEMA} from "../../loopx/control_plane/state_recovery_contract.ts";
+import {authorityBackupDirectory} from "../../loopx/control_plane/coordination/authority_backup.ts";
 
 const member = {archive_path: "runtime-root/retained", size_bytes: 1, sha256: "c".repeat(64)};
 function request() {
@@ -68,6 +69,41 @@ test("captured runtime bindings are optional data and must name included inert r
   }
 });
 
+test("native capture metadata cannot change its Goal/configuration dependency or exceed the historical capacity", () => {
+  const original = request();
+  const path = `${authorityBackupDirectory("goal-a")}/authority.jsonl`;
+  const capture = {goal_id: "goal-a", state: "readable", runtime_archive_path: "runtime-root",
+    configuration_sha256: "d".repeat(64), started_at: "2026-10-15T00:00:00.000Z", completed_at: "2026-10-15T00:00:01.000Z",
+    archive_path: path, archive: {goal_id: "goal-a", commits: "1", archive_sha256: "e".repeat(64)},
+    dependencies: {selector_sha256: null, writer_fence_sha256: null}};
+  const artifact = {...member, archive_path: path};
+  original.files = [artifact];
+  for (const manifest of [original.external_manifest, original.embedded_manifest]) {
+    Object.assign(manifest, {registry_discovery: {goal_runtime_roots: {"goal-a": "runtime-root"}},
+      included: [{archive_path: "runtime-root"}]});
+    Object.assign(manifest.execution, {authority_captures: [capture], file_members: [artifact],
+      configuration_backup: {sha256: "d".repeat(64)}});
+  }
+  // The logical archive seal is deliberately different from the member byte
+  // digest: each has its own verifier and must never be substituted for the other.
+  assert.equal(verifyStateRecoveryManifest(original).ok, true);
+  for (const delta of [{runtime_archive_path: null}, {runtime_archive_path: "other"},
+    {configuration_sha256: "f".repeat(64)}, {completed_at: "2026-10-14T00:00:00.000Z"},
+    {archive_path: "../outside"}, {archive: {...capture.archive, commits: "20001"}}]) {
+    const input = structuredClone(original);
+    for (const manifest of [input.external_manifest, input.embedded_manifest]) {
+      Object.assign((manifest.execution as any).authority_captures[0], delta);
+    }
+    assert.equal(verifyStateRecoveryManifest(input).ok, false, JSON.stringify(delta));
+  }
+  const duplicate = structuredClone(original);
+  for (const manifest of [duplicate.external_manifest, duplicate.embedded_manifest]) {
+    (manifest.execution as any).authority_captures.push(capture);
+  }
+  assert.equal(verifyStateRecoveryManifest(duplicate).ok, false);
+  assert.equal(verifyStateRecoveryManifest({...original, files: []}).error_code, "authority_capture_member_mismatch");
+});
+
 function audit() {
   const body = {ok: true, schema_version: STATE_RECOVERY_AUDIT_SCHEMA,
     recovery_id: "e".repeat(64), status: "held", written: true, manifest_verified: true,
diff --git a/tests/test_state_recovery.py b/tests/test_state_recovery.py
index fe3d5209a4..7efaaf1038 100644
--- a/tests/test_state_recovery.py
+++ b/tests/test_state_recovery.py
@@ -143,6 +143,7 @@ def test_legacy_mixed_revision_archive_remains_readable_and_inert(tmp_path, prov
     legacy = json.loads(manifest.read_text())
     legacy["registry_discovery"].pop("goal_runtime_roots")
     execution = legacy["execution"]
+    execution.pop("authority_captures", None)
     execution.pop("configuration_capture")
     execution["configuration_backup"] = verify_configuration_backup(later)
     execution["archive_sha256"] = execution["archive_size_bytes"] = None
@@ -281,6 +282,7 @@ def test_legacy_archive_without_source_bindings_keeps_original_runtime_readback(
         embedded = json.load(source.extractfile("manifest.json"))
     for payload in (external, embedded):
         payload["registry_discovery"].pop("goal_runtime_roots", None)
+        payload["execution"].pop("authority_captures", None)
     manifest.write_text(json.dumps(external))
     _rewrite(archive, manifest, replacements={"manifest.json": json.dumps(embedded).encode()})
     audit = verify_state_backup(archive=archive, manifest=manifest, destination=tmp_path / "inert", execute=True)
@@ -438,7 +440,8 @@ def test_restored_code_remains_unclassified_data_and_is_never_loaded(tmp_path):
     assert remaining["state"] == "unknown" and audit["execution_authority_granted"] is False
 
 
-def test_sqlite_views_are_never_evaluated_as_authority_tables(tmp_path):
+@pytest.mark.parametrize("native_capture", [False, True], ids=["legacy-raw", "native-export"])
+def test_sqlite_views_are_never_evaluated_as_authority_tables(tmp_path, native_capture):
     archive, manifest = _backup(tmp_path, "sqlite")
     with tarfile.open(archive) as source:
         database_member = next(row.name for row in source if row.name.endswith(".sqlite"))
@@ -453,14 +456,43 @@ def test_sqlite_views_are_never_evaluated_as_authority_tables(tmp_path):
     changed = database_path.read_bytes()
     external = json.loads(manifest.read_text())
     for envelope in (external, embedded):
+        if not native_capture:
+            envelope["execution"].pop("authority_captures")
         witness = next(row for row in envelope["execution"]["file_members"] if row["archive_path"] == database_member)
         witness.update(sha256=hashlib.sha256(changed).hexdigest(), size_bytes=len(changed))
     manifest.write_text(json.dumps(external))
     _rewrite(archive, manifest, replacements={database_member: changed, "manifest.json": json.dumps(embedded).encode()})
     result = verify_state_backup(archive=archive, manifest=manifest, destination=tmp_path / "view-inert", execute=True)
     [goal] = [row for row in result["components"] if row["owner_kind"] == "coordination_authority"]
-    assert goal["state"] == "unknown"
-    assert goal["reason_code"] == "sqlite_requires_owner_recovery"
+    if native_capture:
+        # The independent logical export remains readable. The poisoned raw
+        # database still fails its storage owner and is never queried as work.
+        assert goal["state"] == "readable" and goal["facts"]["todo_count"] == 1
+    else:
+        assert goal["state"] == "unknown"
+        assert goal["reason_code"] == "sqlite_requires_owner_recovery"
+    [storage] = [row for row in result["components"] if row["owner_kind"] == "sqlite_storage"]
+    assert storage["state"] == "unknown" and storage["reason_code"] == "sqlite_schema_requires_owner_recovery"
+
+
+@pytest.mark.parametrize("damaged", ["authority.jsonl", "writer-fence.json"])
+def test_native_capture_corruption_stays_unknown_instead_of_falling_back_to_valid_raw_authority(tmp_path, damaged):
+    archive, manifest = _backup(tmp_path, "file")
+    external = json.loads(manifest.read_text())
+    with tarfile.open(archive) as source:
+        embedded = json.load(source.extractfile("manifest.json"))
+    capture = external["execution"]["authority_captures"][0]
+    member = str(Path(capture["archive_path"]).with_name(damaged))
+    changed = b'{"corrupt": true}\n'
+    for envelope in (external, embedded):
+        witness = next(row for row in envelope["execution"]["file_members"] if row["archive_path"] == member)
+        witness.update(sha256=hashlib.sha256(changed).hexdigest(), size_bytes=len(changed))
+    manifest.write_text(json.dumps(external))
+    _rewrite(archive, manifest, replacements={member: changed, "manifest.json": json.dumps(embedded).encode()})
+    result = verify_state_backup(archive=archive, manifest=manifest, destination=tmp_path / "inert", execute=True)
+    [goal] = [row for row in result["components"] if row["owner_kind"] == "coordination_authority"]
+    assert goal["state"] == "unknown" and goal["reason_code"] == "authority_capture_unavailable"
+    assert result["status"] == "held" and result["execution_authority_granted"] is False
 
 
 def test_offline_worker_timeout_does_not_leave_a_copy_or_stop_shared_runtime(tmp_path, monkeypatch):
@@ -676,10 +708,14 @@ def test_symlink_parent_and_occupied_directory_are_not_recovery_targets(tmp_path
     assert (occupied / "keep").read_text() == "unchanged"
 
 
-def test_real_http_upload_verify_reload_and_corrupt_input_keep_live_state_unchanged(tmp_path):
+@pytest.mark.parametrize("provider", [None, "file", "sqlite"], ids=["empty", "file", "sqlite"])
+def test_real_http_upload_verify_reload_and_corrupt_input_keep_live_state_unchanged(tmp_path, provider):
     from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
 
-    archive, manifest = _backup(tmp_path)
+    archive, manifest = _backup(tmp_path, provider)
+    if provider:
+        for name in ("runtime", "project", "registry.json"):
+            (tmp_path / name).rename(tmp_path / f"lost-{name}")
     runtime = tmp_path / "replacement-runtime"
     runtime.mkdir()
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
@@ -707,6 +743,10 @@ def request(path, body=None, *, binary=False):
         status, result = request("/verify", inputs)
         assert status == 200, result
         assert result["execution_authority_granted"] is False
+        if provider:
+            [authority] = [row for row in result["components"] if row["owner_kind"] == "coordination_authority"]
+            assert authority["state"] == "readable" and authority["facts"]["todo_count"] == 1
+            assert authority["facts"]["capture_started_at"]
         assert not server.registry_path.exists()
         status, rows = request("")
         assert status == 200 and rows["items"][0]["checkpoint_ref"] == result["checkpoint_ref"]