Preserve Explore attachment commit time precision

docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md

@@ -204,10 +204,13 @@ there is no extra journal, metadata note, planner or mandatory writeback.
 A changed decision must cite the current same-Todo finding; validation and append
 CAS use the same graph snapshot. An exact durable replay with an original commit
 strictly older than the current choice may restore its missing historical
-capture. This does not relax predecessor checks for equal-time or newer work,
+capture. The primary writeback preserves microsecond precision in the
+attachment timestamp so distinct commits in the same second remain ordered.
+This does not relax predecessor checks for equal-time or newer work,
 and attachment callers cannot supply the replay timestamp. Current context
-selects by original committed time, so late ingestion cannot replace a newer
-choice. Ordinary work preserves its original settlement type and never creates
+uses the original commit time across independent histories and follows validated
+successors within a history, including when the clock moves backwards.
+Ordinary work preserves its original settlement type and never creates
 an obligation to satisfy Stride v2. Replans retain their existing v2 contract.
 An absent attachment preserves the current decision; an explicit `proxy: null`
 chooses direct evaluation while historical profiles remain readable.

docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md

@@ -163,9 +163,10 @@ M1 在既有 `explore_result_attachment_v0` 中添加可选 `work_yardstick_deci
 原 settlement 提供 Goal/Agent/Todo/Turn 身份。既有 run journal、Explore event log
 和 Todo 追加关联管理持久化与重试；不新建 journal、metadata note、planner 或必填写回。
 决策变化必须引用同 Todo 当前 finding，校验与 append CAS 使用同一 graph 快照。
-原提交时间严格早于当前选择的 durable 精确重放可以补回缺失的历史 capture，
-但不放宽同刻或较新工作的前驱检查，附件调用方也不能提供重放时间戳。
-按原提交时间选择当前决策，迟到接收不能覆盖更新的选择。
+原提交时间严格早于当前选择的 durable 精确重放可以补回缺失的历史 capture。
+主写回保留附件时间戳的微秒精度，以区分同一秒内的不同提交。
+同刻或较新工作的前驱检查保持严格，附件调用方也不能提供重放时间戳。
+独立历史之间按原提交时间选择，历史内沿已校验的前驱关系选到后继，即使时钟回拨也一样。
 普通工作保留原 settlement 类型，不为 Stride v2 伪造 obligation；
 replan 沿用既有 v2 契约。省略附件保留当前决策；显式 `proxy: null` 选择直接评测，
 历史 profile 仍可读取。

loopx/control_plane/runtime/time.py

@@ -107,8 +107,8 @@ def now_utc_iso() -> str:
     return utc_isoformat(now_utc())
 
 
-def now_local_iso() -> str:
-    return datetime.now(timezone.utc).astimezone().replace(microsecond=0).isoformat()
+def now_local_iso(*, timespec: str = "seconds") -> str:
+    return datetime.now(timezone.utc).astimezone().isoformat(timespec=timespec)
 
 
 def utc_isoformat(value: datetime) -> str:

loopx/control_plane/work_items/yardstick_decision.ts

@@ -119,13 +119,27 @@ function belongsToWork(row: JsonObject, binding: JsonObject): boolean {
 }
 
 export function currentWorkYardstickFinding(findings: JsonObject[], binding: JsonObject): JsonObject | undefined {
-  return findings.filter(row => row.work_yardstick !== undefined && belongsToWork(row, binding)).sort((a, b) => {
-    // Late delivery reuses the original committed time, not append order.
+  const ordered = findings.filter(row => row.work_yardstick !== undefined && belongsToWork(row, binding)).sort((a, b) => {
+    // Across independent histories, late delivery keeps its original commit time.
     const left = parseTodoTimestampMicros(String(a.last_updated_at ?? ""));
     const right = parseTodoTimestampMicros(String(b.last_updated_at ?? ""));
     if (left !== null && right !== null && left !== right) return left > right ? -1 : 1;
     return Number(b.recorded_order ?? 0) - Number(a.recorded_order ?? 0);
-  })[0];
+  });
+  const successors = new Map<unknown, JsonObject>();
+  for (const row of ordered) {
+    const prior = (row.work_yardstick as WorkYardstickCapture).decision?.prior_finding_id;
+    if (typeof prior === "string" && !successors.has(prior)) successors.set(prior, row);
+  }
+  // The recorded predecessor establishes the order even across clock rollback.
+  const visited = new Set<JsonObject>();
+  let current: JsonObject | undefined = ordered[0];
+  while (current && successors.has(current.finding_id)) {
+    if (visited.has(current)) throw new EffectRuntimeRequestError("cyclic work yardstick predecessors");
+    visited.add(current);
+    current = successors.get(current.finding_id);
+  }
+  return current;
 }
 
 function measuredPair(row: ExploreMeasurementFeedback | undefined) {

loopx/state_refresh.py

@@ -1337,6 +1337,7 @@ def read_explore_context(read_goal_id: str, read_agent_id: str | None):
                     "For a selected repository delivery, use the actual producing worktree; "
                     "the registered Goal workspace does not replace that requirement."
                 )
+        explore_result_recorded_at = None
         if checkpoint_supplement:
             # The supplemental row must not reattribute the original delivery to
             # the recovery caller's current directory or change its accounting.
@@ -1350,7 +1351,13 @@ def read_explore_context(read_goal_id: str, read_agent_id: str | None):
                 "progress_observation"
             )
             normalized_explore_result = prior_writeback_run.get("explore_result")
+            explore_result_recorded_at = prior_writeback_run.get(
+                "explore_result_recorded_at", prior_writeback_run["generated_at"]
+            )
             classification = prior_writeback_run["classification"]
+        elif normalized_explore_result is not None:
+            # Attachment recovery orders committed decisions within one wall second.
+            explore_result_recorded_at = now_local_iso(timespec="microseconds")
         record = build_state_refresh_record(
             goal_id=safe_goal_id,
             state_file=resolved_state_file,
@@ -1373,10 +1380,7 @@ def read_explore_context(read_goal_id: str, read_agent_id: str | None):
             vision_checkpoint=vision_checkpoint,
             progress_observation=normalized_progress_observation,
             explore_result=normalized_explore_result,
-            explore_result_recorded_at=(
-                prior_writeback_run.get("explore_result_recorded_at", prior_writeback_run["generated_at"])
-                if checkpoint_supplement and prior_writeback_run else None
-            ),
+            explore_result_recorded_at=explore_result_recorded_at,
             explore_result_source_snapshot=(
                 prior_writeback_run.get("explore_result_source_snapshot")
                 if checkpoint_supplement and prior_writeback_run

tests/control_plane/test_work_yardstick_cli.py

@@ -4,6 +4,7 @@
 or a causal benchmark gain. No evaluator, credentials or active Goal are used.
 """
 import hashlib
+from datetime import datetime, timezone
 from itertools import groupby
 import json
 import shlex
@@ -1125,24 +1126,88 @@ def test_disabled_new_attachment_rejects_before_primary_commit(tmp_path):
     assert not explore_result_log_path(runtime, GOAL_ID).exists()
 
 
-def test_committed_pending_capture_recovers_after_disable_without_new_authority(tmp_path, monkeypatch, capsys):
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_work_yardstick_successor_remains_current_after_clock_rollback(tmp_path, provider, monkeypatch, capsys):
+    from loopx.cli import main
+
+    project, runtime, path = setup(tmp_path, provider=provider)
+    source = tmp_path / "finding.json"
+    wall_time = datetime.now(timezone.utc).replace(microsecond=100002)
+
+    class AttachmentClock(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return wall_time.astimezone(tz)
+
+    def record(turn, value):
+        source.write_text(json.dumps(value))
+        run(project, runtime, path, "quota", "should-run", "--codex-app", *binding(turn))
+        with monkeypatch.context() as patch:
+            patch.chdir(project)
+            patch.setattr("loopx.control_plane.runtime.time.datetime", AttachmentClock)
+            code = main(["--registry", str(path), "--runtime-root", str(runtime), "--format", "json",
+                "refresh-state", *binding(turn), "--classification", "validated_change",
+                "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+                "--explore-result-json", str(source), "--no-global-sync", "--suppress-external-sinks"])
+        packet = json.loads(capsys.readouterr().out)
+        assert code == 0 and packet["explore_result_delivery"]["ok"], packet
+        return packet
+
+    first = record("before-clock-change", attachment())
+    run(project, runtime, path, "quota", "spend-slot", *binding("before-clock-change"),
+        "--slots", "1", "--source", "heartbeat", "--execute")
+    changed = attachment()
+    changed["work_yardstick_decision"]["prior_finding_id"] = first["explore_result_delivery"]["finding_id"]
+    changed["work_yardstick_decision"]["target"]["summary"] = "Follow the validated successor"
+    wall_time = wall_time.replace(microsecond=100001)
+    second = record("after-clock-change", changed)
+    assert second["explore_result_recorded_at"] < first["explore_result_recorded_at"]
+    cue = context(runtime, path)["graph"]["work_yardsticks"][0]
+    assert cue["source_finding_id"] == second["explore_result_delivery"]["finding_id"]
+    assert cue["target"]["summary"] == "Follow the validated successor"
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_committed_pending_capture_recovers_after_disable_without_new_authority(
+    tmp_path, provider, monkeypatch, capsys,
+):
     from loopx.cli import main
     import loopx.capabilities.explore.result_writeback as writer
-    project, runtime, path = setup(tmp_path)
+    from loopx.control_plane.runtime.time import parse_timestamp
+
+    project, runtime, path = setup(tmp_path, provider=provider)
     source = tmp_path / "capture.json"
     source.write_text(json.dumps(attachment()))
     turn = "turn-interrupted"
     run(project, runtime, path, "quota", "should-run", "--codex-app", *binding(turn))
-    command = ["refresh-state", *binding(turn), "--classification", "validated_change",
-        "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
-        "--explore-result-json", str(source), "--no-global-sync", "--suppress-external-sinks"]
+
+    def refresh_command(turn_id, attachment_path):
+        return ["refresh-state", *binding(turn_id), "--classification", "validated_change",
+            "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+            "--explore-result-json", str(attachment_path), "--no-global-sync", "--suppress-external-sinks"]
+
+    command = refresh_command(turn, source)
+
+    wall_time = datetime.now(timezone.utc).replace(microsecond=100001)
+
+    class AttachmentClock(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return wall_time.astimezone(tz)
+
+    def run_refresh(args):
+        with monkeypatch.context() as patch:
+            patch.chdir(project)
+            patch.setattr("loopx.control_plane.runtime.time.datetime", AttachmentClock)
+            rc = main(["--registry", str(path), "--runtime-root", str(runtime), "--format", "json", *args])
+        return rc, json.loads(capsys.readouterr().out)
+
     def unavailable(*args, **kwargs):
         raise OSError("injected graph write interruption")
+
     with monkeypatch.context() as patch:
-        patch.chdir(project)
         patch.setattr(writer, "append_explore_result_events", unavailable)
-        rc = main(["--registry", str(path), "--runtime-root", str(runtime), "--format", "json", *command])
-        failed = json.loads(capsys.readouterr().out)
+        rc, failed = run_refresh(command)
     assert rc == 1 and failed["appended"] and failed["explore_result_delivery"]["primary_committed"]
     run(project, runtime, path, "quota", "spend-slot", *binding(turn), "--slots", "1", "--source", "heartbeat", "--execute")
     # Work continues while the old graph delivery is pending. Its later exact
@@ -1151,14 +1216,24 @@ def unavailable(*args, **kwargs):
     newer = attachment()
     newer["work_yardstick_decision"]["target"]["summary"] = "Use the updated current target"
     newer_source.write_text(json.dumps(newer))
-    latest = refresh(project, runtime, path, "turn-newer", newer_source)
-    assert latest["explore_result_delivery"]["ok"]
+    run(project, runtime, path, "quota", "should-run", "--codex-app", *binding("turn-newer"))
+    wall_time = wall_time.replace(microsecond=100002)
+    rc, latest = run_refresh(refresh_command("turn-newer", newer_source))
+    assert rc == 0 and latest["explore_result_delivery"]["ok"], latest
+    assert latest["generated_at"] == failed["generated_at"]
     config = json.loads(path.read_text())
     config["goals"][0]["execution_profile"]["yardstick_enabled"] = False
     path.write_text(json.dumps(config))
     recovered = run(project, runtime, path, *command)
     assert recovered["idempotent_replay"] and recovered["explore_result_delivery"]["ok"]
-    assert len(load_explore_result_events(explore_result_log_path(runtime, GOAL_ID))) == 3
+    assert recovered["explore_result_recorded_at"] == failed["explore_result_recorded_at"]
+    assert parse_timestamp(recovered["explore_result_recorded_at"]) < parse_timestamp(latest["explore_result_recorded_at"])
+    events = load_explore_result_events(explore_result_log_path(runtime, GOAL_ID))
+    assert len(events) == 3
+    recovered_finding = next(
+        row for row in events if row.get("result_id") == recovered["explore_result_delivery"]["finding_id"]
+    )
+    assert recovered_finding["recorded_at"] == failed["explore_result_recorded_at"]
     assert "work_yardsticks" not in context(runtime, path)["graph"]
     config["goals"][0]["execution_profile"]["yardstick_enabled"] = True
     path.write_text(json.dumps(config))