fix(context): preserve native continuity across command workdirs

docs/heartbeat-automation-prompt.md

@@ -518,10 +518,20 @@ Native reuse assumes a normally completed full read was read and remains
 available until a lifecycle invalidation. `PostToolUse` merges complete reader
 ranges in the existing private session journal. EOF alone, known read failures,
 malformed/truncated pages and interrupted paging cannot establish a baseline.
-Compaction, clear, fork, session/workspace change and every SessionStart, including
-same-session resume, invalidate it; ordinary tools and turn endings do not.
+Compaction, clear, fork, session/workspace change and unqualified SessionStart
+invalidate it; ordinary tools and turn endings do not. A same-session resume can
+retain its baseline only when the native replay checkpoint verifies continuity.
 Read the pinned original completely before establishing a new differential chain.
 Repeated events and parallel pages are idempotent.
+Workspace identity comes from native lifecycle events, not a shell command's
+working directory. The same native session can run projection and recovery
+commands in a subdirectory or another checkout without discarding its baseline.
+A native SessionStart reporting a changed workspace still retires the previous
+candidate and recovery selector. Session, Goal and Agent isolation remains in
+force; changing command directories grants no work or publication authority.
+工作区身份由原生生命周期事件确定。命令切换到子目录或另一 checkout 不会清除同一
+原生会话的差分基线；原生 SessionStart 报告工作区变化时仍清除旧基线和恢复选择器。
+Session、Goal 和 Agent 隔离保持不变，切换命令目录不授予工作或发布权限。
 This replaces per-request input-history proof, not the capture integrity check.
 It preserves native providers, authentication, endpoints and connection mode,
 including implicit history. No input diagnostics or forced HTTP is required.

loopx/control_plane/turn_driver/codex_context.py

