fix(codex): recover current activity beyond cached history

apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx

@@ -1472,7 +1472,7 @@ const en = {
   "activity.hostClaimed": "Claimed by host",
   "activity.hostIdle": "{host} thread idle",
   "activity.hostWindowStale": "Activity outside its expected window ({hours} h)",
-  "activity.observationUnavailable": "state unavailable",
+  "activity.observationUnavailable": "execution state unknown",
   "activity.observationGaps": "gaps · {count}",
   "activity.observationIncomplete": "partial coverage",
   "activity.observationDetails": "Observed activity · {goal}",
@@ -3115,7 +3115,7 @@ const zhCN: Record<WorkspaceMessageKey, string> = {
   "activity.hostClaimed": "宿主已领取",
   "activity.hostIdle": "{host} 线程空闲",
   "activity.hostWindowStale": "活动超出预期时间窗（{hours} 小时）",
-  "activity.observationUnavailable": "状态不可读",
+  "activity.observationUnavailable": "执行状态未知",
   "activity.observationGaps": "缺口 {count}",
   "activity.observationIncomplete": "覆盖不完整",
   "activity.observationDetails": "活动观测 · {goal}",

examples/personal-workspace-browser/goal-activity.mjs

@@ -250,7 +250,7 @@ export const goalActivityScenario = {
       assert.equal(new URL(page.url()).searchParams.get("goalId"), selectedBeforeHelp, "Help never navigates or starts work");
       const unavailable = await goalRow(page, "Research Monitor");
       const unknownState = unavailable.locator("small");
-      assert.equal(await unknownState.innerText(), "状态不可读");
+      assert.equal(await unknownState.innerText(), "执行状态未知");
       assert.equal(await unavailable.locator(".personal-goal-window-badge").count(), 0, "An all-unknown status is stated once");
       assert.equal(await unknownState.evaluate(element => {
         const probe = document.createElement("span"); probe.style.color = "var(--pw-muted)"; element.append(probe);
@@ -265,7 +265,7 @@ export const goalActivityScenario = {
       assert.match(await unknownHelp.innerText(), /无法读取宿主的本机记录/);
       assert.match(await unknownHelp.innerText(), /无法识别宿主记录格式/);
       await page.keyboard.press("Escape");
-      assert.doesNotMatch(await (await goalRow(page, "Multi Agent Projection")).locator("small").innerText(), /执行中|观测缺口|状态不可读/,
+      assert.doesNotMatch(await (await goalRow(page, "Multi Agent Projection")).locator("small").innerText(), /执行中|观测缺口|执行状态未知/,
         "Another Goal's completed activity stays isolated");
       await page.screenshot({path: resolve(outputDir, "goal-activity-observation-desktop.png"), animations: "disabled"});
       await page.setViewportSize({width: 390, height: 844});
@@ -289,7 +289,7 @@ export const goalActivityScenario = {
       await page.reload({waitUntil: "networkidle"});
       const recovered = await goalRow(page, "Product Release");
       await recovered.locator(".personal-goal-window-badge").waitFor({state: "hidden"});
-      assert.doesNotMatch(await recovered.locator("small").innerText(), /Running|state unavailable|gaps/);
+      assert.doesNotMatch(await recovered.locator("small").innerText(), /Running|execution state unknown|gaps/);
       assert.equal(await recovered.locator(".personal-goal-window-badge, .personal-goal-mark.is-live").count(), 0,
         "A refreshed complete ended-turn observation clears the old gap without inventing execution");
     } finally {
@@ -310,7 +310,7 @@ export const goalActivityScenario = {
       try {
         const { api, page } = partial;
         const queued = await goalRow(page, "Multi Agent Projection");
-        await queued.locator("small", { hasText: "状态不可读" }).waitFor({ timeout: 15_000 });
+        await queued.locator("small", { hasText: "执行状态未知" }).waitFor({ timeout: 15_000 });
         assert.doesNotMatch(await queued.locator("small").innerText(), /线程空闲|执行中/, "A partial or legacy sample cannot prove every bound thread idle");
         assert.match(await (await goalRow(page, "Product Release")).locator("small").innerText(), /^执行中/, "A positive open-turn observation remains usable with partial coverage");
         assert.doesNotMatch(await (await goalRow(page, "Research Monitor")).locator("small").innerText(), /线程空闲|执行中/, "An unrecognized record never becomes idle or running");

loopx/hosts/codex_rollout_cursor.py

@@ -343,13 +343,27 @@ def observe(
                 return HostThreadActivity.unknown(
                     HostThreadUnknownReason.RECORD_UNRECOGNIZED
                 )
-        if not self.loaded:
+        loading = not self.loaded
+        if loading:
             self.load(cache_root, reader, stat)
             # Cache load/reset cannot certify or erase the independently read header.
             self.header_valid = self.loaded = True
         # Also verify retained marker/event bytes; append metadata is not a clock.
         if self.back is not None and self.back.done and not self._verify_facts(reader):
             self.reset(stat)
+        if (
+            self.back is not None
+            and self.back.done
+            and self.forward is not None
+            and stat.st_size - self.forward.pos > reader.remaining
+            and (loading or stat.st_size - self.size > reader.remaining)
+        ):
+            # A cached old turn must not make current liveness wait for a
+            # large intervening transcript. Search the current tail with the
+            # same bounded reader. Do this once per load/large append so an
+            # unfinished giant record can still converge on later reads.
+            self.back = Backward(stat.st_size)
+            self.forward = None
         result = self.advance(reader, stat)
         self.size, self.mtime = stat.st_size, stat.st_mtime_ns
         self.ctime = stat.st_ctime_ns
@@ -382,6 +396,7 @@ def save(self, root: Path | None) -> None:
             or not self.back.done
             or self.back.marker is None
             or self.back.latest is None
+            or (self.forward is not None and self.forward.pos < self.size)
         ):
             return
         latest = (self.forward.latest if self.forward else None) or self.back.latest

tests/test_codex_rollout_cursor.py

@@ -136,6 +136,45 @@ def test_latest_marker_interrupts_pending_historical_recovery(store, kind, expec
     assert result.last_event_at == T2
 
 
+@pytest.mark.parametrize("restart", [False, True])
+@pytest.mark.parametrize(
+    "kind, expected",
+    [
+        ("task_started", HostThreadState.TURN_OPEN),
+        ("task_complete", HostThreadState.IDLE),
+        ("turn_aborted", HostThreadState.IDLE),
+    ],
+)
+def test_a_cached_old_turn_does_not_delay_a_new_tail_marker(store, restart, kind, expected):
+    path = store[1]
+    write(path, encode(event("task_started")))
+    assert observe(store, cached=True).state is HostThreadState.TURN_OPEN
+    with path.open("ab") as stream:
+        stream.write(encode(item("intervening history" * 40)) * 120)
+        stream.write(encode(event(kind, T2)))
+    if restart:
+        observer._rollout_cache.clear()
+
+    result = observe(store, cached=True)
+    assert result.state is expected
+    assert result.last_event_at == T2
+    observer._rollout_cache.clear()
+    assert observe(store, cached=True) == result
+
+
+def test_incomplete_catchup_does_not_replace_a_verified_cache_checkpoint(store):
+    path = store[1]
+    write(path, encode(event("task_started")))
+    assert observe(store, cached=True).state is HostThreadState.TURN_OPEN
+    checkpoint = {p: p.read_bytes() for p in store[2].glob("*.json")}
+    with path.open("ab") as stream:
+        stream.write(encode(item("unread history" * 40)) * 120)
+
+    assert observe(store, cached=True).state is HostThreadState.UNKNOWN
+    assert {p: p.read_bytes() for p in checkpoint} == checkpoint
+    assert converge(store, cached=True).state is HostThreadState.TURN_OPEN
+
+
 def test_busy_cursor_does_not_delay_or_hide_another_thread(store):
     root, path, _cache = store
     write(path, encode(event("task_started")))