diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 86e758bf16..112ac1d6f3 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -500,8 +500,10 @@ qualify that guidance seam; avoided model retries and sustained cost remain open
 The replan obligation outcome policy now lives in
 `work_items/replan_semantics.ts`: required-outcome selection, vision-path and
 terminal consistency matching, and the matching refresh input projection share
-one owner. Python retains progress normalization/novelty and persistence adapters,
-but no longer duplicates obligation-to-outcome matching. This is a bounded rule
+one owner. `work_items/progress_delta.ts` now derives semantic dimensions,
+terminal coverage and evidence/blocker novelty inside that same qualification
+request. Python retains shared observation/fingerprint normalization, history
+codecs and persistence adapters; its delta and terminal-coverage rules are removed. This is a bounded rule
 convergence, not a settlement-writer or store migration. Existing outcome
 characterization precedes the move; the intentional correction is executable
 authoring for all vision triggers, with real bound CLI closeout/readback and
@@ -510,12 +512,24 @@ remain in their existing owners; no new capability, provider or setting is added
 
 Known blocker qualification also consumes the host-projected coverage ledger.
 Changing a Turn/work-item binding or refreshing evidence cannot make the same
-blocker a new replan outcome. Python normalizes the observations and reports
-identity reuse; the TypeScript outcome owner rejects that claim for every
-replan source while retaining independently evidenced vision paths. This
+blocker a new replan outcome. Python normalizes the observations; TypeScript
+compares the complete normalized window and rejects identity reuse for every
+replan source while retaining independently evidenced vision paths. Expanded
+qualification inputs use the existing private snapshot transport without
+truncating old claims. The qualify request count and input-error order remain
+unchanged. Shared codecs retire with refresh writers, replan history/context
+readers, external review and Explore evidence consumers. This
 corrects false advancement admission; it does not add a blocked-wait settlement
 route or certify Goal completion.
 
+The finite refresh/checkpoint CLI and standalone replan qualification reuse the
+existing managed-runtime request scope. It pins only the code revision: source
+and provider reads, canonical writes, error ordering and Turn receipts remain
+live and unchanged. Complete File/SQLite CLI qualification is required alongside
+the native request measurements; fewer fingerprint scans alone are not payoff
+evidence. Keep earlier tail failures and same-code A/A variability separate from
+the current qualification in the PR validation record.
+
 Replan decision context now follows the same boundary: `work_items/replan_context.ts`
 owns scoping, chronology, repetition reduction, result/route diversity and bounded
 coverage projection. Its Python codec reads the existing Goal and compact index,
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 890677c190..4d7cb3a032 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -375,12 +375,24 @@ token／重试成本、source-mode 修复和常规短上下文／replan 完整
 
 Replan 的义务结果规则现收敛到 `work_items/replan_semantics.ts`：接受结果选择、
 vision path／terminal 一致性校验与对应 refresh 输入投影共用同一 owner。
-Python 保留 progress 归一化／新颖性与持久化适配，不再重复义务匹配规则。
+`work_items/progress_delta.ts` 现在在同一资格请求内推导语义维度、终态覆盖及
+证据／blocker 新颖性；删除 Python 的 delta 与终态覆盖规则。Python 保留共享的
+observation／fingerprint 归一化、历史 codec 和持久化适配。TS 对完整的归一化窗口
+比较身份复用；更大的资格输入复用既有私有快照传输，不截断旧声明。更换 Turn／
+work-item 或刷新证据不能让旧 blocker 成为新结果，独立的 Vision 证据路径仍可使用。
+资格请求次数与输入错误顺序不变。共享 codec 随 refresh writer、replan history／
+context reader、external review 与 Explore evidence 消费者一起迁移后退役。
 这是有界规则收敛，不是 settlement writer 或存储迁移。先刻画既有接受语义，
 再修正所有 vision trigger 的可执行写入投影，并验证真实绑定 CLI 闭环、回读及
 资格范围错配反例。Checkpoint 恢复与 in-flight 规则仍由既有边界负责，
 不新增 capability、provider 或设置。
 
+有限的 refresh／checkpoint CLI 与独立 replan 资格判断复用既有 managed-runtime
+request scope。它只固定代码版本；source／provider 读取、canonical 写入、错误顺序与
+Turn 回执仍实时执行，保持原语义。晋升同时要求完整 File／SQLite CLI 资格与原生请求
+测量；减少指纹扫描本身不能证明收益。既往尾部失败、相同代码的 A/A 波动与当前资格
+分别保留在 PR 验证记录中。
+
 交付历史边界将 `classification`、`health_check` 与 `recommended_action` 视为
 叙述文本。它们不能生成或解除 follow-through obligation，不能证明 outcome，也
 不能判定交付规模。例如，`unblocked after dependency update` 不构成 blocker
diff --git a/docs/development/control-plane-course/08-evidence-refresh-and-self-repair.md b/docs/development/control-plane-course/08-evidence-refresh-and-self-repair.md
index 4538de8d05..862f8eb449 100644
--- a/docs/development/control-plane-course/08-evidence-refresh-and-self-repair.md
+++ b/docs/development/control-plane-course/08-evidence-refresh-and-self-repair.md
@@ -699,17 +699,25 @@ work effect
 
 ### 1. Replan 不能靠 classification 或 ACK，必须带 typed semantic delta
 
-`loopx/control_plane/work_items/progress_observation.py` 从显式 enum 与稳定 ID
-推导语义变化，不从 prose 猜测：
+`loopx/control_plane/work_items/progress_delta.ts` 从显式 enum、稳定 ID 与完整
+证据窗口推导语义变化，再由既有 replan owner 判断是否满足当前义务。Python
+facade 保留输入与历史 codec；调用完整资格入口，不能用预计算 outcome 代替：
 
 ```python
-observation = normalize_progress_observation(progress_observation)
-delta = semantic_progress_delta(
-    observation,
-    baseline=obligation.get("progress_baseline"),
+from loopx.control_plane.work_items.progress_observation import semantic_delta_from_writeback
+
+delta = semantic_delta_from_writeback(
+    obligation=obligation,
+    progress_observation=progress_observation,
+    history_runs=newest_first_runs,
 )
+if not delta["accepted"]:
+    print(delta["reason"])
 ```
 
+仅刷新 evidence ID 不构成新语义维度；旧 blocker 不因更换 work item 而成为新结果。
+若 context 已截断 coverage，必须提供完整历史，缺失时资格入口拒绝写回。
+
 成功 receipt 必须携带当前 `obligation_id`，且 outcome 属于该 obligation 的
 `required_any_of`。`new_runnable_successor` 不走 `refresh-state` 的 successor 声明；
 它由带 `--action-kind`、稳定 target 和
diff --git a/loopx/cli_commands/project_lifecycle_refresh_state.py b/loopx/cli_commands/project_lifecycle_refresh_state.py
index 382058d1d0..6947a2712f 100644
--- a/loopx/cli_commands/project_lifecycle_refresh_state.py
+++ b/loopx/cli_commands/project_lifecycle_refresh_state.py
@@ -21,6 +21,7 @@
     runtime_capabilities_for_cli_projection,
 )
 from ..control_plane.capability_hooks import PostWritebackHookRegistration
+from ..control_plane.effect_runtime import effect_runtime_request_scope
 from ..control_plane.goals.goal_vision_policy import (
     GOAL_VISION_ADVANCEMENT_POLICY_CHOICES,
 )
@@ -420,6 +421,7 @@ def register_refresh_state_command(
     )
 
 