@@ -193,8 +193,7 @@ def read_codex_context_reference(directory: str, *, goal_id: str | None, agent_i
     state = _state(_private_root(directory))
     session = os.environ.get("CODEX_THREAD_ID")
     if (not session or state.get("session_id") != session
-            or not isinstance(state.get("cwd"), str)
-            or Path(state["cwd"]).resolve() != Path.cwd().resolve()):
+            or not isinstance(state.get("cwd"), str) or not state["cwd"]):
         raise ValueError("original capture lookup requires the same native session and workspace")
     recovery = state.get("recovery")
     if (not goal_id or not agent_id or not isinstance(recovery, dict)
@@ -469,10 +468,10 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 if (not isinstance(session, str) or not session
                         or (native_session and session != native_session)):
                     return payload  # Do not replace a different native session's recovery.
-                if (not native_session or not isinstance(state.get("cwd"), str)
-                        or Path(state["cwd"]).resolve() != Path.cwd().resolve()):
-                    # A later tool can restore cwd/identity and make an old
-                    # selector eligible again, so invalidate before full fallback.
+                # Native lifecycle events own workspace identity and epoch.
+                # A shell tool may run in another checkout without changing
+                # the host session or losing its previously delivered context.
+                if (not native_session or not isinstance(state.get("cwd"), str) or not state["cwd"]):
                     raise ValueError("context recovery requires the same native session and workspace")
                 if not isinstance(epoch, str) or not epoch:
                     raise ValueError("context recovery requires a native context epoch")

tests/control_plane/test_codex_context_lifecycle.py

@@ -177,6 +177,74 @@ def test_complete_reader_delivery_allows_delta_without_diagnostics_or_ack(tmp_pa
     assert third["mode"] == "delta"
 
 
+@pytest.mark.parametrize("location", ["nested", "sibling"])
+def test_command_workdir_preserves_native_session_delta_and_recovery(tmp_path, monkeypatch, location):
+    event = setup(tmp_path, monkeypatch)
+    workspace = tmp_path / "checkout"
+    workspace.mkdir()
+    command_cwd = workspace / "nested" if location == "nested" else tmp_path / "other-checkout"
+    command_cwd.mkdir()
+    event["cwd"] = str(workspace)
+    context.codex_context_hook(str(tmp_path), event)
+    monkeypatch.chdir(workspace)
+    payload = {"ok": True, "plan_summary": {
+        "requirements": "Keep the original acceptance and stop conditions. " * 1000}}
+    _, first = project(tmp_path, payload=payload)
+    deliver(tmp_path, event, read_pages(tmp_path, first))
+    epoch = context._state(tmp_path)["context_epoch"]
+    repo = Path(__file__).resolve().parents[2]
+    env = {**os.environ, "PYTHONPATH": str(repo)}
+    script = """
+import json, sys
+from pathlib import Path
+from loopx.control_plane.turn_driver.codex_context import project_codex_context
+root = Path(sys.argv[1])
+capture = root / 'two'
+capture.mkdir(mode=0o700)
+print(json.dumps(project_codex_context(json.load(sys.stdin), directory=str(root),
+    capture_directory=capture, goal_id='fixture', agent_id='agent',
+    surface='quota_packet', capture_id='two')))
+"""
+    result = subprocess.run([sys.executable, "-c", script, str(tmp_path)],
+        input=json.dumps(payload), text=True, capture_output=True,
+        cwd=command_cwd, env=env, timeout=30, check=True)
+    second = json.loads(result.stdout)
+    assert second.get("mode") == "delta"
+    assert second["target"]["context_epoch"] == epoch
+    assert context._state(tmp_path)["cwd"] == str(workspace)
+    journal = (tmp_path / "session.json").read_bytes()
+    for cwd in (command_cwd, workspace):
+        read = subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--format", "json", "quota", "context-read",
+            "--codex-context-dir", str(tmp_path), "--goal-id", "fixture", "--agent-id", "agent"],
+            cwd=cwd, env=env, capture_output=True, text=True, timeout=30, check=True)
+        page = json.loads(read.stdout)
+        assert page["ok"] and page["sha256"] == second["recovery"]["sha256"]
+    assert (tmp_path / "session.json").read_bytes() == journal
+    deliver(tmp_path, event, read_pages(tmp_path, second))
+    _, third = project(tmp_path, "three", payload=payload)
+    assert third["mode"] == "delta"
+
+
+def test_native_workspace_change_retires_previous_recovery(tmp_path, monkeypatch):
+    event = setup(tmp_path, monkeypatch)
+    payload = {"requirements": "Retain authority boundaries. " * 1000}
+    _, first = project(tmp_path, payload=payload)
+    deliver(tmp_path, event, read_pages(tmp_path, first))
+    epoch = context._state(tmp_path)["context_epoch"]
+    workspace = tmp_path / "changed-workspace"
+    workspace.mkdir()
+    context.codex_context_hook(str(tmp_path), {**event, "cwd": str(workspace), "source": "resume"})
+    # A real native workspace transition resets the owning lifecycle, even if
+    # the next CLI still happens to run in the old command directory.
+    state = context._state(tmp_path)
+    assert state["context_epoch"] != epoch
+    assert "candidate" not in state and "recovery" not in state
+    with pytest.raises(ValueError):
+        context.read_codex_context_reference(str(tmp_path), goal_id="fixture", agent_id="agent")
+    _, second = project(tmp_path, "two", payload=payload)
+    assert second["mode"] == "full" and second["context"] == payload
+
+
 def test_native_page_zero_exposes_current_continuity_and_preserves_full_recovery(tmp_path, monkeypatch):
     from argparse import Namespace
     from loopx.cli_commands.quota_capture import native_context_capture_view, read_context_capture
@@ -629,19 +697,15 @@ def fail_current_write(root, state):
 
 
 @pytest.mark.parametrize("source", ["compact", "resume"])
-@pytest.mark.parametrize("caller", ["subdirectory", "missing_session", "foreign_session"])
+@pytest.mark.parametrize("caller", ["missing_session", "foreign_session"])
 def test_fallback_identity_restoration_keeps_recovery_scoped(tmp_path, monkeypatch, source, caller):
     event = setup(tmp_path, monkeypatch)
     _, first = project(tmp_path)
     original_capture = Path(first["recovery"]["path"])
     original = original_capture.read_bytes()
     context.codex_context_hook(str(tmp_path), {**event, "hook_event_name": "PreToolUse", "tool_use_id": "guard"})
     with monkeypatch.context() as native_tool:
