fix(context): reuse immutable reader references across parent aliases

docs/reference/required-work-context.md

@@ -1168,6 +1168,13 @@ Oversized or missing tool output is recovered from the affected range of this
 same capture, never by another admission. Off responses retain their original
 contract. Reader/output sizing alone proves neither model retention nor savings.
 
+Republishing a reader reference through a normal parent-path alias reuses the
+existing immutable descriptor when the resolved parent, exact leaf name, full
+digest and all other metadata agree. The stored descriptor is never rewritten.
+Different leaf names, including hardlinks, remain distinct; capture and descriptor
+leaf symlinks are still rejected by the reader. This compatibility does not
+establish retention or authorize another admission.
+
 `quota context-project` 开启上下文投递后，不再要求另外手工准备缓存目录。
 省略 `--decision-output-root` 时复用原生会话的私有上下文目录；显式指定的新目录
 在已存在的父目录下以 `0700` 创建。已有目录仍须属于当前用户、权限私有且不是
@@ -1180,3 +1187,8 @@ contract. Reader/output sizing alone proves neither model retention nor savings.
 上限 64000 字节；Markdown 保留 4096 字节正文默认值。显式上限和 UTF-8 字节 cursor
 语义不变。工具输出过大或缺失时，只恢复同一捕获的受影响范围，不重新准入。
 关闭态保持原有契约。页大小与输出预算本身不证明模型保留或成本收益。
+
+通过正常父路径别名再次发布读取引用时，只有解析后的父目录、精确末级文件名、完整
+摘要及其余元数据均一致，才复用已有不可变描述文件；不重写其内容。不同末级文件名
+仍是不同引用，即使它们是硬链接；读取器仍拒绝 capture 或描述文件的末级符号链接。
+此兼容不证明模型保留，也不授权再次准入。

loopx/control_plane/turn_driver/decision_capture.py