+@effect_runtime_request_scope()
 def handle_refresh_state_command(
     args: argparse.Namespace,
     *,
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index d9c1bffec3..c766d39b74 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -786,7 +786,7 @@ export function createEffectRuntimeHandlers(
     ["work_item.context.plan", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({planInteractionWorkContext}) => planInteractionWorkContext)],
     ["work_item.context.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionWorkContext}) => projectInteractionWorkContext)],
     ["work_item.replan_settlement.project", lazyHandler(() => import("./work_items/replan_settlement.ts"), ({projectReplanSettlementContract}) => projectReplanSettlementContract)],
-    ["work_item.replan_semantics.project", lazyHandler(() => import("./work_items/replan_semantics.ts"), ({projectReplanSemantics}) => projectReplanSemantics)],
+    ["work_item.replan_semantics.project", lazyHandler(() => import("./work_items/replan_semantics.ts"), ({projectReplanWriteback}) => projectReplanWriteback)],
     ["work_item.replan_context.project", lazyHandler(() => import("./work_items/replan_context.ts"), ({projectReplanContext}) => projectReplanContext)],
     ["work_item.replan_context.project_snapshot", lazyHandler(() => import("./work_items/replan_context.ts"), ({projectReplanContextSnapshot}) => projectReplanContextSnapshot)],
     ["explore.configuration.resolve", lazyHandler(() => import("./capabilities/explore_configuration.ts"), ({resolveExploreConfiguration}) => resolveExploreConfiguration)],
diff --git a/loopx/control_plane/work_items/progress_delta.ts b/loopx/control_plane/work_items/progress_delta.ts
new file mode 100644
index 0000000000..5957dab699
--- /dev/null
+++ b/loopx/control_plane/work_items/progress_delta.ts
@@ -0,0 +1,113 @@
+import type { JsonObject } from "../effect_program.ts";
+import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import {
+  requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray,
+  requireStringLiteral,
+} from "../runtime_decode.ts";
+import { PROGRESS_RESULT_CLASSES } from "./replan_history.ts";
+
+const PROGRESS_IDENTIFIER = /^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$/;
+
+interface Observation {
+  result: typeof PROGRESS_RESULT_CLASSES[number];
+  fingerprint: string;
+  surface_id?: string;
+  hypothesis_id?: string;
+  probe_kind?: string;
+  blocker_id?: string;
+  coverage_scope_id?: string;
+  coverage_complete?: boolean;
+  evidence: readonly string[];
+}
+
+function identifier(source: JsonObject, field: string): string | undefined {
+  if (!Object.hasOwn(source, field)) return undefined;
+  const identifier = requireNonEmptyString(source[field], field);
+  if (!PROGRESS_IDENTIFIER.test(identifier)) {
+    throw new EffectRuntimeRequestError(`${field} must be a normalized progress identifier`);
+  }
+  return identifier;
+}
+
+function evidenceIdentifiers(source: JsonObject): string[] {
+  if (!Object.hasOwn(source, "evidence_ids")) return [];
+  const identifiers = requireStringArray(source.evidence_ids, "evidence_ids");
+  if (!identifiers.length || identifiers.length > 12 ||
+      identifiers.some(identifier => !PROGRESS_IDENTIFIER.test(identifier))) {
+    throw new EffectRuntimeRequestError("evidence_ids must contain 1-12 normalized progress identifiers");
+  }
+  return identifiers;
+}
+
+function decodeObservation(value: unknown): Observation {
+  const source = requireJsonObject(value, "progress observation");
+  requireStringLiteral(source.schema_version, ["typed_progress_observation_v0"], "progress schema");
+  const fingerprint = requireNonEmptyString(source.fingerprint, "fingerprint");
+  if (!/^[a-f0-9]{16}$/.test(fingerprint)) {
+    throw new EffectRuntimeRequestError("fingerprint must be a normalized progress digest");
+  }
+  identifier(source, "work_item_id");
+  const observation = {
+    result: requireStringLiteral(source.result_class, PROGRESS_RESULT_CLASSES, "result_class"),
+    fingerprint, surface_id: identifier(source, "surface_id"),
+    hypothesis_id: identifier(source, "hypothesis_id"), probe_kind: identifier(source, "probe_kind"),
+    blocker_id: identifier(source, "blocker_id"), coverage_scope_id: identifier(source, "coverage_scope_id"),
+    coverage_complete: Object.hasOwn(source, "coverage_complete")
+      ? requireBoolean(source.coverage_complete, "coverage_complete") : undefined,
+    evidence: evidenceIdentifiers(source),
+  };
+  if (!["work_item_id", "surface_id", "hypothesis_id", "probe_kind", "blocker_id",
+    "coverage_scope_id", "evidence_ids", "coverage_complete"].some(field => Object.hasOwn(source, field))) {
+    throw new EffectRuntimeRequestError("progress observation requires a stable work, coverage or evidence field");
+  }
+  return observation;
+}
+
+function newTerminalCoverage(current: Observation, prior: Observation | null): boolean {
+  if (prior === null || current.result !== prior.result ||
+      current.coverage_scope_id !== prior.coverage_scope_id) return true;
+  return current.result === "exploration_exhausted" && current.coverage_complete !== prior.coverage_complete;
+}
+
+function deltaKinds(current: Observation, prior: Observation | null): string[] {
+  if (!current.evidence.length) return [];
+  if (current.result === "advanced") {
+    const dimensions = [
+      ["surface_id", "new_surface"], ["hypothesis_id", "new_hypothesis"], ["probe_kind", "new_probe_family"],
+    ] as const;
+    return dimensions.filter(([field]) => current[field] && current[field] !== prior?.[field])
+      .map(([, outcome]) => outcome);
+  }
+  if (current.result === "blocked") {
+    return current.blocker_id && current.blocker_id !== prior?.blocker_id ? ["new_concrete_blocker"] : [];
+  }
+  if (!current.coverage_scope_id || !newTerminalCoverage(current, prior)) return [];
+  if (current.result === "exploration_exhausted" && current.coverage_complete === true) {
+    return ["coverage_backed_exploration_exhausted"];
+  }
+  return current.result === "no_followup" ? ["coverage_backed_no_followup"] : [];
+}
+
+/** Python supplies normalized source facts; only this owner qualifies their delta. */
+export function projectProgressDelta(value: unknown): JsonObject {
+  if (value === null) return {accepted: false, reason: "typed progress observation missing"};
+  const source = requireJsonObject(value, "progress_source");
+  const current = decodeObservation(source.current);
+  const prior = source.baseline === null ? null : decodeObservation(source.baseline);
+  if (!Array.isArray(source.window)) throw new EffectRuntimeRequestError("progress window must be an array");
+  const claimed = source.window.map(decodeObservation);
+  const known = prior === null ? claimed : [prior, ...claimed];
+  const knownEvidence = new Set(known.flatMap(observation => observation.evidence));
+  const kinds = deltaKinds(current, prior);
+  return {
+    schema_version: "replan_semantic_delta_v0", accepted: kinds.length > 0, delta_kinds: kinds,
+    evidence_novel: current.evidence.some(evidence => !knownEvidence.has(evidence)),
+    observation_repeated: known.some(observation => observation.fingerprint === current.fingerprint),
+    blocker_repeated: !!current.blocker_id && known.some(observation =>
+      observation.result === "blocked" && observation.blocker_id === current.blocker_id),
+    window_size: claimed.length, observation_fingerprint: current.fingerprint,
+    baseline_fingerprint: prior?.fingerprint ?? null,
+    reason: kinds.length ? "typed observation changes the replan frontier"
+      : "typed observation does not change an accepted semantic dimension",
+  };
+}
diff --git a/loopx/control_plane/work_items/progress_observation.py b/loopx/control_plane/work_items/progress_observation.py
index a6239af891..b1e2becc5b 100644
--- a/loopx/control_plane/work_items/progress_observation.py
+++ b/loopx/control_plane/work_items/progress_observation.py
@@ -200,24 +200,6 @@ def typed_progress_repeat_trigger(
     )
 
 
-def _has_new_terminal_coverage(
-    current: Mapping[str, Any],
-    prior: Mapping[str, Any] | None,
-) -> bool:
-    """Compare terminal coverage semantics without treating proof churn as novelty."""
-
-    if prior is None:
-        return True
-    result_class = current["result_class"]
-    if result_class != prior.get("result_class"):
-        return True
-    if current.get("coverage_scope_id") != prior.get("coverage_scope_id"):
-        return True
-    if result_class == ProgressResultClass.EXPLORATION_EXHAUSTED.value:
-        return current.get("coverage_complete") != prior.get("coverage_complete")
-    return False
-
-
 def _normalized_window(
     window: Iterable[Mapping[str, Any]] | None,
 ) -> list[dict[str, Any]]:
@@ -234,99 +216,6 @@ def _normalized_window(
     return normalized
 
 
-def semantic_progress_delta(
-    observation: Mapping[str, Any] | None,
-    *,
-    baseline: Mapping[str, Any] | None,
-    window: Iterable[Mapping[str, Any]] | None = None,
-) -> dict[str, Any]:
-    """Qualify a typed observation as a replan-closing semantic delta.
-
-    `baseline` is the observation the delta kinds are computed against.
-    `window` lists every typed observation already claimed while the
-    obligation formed; the codec reports novelty facts against the whole
-    window so an outcome owner can refuse a replayed claim.
-    """
-
-    if not isinstance(observation, Mapping):
-        return {"accepted": False, "reason": "typed progress observation missing"}
-    current = normalize_progress_observation(observation)
-    prior = (
-        normalize_progress_observation(baseline)
-        if isinstance(baseline, Mapping)
-        else None
-    )
-    claimed = _normalized_window(window)
-    result_class = current["result_class"]
-    delta_kinds: list[str] = []
-    if result_class == ProgressResultClass.ADVANCED.value:
-        dimension_delta_names = {
-            "surface_id": "new_surface",
-            "hypothesis_id": "new_hypothesis",
-            "probe_kind": "new_probe_family",
-        }
-        if current.get("evidence_ids"):
-            for field, delta_kind in dimension_delta_names.items():
-                value = current.get(field)
-                if value and (prior is None or value != prior.get(field)):
-                    delta_kinds.append(delta_kind)
-    elif result_class == ProgressResultClass.BLOCKED.value:
-        blocker_id = current.get("blocker_id")
-        if (
-            blocker_id
-            and current.get("evidence_ids")
-            and (prior is None or blocker_id != prior.get("blocker_id"))
-        ):
-            delta_kinds.append("new_concrete_blocker")
-    elif result_class == ProgressResultClass.EXPLORATION_EXHAUSTED.value:
-        if (
-            current.get("coverage_complete") is True
-            and current.get("coverage_scope_id")
-            and current.get("evidence_ids")
-            and _has_new_terminal_coverage(current, prior)
-        ):
-            delta_kinds.append("coverage_backed_exploration_exhausted")
-    elif result_class == ProgressResultClass.NO_FOLLOWUP.value:
-        if (
-            current.get("coverage_scope_id")
-            and current.get("evidence_ids")
-            and _has_new_terminal_coverage(current, prior)
-        ):
-            delta_kinds.append("coverage_backed_no_followup")
-    # Novelty facts are computed here against the baseline and every claim in
-    # the obligation window; which obligation sources require them behind a
-    # renamed surface, hypothesis or probe family is decided by the TypeScript
-    # outcome owner (work_item.replan_semantics).
-    known_evidence: set[str] = set(prior.get("evidence_ids") or []) if prior else set()
-    known_fingerprints: set[str] = {prior["fingerprint"]} if prior else set()
-    for item in claimed:
-        known_evidence.update(item.get("evidence_ids") or [])
-        known_fingerprints.add(item["fingerprint"])
-    evidence_novel = bool(set(current.get("evidence_ids") or []) - known_evidence)
-    observation_repeated = current["fingerprint"] in known_fingerprints
-    blocker_repeated = bool(current.get("blocker_id")) and any(
-        item.get("result_class") == ProgressResultClass.BLOCKED.value
-        and item.get("blocker_id") == current.get("blocker_id")
-        for item in ([prior] if prior else []) + claimed
-    )
-    return {
-        "schema_version": "replan_semantic_delta_v0",
-        "accepted": bool(delta_kinds),
-        "delta_kinds": delta_kinds,
-        "evidence_novel": evidence_novel,
-        "observation_repeated": observation_repeated,
-        "blocker_repeated": blocker_repeated,
-        "window_size": len(claimed),
-        "observation_fingerprint": current["fingerprint"],
-        "baseline_fingerprint": prior.get("fingerprint") if prior else None,
-        "reason": (
-            "typed observation changes the replan frontier"
-            if delta_kinds
-            else "typed observation does not change an accepted semantic dimension"
-        ),
-    }
-
-
 def replan_writeback_requirements(
     obligation: Mapping[str, Any],
 ) -> dict[str, Any]:
@@ -477,11 +366,13 @@ def semantic_delta_from_writeback(
         )
     elif context is not None and context.get("coverage_truncated"):
         raise ValueError("replan writeback requires the complete evidence history when coverage is truncated")
-    observation_delta = semantic_progress_delta(
-        progress_observation,
-        baseline=baseline if isinstance(baseline, Mapping) else None,
-        window=claimed,
-    )
+    progress_source = None
+    if isinstance(progress_observation, Mapping):
+        progress_source = {
+            "current": normalize_progress_observation(progress_observation),
+            "baseline": normalize_progress_observation(baseline) if isinstance(baseline, Mapping) else None,
+            "window": _normalized_window(claimed),
+        }
     vision = dict(agent_vision) if isinstance(agent_vision, Mapping) else {}
     if vision:
         vision["state"] = normalize_goal_vision_state(vision.get("state"))
@@ -496,7 +387,7 @@ def semantic_delta_from_writeback(
     try:
         result = project_replan_semantics({
             "operation": "qualify", "obligation": dict(obligation),
-            "observation_delta": observation_delta, "agent_vision": vision,
+            "progress_source": progress_source, "agent_vision": vision,
             **({"stride_review_evidence": source_review} if source_review is not None else {}),
         })
     except EffectRuntimeRejected as exc:
diff --git a/loopx/control_plane/work_items/replan_history.ts b/loopx/control_plane/work_items/replan_history.ts
index 474e225ee5..c5e25af77c 100644
--- a/loopx/control_plane/work_items/replan_history.ts
+++ b/loopx/control_plane/work_items/replan_history.ts
@@ -7,9 +7,13 @@ import {
 } from "../runtime_decode.ts";
 import { projectTodoResumePlanning } from "../todos/resume_planning.ts";
 
+export const PROGRESS_RESULT_CLASSES = [
+  "advanced", "unchanged", "blocked", "exploration_exhausted", "no_followup",
+] as const;
+
 interface Progress {
   fingerprint: string;
-  result: "advanced" | "unchanged" | "blocked" | "exploration_exhausted" | "no_followup";
+  result: typeof PROGRESS_RESULT_CLASSES[number];
   payload: JsonObject;
 }
 interface Monitor {
@@ -102,7 +106,7 @@ function decodeProgress(value: unknown): Progress | null {
   return {
     payload, fingerprint: requireNonEmptyString(payload.fingerprint, "fingerprint"),
     result: requireStringLiteral(payload.result_class,
-      ["advanced", "unchanged", "blocked", "exploration_exhausted", "no_followup"], "result_class"),
+      PROGRESS_RESULT_CLASSES, "result_class"),
   };
 }
 function decode(value: unknown): Request {
diff --git a/loopx/control_plane/work_items/replan_semantics.ts b/loopx/control_plane/work_items/replan_semantics.ts
index 71c83a9bfb..17194ba453 100644
--- a/loopx/control_plane/work_items/replan_semantics.ts
+++ b/loopx/control_plane/work_items/replan_semantics.ts
@@ -4,6 +4,7 @@ import { optionalNonEmptyString, requireJsonObject, requireNonEmptyString } from
 import { visionAuthoringContract } from "../goals/vision_checkpoint.ts";
 import { parseTodoTimestampMicros } from "../runtime_timestamp.ts";
 import {normalizeStrideDecision, normalizeStrideStrategy, STRIDE_DECISION_SCHEMA, strideControlContract, stridePlanningGuidance, strideReviewMismatch} from "./stride_policy.ts";
+import { projectProgressDelta } from "./progress_delta.ts";
 
 const PROGRESS_OUTCOMES = [
   "new_surface", "new_hypothesis", "new_probe_family", "new_runnable_successor",
@@ -341,6 +342,15 @@ function acknowledgeVisionEvidence(request: JsonObject): JsonObject {
     gaps.length > 0 && gaps.every(time => time !== null && time <= acknowledgedAt)};
 }
 
+export function projectReplanWriteback(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "work_item.replan_semantics params");
+  if (request.operation !== "qualify") return projectReplanSemantics(request);
+  if (!Object.hasOwn(request, "progress_source") || Object.hasOwn(request, "observation_delta")) {
+    throw new EffectRuntimeRequestError("qualify requires progress_source, not a precomputed observation_delta");
+  }
+  return projectReplanSemantics({...request, observation_delta: projectProgressDelta(request.progress_source)});
+}
+
 export function projectReplanSemantics(value: unknown): JsonObject {
   const request = requireJsonObject(value, "work_item.replan_semantics params");
   if (request.operation === "lifecycle_source") return lifecycleSource(request);
@@ -379,8 +389,8 @@ export function projectReplanSemantics(value: unknown): JsonObject {
   if (request.operation !== "qualify") {
     throw new EffectRuntimeRequestError("replan semantics operation must be requirements or qualify");
   }
-  // The progress codec computes evidence novelty; this boundary decides which
-  // outcomes discharge this obligation. Vision has already passed prepare.
+  // The native progress owner computes novelty before this outcome kernel.
+  // Vision has already passed prepare.
   const observation = object(request.observation_delta);
   const vision = object(request.agent_vision);
   const patch = object(vision.vision_patch);
diff --git a/loopx/control_plane/work_items/semantic_replan_writeback.py b/loopx/control_plane/work_items/semantic_replan_writeback.py
index 71e34ca4a8..c4a14ba07a 100644
--- a/loopx/control_plane/work_items/semantic_replan_writeback.py
+++ b/loopx/control_plane/work_items/semantic_replan_writeback.py
@@ -32,6 +32,7 @@
     project_todo_lifecycle_settlement_reentry,
 )
 from .progress_observation import guarded_replan_transition_delta, semantic_delta_from_writeback
+from ..effect_runtime import effect_runtime_request_scope
 from .repair_delta import (
     build_repair_delta_contract,
     repair_delta_kinds_have_accountable_progress,
@@ -200,6 +201,7 @@ def _obligation_was_created_by_current_completion(
     )
 
 
+@effect_runtime_request_scope()
 def qualify_replan_writeback(
     *,
     newest_first_runs: list[dict[str, Any]] | None,
@@ -227,6 +229,8 @@ def qualify_replan_writeback(
     The write path deliberately reuses quota's goal-frontier reducer. It does
     not classify maintenance from prose and therefore sees obligations derived
     from run history, agent vision, and final-outcome checkpoints alike.
+    Nested decisions share one runtime code revision for this qualification;
+    source and provider reads remain fresh, without caching decision facts.
     """
 
     safe_agent_id = str(agent_id or "").strip()
diff --git a/tests/control_plane/test_external_progress_review.py b/tests/control_plane/test_external_progress_review.py
index bfffd12c87..8ed2552a56 100644
--- a/tests/control_plane/test_external_progress_review.py
+++ b/tests/control_plane/test_external_progress_review.py
@@ -8,7 +8,6 @@
 )
 from loopx.control_plane.work_items.progress_observation import (
     normalize_progress_observation,
-    semantic_progress_delta,
 )
 from loopx.control_plane.work_items.autonomous_replan_ack import (
     autonomous_replan_ack_recorded,
@@ -109,38 +108,6 @@ def test_two_consecutive_completed_drift_receipts_trigger() -> None:
     ]
 
 
-def test_bound_baseline_rejects_a_repeated_observation_and_accepts_new_evidence() -> None:
-    runs = [run(2, turn="t2"), run(1, turn="t1")]
-    for row in runs:
-        row["progress_observation"]["surface_id"] = "surface-retry"
-        row["progress_observation"]["evidence_ids"] = [f"evidence-{row['generated_at'][-3:-1]}"]
-    result = trigger(runs, [receipt(2, turn="t2"), receipt(1, turn="t1")])
-    assert result is not None
-    baseline = result["progress_baseline"]
-    identical = dict(runs[0]["progress_observation"])
-    assert semantic_progress_delta(identical, baseline=baseline)["accepted"] is False
-    same_hypothesis_new_evidence = {**identical, "evidence_ids": ["evidence-fresh"]}
-    assert semantic_progress_delta(same_hypothesis_new_evidence, baseline=baseline)["accepted"] is False
-    new_hypothesis = {**identical, "hypothesis_id": "hypothesis-next", "evidence_ids": ["evidence-fresh"]}
-    assert semantic_progress_delta(new_hypothesis, baseline=baseline)["accepted"] is True
-    # The codec states evidence novelty as a fact; the outcome owner decides
-    # which obligation sources require it behind a renamed identifier.
-    assert semantic_progress_delta(identical, baseline=baseline)["evidence_novel"] is False
-    assert semantic_progress_delta(same_hypothesis_new_evidence, baseline=baseline)["evidence_novel"] is True
-    renamed_only = {**identical, "hypothesis_id": "hypothesis-renamed"}
-    renamed_delta = semantic_progress_delta(renamed_only, baseline=baseline)
-    assert renamed_delta["delta_kinds"] == ["new_hypothesis"] and renamed_delta["evidence_novel"] is False
-    new_blocker = {
-        "schema_version": "typed_progress_observation_v0",
-        "result_class": "blocked",
-        "blocker_id": "blocker-new",
-        "evidence_ids": ["evidence-blocker"],
-    }
-    assert semantic_progress_delta(new_blocker, baseline=baseline)["accepted"] is True
-    # Without a baseline the identical observation would have passed as new work.
-    assert semantic_progress_delta(identical, baseline=None)["accepted"] is True
-
-
 def test_no_typed_observation_on_the_counted_window_raises_nothing() -> None:
     runs = [run(2, turn="t2"), run(1, turn="t1")]
     for row in runs:
diff --git a/tests/control_plane/test_progress_observation.py b/tests/control_plane/test_progress_observation.py
index 45f10dc159..7d0cf7b2b7 100644
--- a/tests/control_plane/test_progress_observation.py
+++ b/tests/control_plane/test_progress_observation.py
@@ -7,7 +7,6 @@
     build_replan_context,
     normalize_progress_observation,
     semantic_delta_from_writeback,
-    semantic_progress_delta,
     typed_progress_repeat_trigger,
 )
 
@@ -180,68 +179,6 @@ def test_text_changes_cannot_hide_an_equivalent_typed_repeat() -> None:
     assert typed_progress_repeat_trigger(runs, agent_id=AGENT_ID) is not None
 
 
-def test_new_probe_family_is_a_semantic_delta_but_new_evidence_alone_is_not() -> None:
-    baseline = normalize_progress_observation(_observation())
-    new_probe = normalize_progress_observation(
-        _observation(result_class="advanced", probe_kind="probe-permission-graph")
-    )
-    evidence_only = normalize_progress_observation(
-        _observation(
-            result_class="advanced",
-            evidence_ids=["evidence-new-output"],
-        )
-    )
-
-    assert semantic_progress_delta(new_probe, baseline=baseline)["delta_kinds"] == [
-        "new_probe_family"
-    ]
-    assert semantic_progress_delta(evidence_only, baseline=baseline)["accepted"] is False
-    # The codec also states whether any evidence id is absent from the baseline;
-    # obligation sources that require it behind a renamed identifier read it.
-    assert semantic_progress_delta(new_probe, baseline=baseline)["evidence_novel"] is False
-    assert semantic_progress_delta(evidence_only, baseline=baseline)["evidence_novel"] is True
-    assert semantic_progress_delta(new_probe, baseline=None)["evidence_novel"] is True
-    # With an obligation window, novelty is judged against every claim in it
-    # and a replayed claim is reported as such.
-    windowed = semantic_progress_delta(evidence_only, baseline=baseline, window=[evidence_only, new_probe])
-    assert windowed["evidence_novel"] is False
-    assert windowed["observation_repeated"] is True
-    assert windowed["window_size"] == 2
-    fresh = semantic_progress_delta(
-        normalize_progress_observation(_observation(result_class="advanced", evidence_ids=["evidence-unseen"])),
-        baseline=baseline, window=[evidence_only, new_probe],
-    )
-    assert fresh["evidence_novel"] is True and fresh["observation_repeated"] is False
-    # Malformed window entries are ignored rather than failing the writeback.
-    assert semantic_progress_delta(new_probe, baseline=baseline, window=[{"bogus": True}, "text"])["window_size"] == 0
-
-
-def test_repeated_blocker_cannot_close_replan() -> None:
-    baseline = normalize_progress_observation(
-        _observation(
-            result_class="blocked",
-            blocker_id="blocker-permission",
-        )
-    )
-    repeated = normalize_progress_observation(
-        _observation(
-            result_class="blocked",
-            blocker_id="blocker-permission",
-        )
-    )
-    novel = normalize_progress_observation(
-        _observation(
-            result_class="blocked",
-            blocker_id="blocker-missing-runtime",
-        )
-    )
-
-    assert semantic_progress_delta(repeated, baseline=baseline)["accepted"] is False
-    assert semantic_progress_delta(novel, baseline=baseline)["delta_kinds"] == [
-        "new_concrete_blocker"
-    ]
-
-
 @pytest.mark.parametrize("history_source", ["coverage", "window", "trigger_window"])
 @pytest.mark.parametrize("evidence_id", ["evidence-route-map", "evidence-readback"])
 def test_known_blocker_does_not_become_new_under_another_work_item(
@@ -281,128 +218,6 @@ def test_known_blocker_does_not_become_new_under_another_work_item(
     assert novel["satisfying_outcomes"] == ["new_concrete_blocker"]
 
 
-def test_exhaustion_requires_coverage_proof() -> None:
-    incomplete = normalize_progress_observation(
-        _observation(
-            result_class="exploration_exhausted",
-            coverage_scope_id="scope-public-entrypoints",
-            coverage_complete=False,
-        )
-    )
-    complete = normalize_progress_observation(
-        _observation(
-            result_class="exploration_exhausted",
-            coverage_scope_id="scope-public-entrypoints",
-            coverage_complete=True,
-        )
-    )
-
-    assert semantic_progress_delta(incomplete, baseline=None)["accepted"] is False
-    assert semantic_progress_delta(complete, baseline=None)["delta_kinds"] == [
-        "coverage_backed_exploration_exhausted"
-    ]
-    assert semantic_progress_delta(complete, baseline=incomplete)["delta_kinds"] == [
-        "coverage_backed_exploration_exhausted"
-    ]
-
-
-@pytest.mark.parametrize(
-    ("result_class", "terminal_fields", "delta_kind"),
-    [
-        (
-            "exploration_exhausted",
-            {"coverage_complete": True},
-            "coverage_backed_exploration_exhausted",
-        ),
-        (
-            "no_followup",
-            {},
-            "coverage_backed_no_followup",
-        ),
-    ],
-)
-def test_terminal_coverage_requires_a_semantically_new_scope(
-    result_class: str,
-    terminal_fields: dict[str, object],
-    delta_kind: str,
-) -> None:
-    baseline = normalize_progress_observation(
-        _observation(
-            result_class=result_class,
-            coverage_scope_id="scope-public-entrypoints",
-            **terminal_fields,
-        )
-    )
-    repeated = normalize_progress_observation(dict(baseline))
-    evidence_only = normalize_progress_observation(
-        _observation(
-            result_class=result_class,
-            coverage_scope_id="scope-public-entrypoints",
-            evidence_ids=["evidence-new-output"],
-            **terminal_fields,
-        )
-    )
-    new_scope = normalize_progress_observation(
-        _observation(
-            result_class=result_class,
-            coverage_scope_id="scope-public-cli",
-            **terminal_fields,
-        )
-    )
-
-    assert baseline["fingerprint"] == repeated["fingerprint"]
-    assert baseline["fingerprint"] != evidence_only["fingerprint"]
-    assert semantic_progress_delta(repeated, baseline=baseline)["accepted"] is False
-    assert (
-        semantic_progress_delta(evidence_only, baseline=baseline)["accepted"] is False
-    )
-    assert semantic_progress_delta(new_scope, baseline=baseline)["delta_kinds"] == [
-        delta_kind
-    ]
-
-
-def test_terminal_result_class_change_is_a_semantic_delta() -> None:
-    baseline = normalize_progress_observation(
-        _observation(
-            result_class="exploration_exhausted",
-            coverage_scope_id="scope-public-entrypoints",
-            coverage_complete=True,
-        )
-    )
-    no_followup = normalize_progress_observation(
-        _observation(
-            result_class="no_followup",
-            coverage_scope_id="scope-public-entrypoints",
-        )
-    )
-
-    assert semantic_progress_delta(no_followup, baseline=baseline)["delta_kinds"] == [
-        "coverage_backed_no_followup"
-    ]
-
-
-def test_no_followup_ignores_coverage_complete_churn() -> None:
-    baseline = normalize_progress_observation(
-        _observation(
-            result_class="no_followup",
-            coverage_scope_id="scope-public-entrypoints",
-            coverage_complete=False,
-        )
-    )
-    coverage_flag_only = normalize_progress_observation(
-        _observation(
-            result_class="no_followup",
-            coverage_scope_id="scope-public-entrypoints",
-            coverage_complete=True,
-        )
-    )
-
-    assert (
-        semantic_progress_delta(coverage_flag_only, baseline=baseline)["accepted"]
-        is False
-    )
-
-
 @pytest.mark.parametrize(
     "agent_vision",
     [
@@ -505,3 +320,31 @@ def test_host_projects_evidence_context_and_minimal_action_packet() -> None:
         "blocked",
         "no_followup",
     ]
+
+
+@pytest.mark.parametrize("result_class", ["exploration_exhausted", "no_followup"])
+def test_terminal_codec_fingerprint_distinguishes_evidence_changes(result_class: str) -> None:
+    baseline = normalize_progress_observation(_observation(
+        result_class=result_class, coverage_scope_id="scope-public-entrypoints",
+        coverage_complete=True,
+    ))
+    repeated = normalize_progress_observation(dict(baseline))
+    evidence_only = normalize_progress_observation({
+        **baseline, "evidence_ids": ["evidence-new-output"],
+    })
+    assert baseline["fingerprint"] == repeated["fingerprint"]
+    assert baseline["fingerprint"] != evidence_only["fingerprint"]
+
+
+def test_malformed_progress_window_rows_do_not_fail_typed_writeback() -> None:
+    delta = semantic_delta_from_writeback(
+        obligation={
+            "progress_baseline": _observation(),
+            "progress_window": [{"bogus": True}, "text"],
+        },
+        progress_observation=_observation(
+            result_class="advanced", probe_kind="probe-permission-graph",
+        ),
+    )
+    assert delta["accepted"] is True
+    assert delta["satisfying_outcomes"] == ["new_probe_family"]
diff --git a/tests/control_plane/test_replan_context_evidence.py b/tests/control_plane/test_replan_context_evidence.py
index bcd03ff1b0..5a2aaa10eb 100644
--- a/tests/control_plane/test_replan_context_evidence.py
+++ b/tests/control_plane/test_replan_context_evidence.py
@@ -233,13 +233,16 @@ def test_generated_omitted_read_recovers_full_agent_history_before_limit(tmp_pat
     assert not (runtime / "goals/evidence-goal/rollout-event-log.jsonl").exists()
 
 
-def test_truncated_display_cannot_make_an_old_blocker_new():
+@pytest.mark.parametrize("count", [40, 12000])
+def test_truncated_display_cannot_make_an_old_blocker_new(count):
+    from datetime import datetime, timedelta, timezone
     from loopx.control_plane.work_items.progress_observation import semantic_delta_from_writeback
 
-    rows = [_run(generated_at=f"2026-08-19T01:{i:02}:00Z", progress_observation={
+    start = datetime(2026, 8, 19, 1, tzinfo=timezone.utc)
+    rows = [_run(generated_at=(start + timedelta(seconds=i)).isoformat(), progress_observation={
         "schema_version": "typed_progress_observation_v0", "result_class": "blocked",
         "blocker_id": f"blocker-{i}", "evidence_ids": [f"proof-{i}"],
-    }) for i in range(40)]
+    }) for i in range(count)]
     obligation = {"obligation_id": "replan-a", "triggers": [{"kind": "typed_progress_repeat"}],
                   "progress_baseline": rows[-1]["progress_observation"]}
     obligation["replan_context"] = project_replan_context(goal_id="evidence-goal", agent_id="agent-a", runs=rows, obligation=obligation)
diff --git a/tests/control_plane/test_replan_progress_native_cli.py b/tests/control_plane/test_replan_progress_native_cli.py
new file mode 100644
index 0000000000..348741eb0d
--- /dev/null
+++ b/tests/control_plane/test_replan_progress_native_cli.py
@@ -0,0 +1,207 @@
+"""Typed progress must settle only the admitted replan through the real CLI."""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from loopx.control_plane.coordination.canonical_snapshot import read_canonical_snapshot
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.effect_runtime import effect_runtime_result, restart_effect_runtime
+from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
+from loopx.control_plane.work_items.progress_observation import normalize_progress_observation
+
+GOAL = "typed-progress-cli-fixture"
+AGENT = "typed-progress-agent"
+TODO = "todo_original_validation"
+TURN = "turn-typed-progress-review"
+
+
+def make_progress_cli_fixture(root: Path, provider: str, scenario: str) -> dict:
+    """Create synthetic source/history for source and installed-package replay."""
+    project, runtime = root / "project", root / "runtime"
+    project.mkdir()
+    state = project / "ACTIVE_GOAL_STATE.md"
+    state.write_text(
+        "---\nstatus: active\n---\n\n# Goal\n\n## Agent Todo\n\n"
+        "- [ ] [P1] Validate the original artifact.\n"
+        f"  <!-- loopx:todo todo_id={TODO} status=open task_class=advancement_task "
+        f"claimed_by={AGENT} action_kind=validate validation_command=pytest -->\n"
+    )
+    baseline = {
+        "schema_version": "typed_progress_observation_v0", "work_item_id": TODO,
+        "surface_id": "surface-existing", "hypothesis_id": "hypothesis-existing",
+        "probe_kind": "probe-existing", "result_class": "unchanged",
+        "evidence_ids": ["evidence-existing"],
+    }
+    if scenario.startswith("terminal_"):
+        baseline["coverage_scope_id"] = "scope-existing"
+    index = runtime / "goals" / GOAL / "runs" / "index.jsonl"
+    index.parent.mkdir(parents=True)
+    source_json, source_md = index.parent / "source.json", index.parent / "source.md"
+    source_json.write_text('{"fixture":"synthetic-typed-progress"}\n')
+    source_md.write_text("# Synthetic prior source audit\n")
+    rows = [{
+        "classification": "state_refreshed", "agent_id": AGENT,
+        "generated_at": f"2026-08-13T00:0{i}:00Z", "turn_instance_id": f"turn-prior-{i}",
+        "progress_observation": normalize_progress_observation(baseline),
+        "json_path": str(source_json), "markdown_path": str(source_md),
+    } for i in range(2)]
+    if scenario == "known_blocker":
+        rows.insert(0, {
+            **rows[0], "generated_at": "2026-08-12T23:59:00Z", "turn_instance_id": "turn-old-blocker",
+            "progress_observation": normalize_progress_observation({
+                **baseline, "result_class": "blocked", "blocker_id": "blocker-permission",
+            }),
+        })
+    index.write_text("".join(json.dumps(row) + "\n" for row in rows))
+    registry = root / "registry.json"
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{
+        "id": GOAL, "status": "active", "repo": str(project), "state_file": state.name,
+        "domain": "synthetic-typed-progress",
+        "adapter": {"kind": "fixture_connected_delivery_v0", "status": "connected-delivery"},
+        "quota": {"compute": 1.0, "window_hours": 24},
+        "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT]},
+    }]}))
+    todos = parse_active_state_todos(state.read_text(), item_limit=None)["agent_todos"]["items"]
+    initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection(
+        goal_id=GOAL, todos=todos, handoff_mode="soft_claim", leases=[],
+    ), state_path=state, provider=provider)
+    observation = {**baseline, "evidence_ids": ["evidence-current"]}
+    if scenario in {"advanced", "advanced_no_evidence"}:
+        observation.update(result_class="advanced", probe_kind="probe-independent")
+        if scenario == "advanced_no_evidence":
+            observation.pop("evidence_ids")
+    elif scenario == "evidence_only":
+        observation["result_class"] = "advanced"
+    elif scenario == "known_blocker":
+        observation.update(work_item_id="todo_different_work", result_class="blocked", blocker_id="blocker-permission")
+    elif scenario.startswith("terminal_"):
+        observation.update(result_class="no_followup", coverage_scope_id="scope-independent")
+    refresh = [
+        "refresh-state", "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", TODO,
+        "--turn-instance-id", TURN, "--classification", "bounded_replan_progress",
+        "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress",
+        "--no-global-sync", "--suppress-external-sinks",
+    ]
+    for field in ("result_class", "work_item_id", "surface_id", "hypothesis_id", "probe_kind", "blocker_id", "coverage_scope_id"):
+        if field in observation:
+            refresh.extend(["--progress-" + field.replace("_", "-"), observation[field]])
+    if observation.get("evidence_ids"):
+        refresh.extend(["--progress-evidence-id", "evidence-current"])
+    if scenario == "terminal_stop":
+        vision = root / "vision.json"
+        vision.write_text(json.dumps({
+            "goal_id": GOAL, "agent_id": AGENT, "state": "no_followup",
+            "vision_patch": {"acceptance_summary": "The bounded source scope has no remaining follow-up."},
+            "path_delta": {
+                "outcome": "stop", "prior_assumption": "The source scope might contain follow-up work.",
+                "observed_reality": "Independent coverage found no follow-up in this bounded scope.",
+                "stopped": ["Repeat exploration of the covered source scope."],
+                "evidence_refs": ["evidence-current"],
+            },
+        }))
+        refresh.extend(["--agent-vision-json", str(vision)])
+    else:
+        refresh.extend(["--vision-unchanged-reason", "The original validation remains open."])
+    return {"project": project, "runtime": runtime, "registry": registry, "state": state,
+            "index": index, "observation": observation, "refresh": refresh}
+
+
+def call_progress_cli(fixture: dict, *args: str, expected_code: int = 0) -> dict:
+    result = subprocess.run([
+        sys.executable, "-m", "loopx.cli", "--registry", str(fixture["registry"]),
+        "--runtime-root", str(fixture["runtime"]), "--format", "json", *args,
+    ], cwd=fixture["project"], env={**os.environ, "LOOPX_USAGE_PING": "0"},
+        capture_output=True, text=True, timeout=60)
+    assert result.returncode == expected_code, (result.stdout, result.stderr)
+    return json.loads(result.stdout)
+
+
+def _authority(fixture: dict) -> dict:
+    result = read_canonical_snapshot(rpc=effect_runtime_result, runtime_root=str(fixture["runtime"]),
+        goal_id=GOAL, include_leases=True, projection_readback=None, timeout=20)
+    assert result["status"] == "loaded", result
+    return result
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("scenario,expected_outcomes,reason_code", [
+    ("advanced", ["new_probe_family"], None),
+    ("evidence_only", [], None),
+    ("known_blocker", [], "progress_observation_replayed"),
+    ("advanced_no_evidence", [], None),
+    ("terminal_inconsistent", [], "no_followup_vision_path_inconsistent"),
+    ("terminal_stop", ["coverage_backed_no_followup"], None),
+])
+def test_typed_progress_cli_preserves_replan_authority(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
+    scenario: str, expected_outcomes: list[str], reason_code: str | None,
+) -> None:
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    fixture = make_progress_cli_fixture(tmp_path, provider, scenario)
+    try:
+        guard_args = ("quota", "should-run", "--codex-app", "--goal-id", GOAL,
+                      "--agent-id", AGENT, "--turn-instance-id", TURN,
+                      "--scan-path", str(fixture["project"]))
+        initial = call_progress_cli(fixture, *guard_args)
+        assert initial["decision"] == "autonomous_replan_required", initial
+        bound = call_progress_cli(fixture, *guard_args, "--todo-id", TODO)
+        obligation = bound["autonomous_replan_obligation"]
+        assert [trigger["kind"] for trigger in obligation["triggers"]] == ["typed_progress_repeat"]
+        assert bound["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO
+        original = {key: fixture[key].read_bytes() for key in ("registry", "state", "index")}
+        authority = _authority(fixture)
+        result = call_progress_cli(fixture, *fixture["refresh"],
+            expected_code=0 if expected_outcomes else 1)
+        if expected_outcomes:
+            assert result["ok"] is True and result["appended"] is True
+            ack = result["autonomous_replan_ack"]
+            assert ack["recorded"] is True
+            delta = ack["semantic_delta"]
+            fingerprint = hashlib.sha256(json.dumps(
+                {key: value for key, value in fixture["observation"].items() if key != "schema_version"},
+                sort_keys=True, ensure_ascii=True, separators=(",", ":"),
+            ).encode()).hexdigest()[:16]
+            assert delta == {
+                "schema_version": "replan_semantic_delta_v0", "accepted": True,
+                "outcomes": expected_outcomes, "satisfying_outcomes": expected_outcomes,
+                "required_any_of": ["new_surface", "new_hypothesis", "new_probe_family",
+                    "new_runnable_successor", "coverage_backed_exploration_exhausted",
+                    "new_concrete_blocker", "coverage_backed_no_followup"],
+                "observation_fingerprint": fingerprint,
+                "reason": "writeback changes an outcome accepted by this obligation source",
+                "trigger_kinds": ["typed_progress_repeat"], "trigger_checkpoints": [],
+                "obligation_id": obligation["obligation_id"],
+            }
+            persisted = json.loads(Path(result["json_path"]).read_text())
+            assert persisted["autonomous_replan_ack"] == ack
+            assert persisted["progress_observation"]["fingerprint"] == fingerprint
+            compact = json.loads(fixture["index"].read_text().splitlines()[-1])
+            assert compact["autonomous_replan_ack"]["semantic_delta"] == delta
+            restart_effect_runtime(wait_for_exit=True)
+            replay = call_progress_cli(fixture, *fixture["refresh"])
+            assert replay["appended"] is False and replay["idempotent_replay"] is True
+        else:
+            assert result["ok"] is False and result["appended"] is False
+            rejection = result["replan_transition"]
+            assert rejection["obligation_id"] == obligation["obligation_id"]
+            assert rejection["host_action"] == "write_typed_semantic_delta"
+            assert rejection["reason_code"] == reason_code
+            assert rejection["next_cli_actions"]
+            restart_effect_runtime(wait_for_exit=True)
+            assert call_progress_cli(fixture, *fixture["refresh"], expected_code=1) == result
+            assert {key: fixture[key].read_bytes() for key in original} == original
+        assert fixture["registry"].read_bytes() == original["registry"]
+        displayed = parse_active_state_todos(fixture["state"].read_text(), item_limit=None)["agent_todos"]["items"]
+        assert [(todo["todo_id"], todo["status"]) for todo in displayed] == [(TODO, "open")]
+        assert _authority(fixture) == authority
+        assert authority["todos"][0]["status"] == "open"
+    finally:
+        restart_effect_runtime(wait_for_exit=True)
diff --git a/tests/control_plane/test_replan_writeback_projection.py b/tests/control_plane/test_replan_writeback_projection.py
index ecfa7237f7..ae323e4f8f 100644
--- a/tests/control_plane/test_replan_writeback_projection.py
+++ b/tests/control_plane/test_replan_writeback_projection.py
@@ -129,3 +129,33 @@ def test_turn_envelope_preserves_executable_vision_authoring_contract() -> None:
     writeback = source["replan_action_packet"]["writeback_contract"]
     assert "path_delta.evidence_refs" in writeback["required_fields"]
     assert envelope["replan_action_packet"]["writeback_contract"] == writeback
+
+
+@pytest.mark.parametrize("current", [None, [], 42])
+def test_missing_current_does_not_normalize_invalid_baseline_or_window(current):
+    result = semantic_delta_from_writeback(
+        obligation={"progress_baseline": {"result_class": "invalid"},
+                    "progress_window": [{"result_class": "invalid"}]},
+        progress_observation=current,
+    )
+    assert result["accepted"] is False
+    assert result["observation_fingerprint"] is None
+    assert result["outcomes"] == []
+
+
+def test_current_normalization_error_precedes_baseline_normalization():
+    with pytest.raises(ValueError, match="result_class must be one of"):
+        semantic_delta_from_writeback(
+            obligation={"progress_baseline": {"result_class": "advanced", "surface_id": "not an identifier"}},
+            progress_observation={"result_class": "invalid"},
+        )
+
+
+def test_malformed_window_entries_do_not_replace_valid_novel_progress():
+    result = semantic_delta_from_writeback(
+        obligation={"progress_window": [{"result_class": "invalid"}, "untyped"]},
+        progress_observation={"result_class": "advanced", "surface_id": "surface-new",
+                              "evidence_ids": ["evidence-new"]},
+    )
+    assert result["accepted"] is True
+    assert result["satisfying_outcomes"] == ["new_surface"]
diff --git a/tests/control_plane_ts/progress_delta.test.ts b/tests/control_plane_ts/progress_delta.test.ts
new file mode 100644
index 0000000000..8ab41c1f88
--- /dev/null
+++ b/tests/control_plane_ts/progress_delta.test.ts
@@ -0,0 +1,180 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
+import { projectProgressDelta } from "../../loopx/control_plane/work_items/progress_delta.ts";
+import { projectReplanWriteback } from "../../loopx/control_plane/work_items/replan_semantics.ts";
+
+// Digests are opaque codec outputs: the native policy must compare, never recompute them.
+function observation(digest: number, patch: JsonObject = {}): JsonObject {
+  return {schema_version: "typed_progress_observation_v0", work_item_id: "todo-progress",
+    surface_id: "surface-auth", hypothesis_id: "hypothesis-boundary", probe_kind: "probe-route-map",
+    result_class: "unchanged", evidence_ids: ["evidence-route-map"],
+    fingerprint: digest.toString(16).padStart(16, "0"), ...patch};
+}
+function source(current: JsonObject, baseline: JsonObject | null = null, window: JsonObject[] = []): JsonObject {
+  return {current, baseline, window};
+}
+function unproved(observation: JsonObject): JsonObject {
+  const {evidence_ids, ...withoutEvidence} = observation;
+  return withoutEvidence;
+}
+const baseline = observation(1);
+
+test("missing progress remains a two-field refusal; new dimensions have stable output order", () => {
+  assert.deepEqual(projectProgressDelta(null), {accepted: false, reason: "typed progress observation missing"});
+  const current = observation(2, {result_class: "advanced"});
+  assert.deepEqual(projectProgressDelta(source(current)), {
+    schema_version: "replan_semantic_delta_v0", accepted: true,
+    delta_kinds: ["new_surface", "new_hypothesis", "new_probe_family"],
+    evidence_novel: true, observation_repeated: false, blocker_repeated: false,
+    window_size: 0, observation_fingerprint: current.fingerprint, baseline_fingerprint: null,
+    reason: "typed observation changes the replan frontier",
+  });
+});
+
+test("probe novelty and evidence novelty remain distinct across the complete obligation window", () => {
+  const probe = observation(2, {result_class: "advanced", probe_kind: "probe-permission-graph"});
+  const evidence = observation(3, {result_class: "advanced", evidence_ids: ["evidence-new-output"]});
+  assert.deepEqual(projectProgressDelta(source(probe, baseline)).delta_kinds, ["new_probe_family"]);
+  assert.equal(projectProgressDelta(source(probe, baseline)).evidence_novel, false);
+  const evidenceOnly = projectProgressDelta(source(evidence, baseline));
+  assert.equal(evidenceOnly.accepted, false);
+  assert.equal(evidenceOnly.evidence_novel, true);
+  assert.equal(evidenceOnly.reason, "typed observation does not change an accepted semantic dimension");
+  assert.equal(projectProgressDelta(source(probe)).evidence_novel, true);
+  const claimed = [evidence, probe, evidence];
+  const replayed = projectProgressDelta(source(evidence, baseline, claimed));
+  assert.equal(replayed.evidence_novel, false);
+  assert.equal(replayed.observation_repeated, true);
+  assert.equal(replayed.window_size, 3, "duplicate window rows still count");
+  assert.equal(replayed.baseline_fingerprint, baseline.fingerprint);
+  const fresh = observation(4, {result_class: "advanced", evidence_ids: ["evidence-unseen"]});
+  const novel = projectProgressDelta(source(fresh, baseline, claimed));
+  assert.equal(novel.evidence_novel, true);
+  assert.equal(novel.observation_repeated, false);
+  assert.deepEqual(claimed, [evidence, probe, evidence], "projection preserves source ordering");
+});
+
+test("the bound external-review baseline distinguishes identity changes, proof churn and replay", () => {
+  const prior = observation(1, {result_class: "advanced", hypothesis_id: "hypothesis-2"});
+  const proof = {...prior, fingerprint: "0000000000000002", evidence_ids: ["evidence-fresh"]};
+  const hypothesis = {...proof, fingerprint: "0000000000000003", hypothesis_id: "hypothesis-next"};
+  const renamed = {...prior, fingerprint: "0000000000000004", hypothesis_id: "hypothesis-renamed"};
+  assert.equal(projectProgressDelta(source(prior, prior)).accepted, false);
+  assert.equal(projectProgressDelta(source(prior, prior)).evidence_novel, false);
+  assert.equal(projectProgressDelta(source(proof, prior)).accepted, false);
+  assert.equal(projectProgressDelta(source(proof, prior)).evidence_novel, true);
+  assert.equal(projectProgressDelta(source(hypothesis, prior)).accepted, true);
+  const renamedDelta = projectProgressDelta(source(renamed, prior));
+  assert.deepEqual(renamedDelta.delta_kinds, ["new_hypothesis"]);
+  assert.equal(renamedDelta.evidence_novel, false);
+  assert.equal(projectProgressDelta(source(prior)).accepted, true);
+});
+
+test("concrete blockers require evidence and a changed identifier; only historical blocked results repeat", () => {
+  const prior = observation(1, {result_class: "blocked", blocker_id: "blocker-permission"});
+  const current = observation(2, {result_class: "blocked", blocker_id: "blocker-missing-runtime"});
+  assert.equal(projectProgressDelta(source(prior, prior)).accepted, false);
+  assert.deepEqual(projectProgressDelta(source(current, prior)).delta_kinds, ["new_concrete_blocker"]);
+  assert.equal(projectProgressDelta(source(unproved(current), prior)).accepted, false);
+  assert.equal(projectProgressDelta(source(current, baseline, [current])).blocker_repeated, true);
+  assert.equal(projectProgressDelta(source(current, {...current, result_class: "advanced"})).blocker_repeated, false);
+  assert.equal(projectProgressDelta(source(current, null, [{...current, result_class: "unchanged"}])).blocker_repeated, false);
+  const externalBaseline = observation(3, {result_class: "advanced"});
+  assert.equal(projectProgressDelta(source(current, externalBaseline)).accepted, true);
+});
+
+test("exhaustion requires complete coverage, and completing prior coverage changes its semantics", () => {
+  const incomplete = observation(1, {result_class: "exploration_exhausted",
+    coverage_scope_id: "scope-public-entrypoints", coverage_complete: false});
+  const complete = {...incomplete, fingerprint: "0000000000000002", coverage_complete: true};
+  assert.equal(projectProgressDelta(source(incomplete)).accepted, false);
+  for (const prior of [null, incomplete]) {
+    assert.deepEqual(projectProgressDelta(source(complete, prior)).delta_kinds,
+      ["coverage_backed_exploration_exhausted"]);
+  }
+});
+
+for (const [result, fields, kind] of [
+  ["exploration_exhausted", {coverage_complete: true}, "coverage_backed_exploration_exhausted"],
+  ["no_followup", {}, "coverage_backed_no_followup"],
+] as const) {
+  test(`${result} requires a new coverage scope rather than refreshed evidence`, () => {
+    const prior = observation(1, {result_class: result, coverage_scope_id: "scope-public-entrypoints", ...fields});
+    const proof = {...prior, fingerprint: "0000000000000002", evidence_ids: ["evidence-new-output"]};
+    const scope = {...prior, fingerprint: "0000000000000003", coverage_scope_id: "scope-public-cli"};
+    assert.equal(projectProgressDelta(source({...prior}, prior)).accepted, false);
+    assert.equal(projectProgressDelta(source(proof, prior)).accepted, false);
+    assert.deepEqual(projectProgressDelta(source(scope, prior)).delta_kinds, [kind]);
+    assert.equal(projectProgressDelta(source(unproved(scope), prior)).accepted, false);
+    assert.equal(projectProgressDelta(source(observation(4, {result_class: result, ...fields}), prior)).accepted, false);
+  });
+}
+
+test("terminal class changes count, while no-followup ignores complete-flag churn", () => {
+  const exhausted = observation(1, {result_class: "exploration_exhausted",
+    coverage_scope_id: "scope-public-entrypoints", coverage_complete: true});
+  const terminal = observation(2, {result_class: "no_followup", coverage_scope_id: "scope-public-entrypoints"});
+  assert.deepEqual(projectProgressDelta(source(terminal, exhausted)).delta_kinds, ["coverage_backed_no_followup"]);
+  assert.equal(projectProgressDelta(source({...terminal, coverage_complete: true},
+    {...terminal, coverage_complete: false})).accepted, false);
+});
+
+test("the effect entrypoint consumes normalized source and refuses caller-supplied outcomes", () => {
+  const obligation = {triggers: [{kind: "external_progress_review_drift"}]};
+  const qualify = (progress_source: JsonObject | null, patch: JsonObject = {}) =>
+    projectReplanWriteback({operation: "qualify", obligation, progress_source, ...patch});
+  const renamed = observation(2, {result_class: "advanced", hypothesis_id: "hypothesis-renamed"});
+  assert.equal(qualify(null).accepted, false);
+  assert.equal(qualify(source(renamed, baseline)).reason_code, "progress_identity_without_new_evidence");
+  const fresh = {...renamed, fingerprint: "0000000000000003", evidence_ids: ["evidence-new"]};
+  const accepted = qualify(source(fresh, baseline));
+  assert.equal(accepted.accepted, true);
+  assert.deepEqual(accepted.satisfying_outcomes, ["new_hypothesis"]);
+  assert.equal(accepted.observation_fingerprint, fresh.fingerprint);
+  assert.equal(qualify(source(fresh, baseline, [fresh])).reason_code, "progress_observation_replayed");
+  assert.throws(() => qualify(null, {observation_delta: {delta_kinds: ["new_hypothesis"]}}), /precomputed/);
+  assert.throws(() => projectReplanWriteback({operation: "qualify", obligation}), /progress_source/);
+  assert.deepEqual(projectReplanWriteback({operation: "requirements", obligation}).required_any_of,
+    ["fresh_vision_path_outcome", "new_surface", "new_hypothesis", "new_probe_family", "new_runnable_successor",
+      "coverage_backed_exploration_exhausted", "new_concrete_blocker", "coverage_backed_no_followup"]);
+});
+
+test("the native source boundary rejects malformed normalized facts", () => {
+  for (const patch of [{schema_version: "legacy"}, {result_class: "looks_advanced"}, {fingerprint: "raw-text"},
+    {hypothesis_id: "unsupported prose"}, {coverage_complete: "true"}, {evidence_ids: [1]},
+    {work_item_id: "unsupported prose"}, {work_item_id: "w".repeat(129)}]) {
+    assert.throws(() => projectProgressDelta(source({...baseline, ...patch})), /progress|result_class|fingerprint|hypothesis|coverage|evidence|work_item/);
+  }
+  for (const invalid of [{current: baseline, baseline: null}, {current: baseline, window: []},
+    {current: null, baseline: null, window: []}]) {
+    assert.throws(() => projectProgressDelta(invalid), /progress/);
+  }
+});
+
+for (const [label, evidence] of [
+  ["empty list", []], ["empty identifier", [""]], ["prose identifier", ["look at the same route"]],
+  ["more than twelve identifiers", Array.from({length: 13}, (_, index) => `evidence-${index}`)],
+] as const) {
+  test(`the actual writeback entrypoint rejects ${label} evidence`, () => {
+    const current = observation(2, {result_class: "advanced", hypothesis_id: "hypothesis-new", evidence_ids: evidence});
+    assert.throws(() => projectReplanWriteback({operation: "qualify",
+      obligation: {triggers: [{kind: "external_progress_review_drift"}]},
+      progress_source: source(current, baseline)}), /evidence_ids/);
+  });
+}
+
+test("fresh Vision cannot bypass missing attribution, while a false coverage flag remains compatible", () => {
+  const carrier = {schema_version: "typed_progress_observation_v0", result_class: "unchanged",
+    fingerprint: "0000000000000001"};
+  const qualify = (current: JsonObject) => projectReplanWriteback({operation: "qualify",
+    obligation: {triggers: [{kind: "external_progress_review_drift"}]}, progress_source: source(current),
+    agent_vision: {vision_patch: {acceptance_summary: "Retain the evidenced path"},
+      path_delta: {outcome: "continue", evidence_refs: ["evidence-path"]}}});
+  for (const current of [carrier, {...carrier, unknown_identity: "todo-progress"}]) {
+    assert.throws(() => qualify(current), /stable work, coverage or evidence field/);
+  }
+  const compatible = qualify({...carrier, coverage_complete: false});
+  assert.equal(compatible.accepted, true);
+  assert.deepEqual(compatible.satisfying_outcomes, ["fresh_vision_path_outcome"]);
+});
diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json
index bbe996c54b..43a84b6831 100644
--- a/tsconfig.control-plane.json
+++ b/tsconfig.control-plane.json
@@ -108,6 +108,7 @@
     "tests/control_plane_ts/vision_wait_coverage.test.ts",
     "tests/control_plane_ts/shared_goal_alignment.test.ts",
     "tests/control_plane_ts/replan_history.test.ts",
+    "tests/control_plane_ts/progress_delta.test.ts",
     "tests/control_plane_ts/goal_amendment_proposal.test.ts",
     "tests/control_plane_ts/goal_instance_identity.test.ts",
     "tests/control_plane_ts/chat_session_lifecycle.test.ts",