-        if caller == "subdirectory":
-            subdirectory = tmp_path / "subdirectory"
-            subdirectory.mkdir()
-            native_tool.chdir(subdirectory)
-        elif caller == "missing_session":
+        if caller == "missing_session":
             native_tool.delenv("CODEX_THREAD_ID")
         else:
             native_tool.setenv("CODEX_THREAD_ID", "foreign-session")
@@ -682,7 +746,7 @@ def cannot_invalidate(path, *args, **kwargs):
     assert json.loads(Path(first["recovery"]["path"]).read_bytes())["requirements"] == "Preserve final stop condition."
 
 
-@pytest.mark.parametrize("damage", ["missing", "corrupt", "symlink", "public", "other_workspace",
+@pytest.mark.parametrize("damage", ["missing", "corrupt", "symlink", "public", "invalid_workspace",
     "missing_epoch", "invalid_epoch", "read_failure", "projection_failure"])
 def test_invalid_journal_falls_back_to_original_full(tmp_path, monkeypatch, damage):
     event = setup(tmp_path, monkeypatch)
@@ -699,9 +763,9 @@ def test_invalid_journal_falls_back_to_original_full(tmp_path, monkeypatch, dama
         path.symlink_to(other)
     elif damage == "public":
         path.chmod(0o644)
-    elif damage == "other_workspace":
+    elif damage == "invalid_workspace":
         state = context._state(tmp_path)
-        state["cwd"] = str(tmp_path.parent)
+        state["cwd"] = None
         context._save(tmp_path, state)
     elif damage in {"missing_epoch", "invalid_epoch"}:
         state = context._state(tmp_path)
@@ -819,15 +883,15 @@ def test_lost_hook_reference_is_scoped_read_only_and_pins_later_pages(tmp_path,
     import argparse
     from loopx.cli_commands.quota_capture import read_context_capture
 
-    setup(tmp_path, monkeypatch)
+    event = setup(tmp_path, monkeypatch)
     _, first = project(tmp_path, payload={"goal_id": "fixture", "requirements": "保留🙂" * 400})
     journal = tmp_path / "session.json"
     args = argparse.Namespace(codex_context_dir=str(tmp_path), goal_id="fixture", agent_id="agent",
                               offset=0, page_bytes=256)
     if damage == "session":
         monkeypatch.setenv("CODEX_THREAD_ID", "other")
     elif damage == "workspace":
-        monkeypatch.chdir(tmp_path.parent)
+        context.codex_context_hook(str(tmp_path), {**event, "cwd": str(tmp_path.parent), "source": "resume"})
     elif damage in {"goal", "agent"}:
         setattr(args, damage + "_id", "other")
     elif damage == "missing_scope":

tests/control_plane/test_context_delivery_cli.py

@@ -402,11 +402,11 @@ def test_native_fresh_work_turn_reads_current_full_without_old_recovery(tmp_path
                  hook_event_name="SessionStart", source="startup")
     context.codex_context_hook(str(root), event)
 
-    def guard(turn):
+    def guard(turn, cwd=project):
         rc, view = _run_cli(registry, runtime, "quota", "should-run", "--codex-app",
             "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
             "--turn-instance-id", turn, "--decision-output-root", str(captures),
-            "--context-delivery", "on", "--codex-context-dir", str(root), cwd=project)
+            "--context-delivery", "on", "--codex-context-dir", str(root), cwd=cwd)
         assert rc == 0, view
         return view
 
@@ -449,8 +449,11 @@ def consume(view):
     assert body["heartbeat_receipt"]["turn_instance_id"] == "native-work-next"
     assert context._state(root)["candidate"]["target"] == current["target"]
     assert Path(first["recovery"]["path"]).read_bytes() == original
-    warm = guard("native-work-next")
+    command_workspace = tmp_path / "independent-checkout"
+    command_workspace.mkdir()
+    warm = guard("native-work-next", cwd=command_workspace)
     assert warm["mode"] == "delta" and warm["target"] != current["target"]
+    assert context._state(root)["cwd"] == str(project)
     assert _heartbeat_receipt_count(runtime, "native-work-first") == 1
     assert _heartbeat_receipt_count(runtime, "native-work-next") == 1
     assert _spend_run_count(runtime) == 1