diff --git a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md b/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md index 09cbc5e956..bb8027908a 100644 --- a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.md +++ b/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. diff --git a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md b/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md index cf60116e78..8414645063 100644 --- a/docs/architecture/rfcs/target-yardstick-stride-decisions-v0.zh-CN.md +++ b/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 仍可读取。 diff --git a/loopx/control_plane/runtime/time.py b/loopx/control_plane/runtime/time.py index 079c24610a..3013570a2b 100644 --- a/loopx/control_plane/runtime/time.py +++ b/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: diff --git a/loopx/control_plane/work_items/yardstick_decision.ts b/loopx/control_plane/work_items/yardstick_decision.ts index 2ae386512c..2e4bf7d136 100644 --- a/loopx/control_plane/work_items/yardstick_decision.ts +++ b/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(); + 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(); + 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) { diff --git a/loopx/state_refresh.py b/loopx/state_refresh.py index fd1e02576d..99d790f275 100644 --- a/loopx/state_refresh.py +++ b/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 diff --git a/tests/control_plane/test_work_yardstick_cli.py b/tests/control_plane/test_work_yardstick_cli.py index 1bfba7d982..71121729c2 100644 --- a/tests/control_plane/test_work_yardstick_cli.py +++ b/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))