@@ -264,6 +264,14 @@ def write_context_read_reference(path: Path, digest: str, *, selection: dict[str
         write_decision_capture(reference, payload)
     except FileExistsError:
         existing = read_context_read_reference(str(reference))[1]
+        if existing["path"] != payload["path"]:
+            # Readers resolve parent aliases, while retaining the exact leaf.
+            # Compare that same spelling without rewriting immutable metadata
+            # or treating a different leaf (including a hardlink) as identical.
+            original = Path(existing["path"]).expanduser().absolute()
+            current = path.expanduser().absolute()
+            if original.parent.resolve(strict=True) / original.name == current.parent.resolve(strict=True) / current.name:
+                existing["path"] = payload["path"]
         if json.dumps(existing, sort_keys=True) != json.dumps(payload, sort_keys=True):
             raise ValueError("original reader reference changed; restore it without rerunning admission") from None
     return reference

skills/loopx-self-repair/references/repair-patterns.md

@@ -5,6 +5,7 @@ teaches a reusable control-plane lesson.
 
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
+| `immutable_reader_parent_alias` | A warm native view refuses an unchanged reader reference after a verified capture read normalizes its parent path. | Cold descriptor bytes, warm baseline path and full digest, parent aliases versus distinct leaf names, and actual CLI recovery after compaction. | The reader accepts normal parent aliases, but immutable publication compares their literal path spellings. | Compare only equivalent parent spellings at the existing IO owner, retaining the exact leaf and every other metadata value; never rewrite the descriptor. Preserve full-digest, selection, permission and final-symlink checks. Cover both publication orders, distinct files and hardlinks, cold/warm native publication and registry-free current recovery; keep transport repair separate from retention and net-benefit claims. |
 | `todo_read_source_authority_label` | A canonical Todo read reports one provider in `authority_read` but labels the public result or projection source as File. | The same read's canonical provider revision, public source, envelope sources and Markdown across full, filtered, exact and compact views. | The shared formatter retained a provider-specific constant after provider routing moved to the canonical reader. | Reuse the existing canonical source authority for presentation and provenance. Exercise real CLI File/SQLite reads, missing results and source failure/recovery; preserve filtering, read-only behavior and fail-closed routing. A display correction does not select, migrate or qualify a provider. |
 | `vision_history_null_stride_decision` | Reusing the current Agent's complete Vision from `history` in `refresh-state --agent-vision-json` fails with `stride_decision must be an object`. | Compare the authored packet, compact history and status readback; retain the original operation identity and verify that rejection appended nothing. | The compact index represents an absent optional decision as JSON null while the write codec accepts only an omitted field. | Normalize null and omission as absence in the existing typed stride owner, shared by read and write codecs. Keep malformed proposals rejected and enabled stride reviews dependent on a current valid decision. Prove real File/SQLite history-to-writeback recovery, retained cumulative intent, unchanged-state refusals and off/on policy boundaries; do not rewrite historical receipts or count absence as a decision. |
 | `async_command_completion_metadata_lost` | A code cell appears complete with empty stdout, then the agent searches saved files or reissues a stateful command without checking whether its child is still running. | The original tool result's running handle, terminal exit code and stderr; caller projection code; the same command's completion; original operation/capture identity. Keep empty success, empty failure and a pending command distinct. | The caller forwarded only stdout and dropped the command tool's completion metadata; a completed wrapper was mistaken for a completed child. Empty output alone cannot establish this cause in a historical trace whose inner result is unavailable. | Preserve the tool result envelope, retain a running handle and collect the same execution's terminal result. If the handle was lost, follow original-receipt or digest-bound capture recovery before any retry. Keep admission/lease/settlement fences and original identities; qualify with pending, silent-success and silent-failure controls. See targeted diagnostics; no timeout, quota, packet or authority change is implied. |

tests/control_plane/test_context_read_reference.py

@@ -196,3 +196,128 @@ def test_reference_publication_never_overwrites_an_existing_document(tmp_path):
     assert write_context_read_reference(path, "a" * 64) == original
     with pytest.raises(ValueError, match="reference changed"):
         write_context_read_reference(path, "b" * 64)
+
+
+@pytest.mark.parametrize("alias_first", [False, True])
+def test_reference_reuses_a_parent_alias_without_rewriting(tmp_path, alias_first):
+    physical = tmp_path / "physical"
+    physical.mkdir()
+    alias = tmp_path / "alias"
+    alias.symlink_to(physical, target_is_directory=True)
+    path = physical / "capture.json"
+    raw = b'{"ok":true,"last_stop":"Keep the original budget."}\n'
+    path.write_bytes(raw)
+    digest = hashlib.sha256(raw).hexdigest()
+    spellings = [path, alias / path.name]
+    if alias_first:
+        spellings.reverse()
+    first = write_context_read_reference(spellings[0], digest)
+    before = first.read_bytes()
+    second = write_context_read_reference(spellings[1], digest)
+    assert second.resolve() == first.resolve()
+    assert first.read_bytes() == before
+    from loopx.control_plane.turn_driver.decision_capture import context_capture_read_command
+
+    page = read(context_capture_read_command(path, digest, reference_file=second))
+    assert page["ok"] and page["eof"] and json.loads(page["content"]) == json.loads(raw)
+    assert first.read_bytes() == before
+
+
+@pytest.mark.parametrize("damage", ["digest", "other_file", "leaf_symlink", "hardlink", "selection", "navigation", "extra"])
+def test_parent_alias_reuse_preserves_other_reference_checks(tmp_path, damage):
+    physical = tmp_path / "physical"
+    physical.mkdir()
+    alias = tmp_path / "alias"
+    alias.symlink_to(physical, target_is_directory=True)
+    path = physical / "capture.json"
+    path.write_text('{}\n')
+    digest = hashlib.sha256(path.read_bytes()).hexdigest()
+    reference = write_context_read_reference(path, digest)
+    payload = json.loads(reference.read_bytes())
+    if damage == "digest":
+        digest = "0" * 64
+    elif damage in {"other_file", "leaf_symlink", "hardlink"}:
+        other = physical / "different.json"
+        if damage == "leaf_symlink":
+            other.symlink_to(path)
+        elif damage == "hardlink":
+            other.hardlink_to(path)
+        else:
+            other.write_bytes(path.read_bytes())
+        payload["path"] = str(other)
+    else:
+        payload[damage] = {"changed": True}
+    reference.write_text(json.dumps(payload))
+    before = reference.read_bytes()
+    with pytest.raises(ValueError, match="reference changed"):
+        write_context_read_reference(alias / path.name, digest)
+    assert reference.read_bytes() == before
+
+
+@pytest.mark.parametrize("parent_alias", [False, True], ids=["physical", "parent-alias"])
+def test_native_warm_opaque_reference_and_current_recovery(tmp_path, monkeypatch, parent_alias):
+    from loopx.control_plane.turn_driver import codex_context as context
+    from test_codex_context_lifecycle import setup, project
+    from test_progressive_native_context import payload as native_payload, outcome
+    from test_quota_settlement_cli import _run_cli
+
+    parent = tmp_path / "physical"
+    parent.mkdir(mode=0o700)
+    physical = parent / "session"
+    physical.mkdir(mode=0o700)
+    root = physical
+    if parent_alias:
+        root = tmp_path / "alias"
+        root.symlink_to(parent, target_is_directory=True)
+        root = root / "session"
+    event = setup(root, monkeypatch)
+    payload = native_payload()
+    payload["interaction_contract"] = {"schema_version": "loopx_interaction_contract_v0",
+        "agent_channel": {"work_context": {"sources": []}}}
+    outcome(payload)["requirements"][0]["description"] = "Keep every original requirement. " * 600
+    payload["last"] = {"value": 9007199254740991, "stop": "Do not extend the budget. 不发布。"}
+    _, cold = project(root, "cold", payload=payload)
+    cold_view = native_context_capture_view(cold, root / "cold", context_directory=str(root),
+        max_page_output_bytes=12000)
+    cold_model = model_context_reference(cold_view)
+    reference = Path(cold_model["recovery"]["reference_file"])
+    descriptor_before = reference.read_bytes()
+    _, warm = project(root, "warm", payload=payload)
+    assert warm["mode"] == "delta" and warm["continuity_basis"] == "captured_file", warm.get("reason")
+    warm_view = native_context_capture_view(warm, root / "warm", context_directory=str(root),
+        max_page_output_bytes=12000)
+    warm_model = model_context_reference(warm_view)
+    assert "reference_file" in warm_model and warm_model["mode"] == "delta"
+    assert reference.read_bytes() == descriptor_before
+    assert Path(warm_model["baseline_reference"]["reference_file"]).resolve() == reference.resolve()
+    for file in (root / "cold").iterdir():
+        if not file.name.endswith(".reader.json"):
+            file.unlink()
+
+    def read_cli(command):
+        parts, offset = [], 0
+        while True:
+            words = shlex.split(command)
+            code, page = _run_cli(root / "no-registry.json", root / "no-runtime",
+                *words[words.index("quota"):], "--context-delivery", "off", cwd=physical)
+            assert code == 0 and page["ok"] and page["offset"] == offset, page
+            parts.append(page["content"])
+            if page["eof"]:
+                return json.loads("".join(parts))
+            offset = page["next_offset"]
+            command = page["next_command"]
+
+    expected = json.loads(Path(warm["recovery"]["path"]).read_bytes())
+    assert read_cli(warm_model["recovery"]["command"]) == expected
+    context.codex_context_hook(str(root), {**event, "source": "compact"})
+    assert "candidate" not in context._state(root)
+    assert read_cli(warm_model["recovery"]["command"]) == expected
+    _, after_compact = project(root, "after", payload=payload)
+    assert after_compact["mode"] == "full"
+    fresh = model_context_reference(native_context_capture_view(after_compact, root / "after",
+        context_directory=str(root), max_page_output_bytes=12000))
+    current = read_cli(fresh["recovery"]["command"])
+    assert current["last"] == payload["last"]
+    assert outcome(current)["requirements"][0]["description"] == outcome(payload)["requirements"][0]["description"]
+    assert reference.read_bytes() == descriptor_before
+    assert not (root / "no-registry.json").exists() and not (root / "no-runtime").exists()