Skip to content

Commit 056390f

Browse files
authored
perf(history): defer artifact probes until after semantic retention (#5240)
* perf(history): observe artifacts after retaining decision evidence Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> * chore: refresh history registry I/O census locations Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --------- Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Co-authored-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
1 parent 2e243a1 commit 056390f

5 files changed

Lines changed: 258 additions & 21 deletions

File tree

‎docs/status-data-contract.md‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1833,6 +1833,23 @@ It mirrors the compact run index, but strips local artifact paths. UIs should
18331833
show artifact availability with `json_exists` and `markdown_exists` instead of
18341834
linking directly to local files.
18351835

1836+
Artifact flags are fresh observations for every returned Run, not facts trusted
1837+
from the persisted index. History collection first reads and deduplicates the
1838+
complete index and computes quota and semantic history; only then does it check
1839+
files for retained recent, latest-status, agent-lane and semantic-evidence rows.
1840+
`--limit 0` does not discard old vision, owner correction or active blocked-retry
1841+
evidence. Creating or deleting an artifact is visible on the next read even when
1842+
the index bytes are unchanged. Full `load_index`/`load_index_snapshot` readers
1843+
still observe every row by default. This is a caller filesystem optimization,
1844+
independent of the Todo authority provider, not an index cache or a limit on
1845+
decision history.
1846+
1847+
文件存在性是每次读取对返回记录的实时观测,不信任索引中保存的布尔值。先用完整索引
1848+
完成去重、额度和语义历史归约,再检查近期记录、最新状态、Agent 窗口及保留语义证据
1849+
的文件;即使 `--limit 0`,旧愿景、用户纠偏和有效阻塞重试也不会因此丢失。索引不变时,
1850+
文件创建/删除仍在下次读取生效;底层完整索引接口默认保留逐条检查行为。这不改变
1851+
File/SQLite 等 Todo provider 的权威,也不限制决策历史。
1852+
18361853
On the `status`, `quota should-run`, and `history` read paths, relative
18371854
`common_runtime_root` values, relative `--runtime-root` overrides, and relative
18381855
run-index artifact paths are resolved against the project root that owns the

‎loopx/control_plane/runtime/run_context_retention.py‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
from __future__ import annotations
22

3-
from collections.abc import Callable
3+
from collections.abc import Callable, Iterator
44
from typing import Any
55

66
from ..quota.blocked_retry import active_turn_retry_for_run
@@ -232,6 +232,23 @@ def goal_semantic_history_from_runs(
232232
return semantic_history
233233

234234

235+
def iter_goal_semantic_history_runs(value: dict[str, Any]) -> Iterator[dict[str, Any]]:
236+
"""Enumerate retained Run references without reselecting or limiting history.
237+
238+
The filesystem adapter uses this shape-owned traversal to observe artifacts
239+
after semantic reduction, including evidence outside the recent-run window.
240+
"""
241+
for context in value.get("agents", []):
242+
for field in SEMANTIC_CONTEXT_RUN_FIELDS:
243+
run = context.get(field)
244+
if isinstance(run, dict):
245+
yield run
246+
yield from value.get("active_blocked_retry_runs", [])
247+
correction = value.get("latest_owner_correction_run")
248+
if isinstance(correction, dict):
249+
yield correction
250+
251+
235252
def compact_goal_semantic_history(
236253
value: Any,
237254
*,

‎loopx/history.py‎

Lines changed: 40 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,10 @@
33
import hashlib
44
import json
55
from contextlib import nullcontext
6-
from collections.abc import Callable
6+
from collections.abc import Callable, Iterable
77
from dataclasses import dataclass
88
from heapq import merge
9-
from itertools import islice
9+
from itertools import chain, islice
1010
from pathlib import Path
1111
from typing import Any
1212

@@ -35,6 +35,7 @@
3535
)
3636
from .control_plane.runtime.run_context_retention import (
3737
goal_semantic_history_from_runs,
38+
iter_goal_semantic_history_runs,
3839
latest_runs_with_agent_context,
3940
)
4041
from .control_plane.runtime.run_index_duplicates import (
@@ -229,32 +230,40 @@ def _indexed_artifact_exists(value: Any, *, artifact_root: Path | None) -> bool:
229230
return path.exists()
230231

231232

233+
def _observe_run_artifacts(
234+
records: Iterable[dict[str, Any]], *, artifact_root: Path | None,
235+
) -> None:
236+
"""Fresh request-local observations, shared by full and selected reads."""
237+
observed: dict[str, bool] = {}
238+
for record in records:
239+
for source, target in (("json_path", "json_exists"), ("markdown_path", "markdown_exists")):
240+
path = str(record.get(source) or "").strip()
241+
if path not in observed:
242+
observed[path] = _indexed_artifact_exists(path, artifact_root=artifact_root)
243+
record[target] = observed[path]
244+
245+
232246
def load_index_snapshot(
233247
path: Path,
234248
*,
235249
artifact_root: Path | None = None,
250+
include_artifact_status: bool = True,
236251
) -> RunIndexSnapshot:
252+
"""Decode the complete index; derived artifact status is optional internally.
253+
254+
Full readers keep fresh status by default. History collection defers only
255+
filesystem observation until its complete semantic selection is known.
256+
"""
237257
try:
238258
stream = path.open("rb")
239259
except FileNotFoundError:
240260
return RunIndexSnapshot(records=[], raw_count=0, digest=None)
241261

242262
records: list[dict[str, Any]] = []
243263
positions: dict[tuple[str, str, str], int] = {}
244-
artifact_exists: dict[tuple[str, str], bool] = {}
245264
raw_count = 0
246265
digest = hashlib.sha256()
247266

248-
def artifact_is_present(value: Any) -> bool:
249-
text = str(value or "").strip()
250-
cache_key = (text, str(artifact_root or ""))
251-
if cache_key not in artifact_exists:
252-
artifact_exists[cache_key] = _indexed_artifact_exists(
253-
value,
254-
artifact_root=artifact_root,
255-
)
256-
return artifact_exists[cache_key]
257-
258267
with stream:
259268
for encoded_line in stream:
260269
digest.update(encoded_line)
@@ -275,13 +284,16 @@ def artifact_is_present(value: Any) -> bool:
275284
str(item.get("markdown_path") or ""),
276285
)
277286
item = dict(item)
278-
item["json_exists"] = artifact_is_present(item.get("json_path"))
279-
item["markdown_exists"] = artifact_is_present(item.get("markdown_path"))
287+
# These are observations, never persisted index authority.
288+
item.pop("json_exists", None)
289+
item.pop("markdown_exists", None)
280290
if key in positions:
281291
records[positions[key]].update(item)
282292
else:
283293
positions[key] = len(records)
284294
records.append(item)
295+
if include_artifact_status:
296+
_observe_run_artifacts(records, artifact_root=artifact_root)
285297
return RunIndexSnapshot(
286298
records=records,
287299
raw_count=raw_count,
@@ -365,7 +377,7 @@ def collect_history(
365377
index_path = runtime_root / "goals" / current_goal_id / "runs" / "index.jsonl"
366378
index_snapshot = load_index_snapshot(
367379
index_path,
368-
artifact_root=registry_project_root(registry_path),
380+
include_artifact_status=False,
369381
)
370382
runs = index_snapshot.records
371383
raw_count = index_snapshot.raw_count
@@ -433,6 +445,18 @@ def collect_history(
433445
),
434446
"semantic_history": goal_semantic_history_from_runs(runs),
435447
}
448+
# Reduce the full index first. Old semantic evidence and lane windows
449+
# may survive outside --limit; every returned Run still needs fresh
450+
# flags. These references also cover the global recent_runs window.
451+
status_run = goal_record["latest_status_run"]
452+
_observe_run_artifacts(
453+
chain(
454+
goal_record["latest_runs"],
455+
[status_run] if status_run is not None else [],
456+
iter_goal_semantic_history_runs(goal_record["semantic_history"]),
457+
),
458+
artifact_root=registry_project_root(registry_path),
459+
)
436460
if registry_member:
437461
for field in REGISTRY_STATUS_FIELDS:
438462
if meta.get(field):

‎loopx/semantics/project_registry_io_manifest_v1.json‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1759,31 +1759,31 @@
17591759
},
17601760
{
17611761
"site": "loopx/history.py::<module>.collect_history::codec_read:load_registry#1",
1762-
"line": 329,
1762+
"line": 341,
17631763
"column": 20,
17641764
"kind": "codec_read",
17651765
"api": "load_registry",
17661766
"classification": "codec_api"
17671767
},
17681768
{
17691769
"site": "loopx/history.py::<module>.inspect_index_duplicates::codec_read:load_registry#1",
1770-
"line": 573,
1770+
"line": 597,
17711771
"column": 16,
17721772
"kind": "codec_read",
17731773
"api": "load_registry",
17741774
"classification": "codec_api"
17751775
},
17761776
{
17771777
"site": "loopx/history.py::<module>.rebuild_index_artifact_collisions::codec_read:load_registry#1",
1778-
"line": 787,
1778+
"line": 811,
17791779
"column": 16,
17801780
"kind": "codec_read",
17811781
"api": "load_registry",
17821782
"classification": "codec_api"
17831783
},
17841784
{
17851785
"site": "loopx/history.py::<module>.repair_index_duplicates::codec_read:load_registry#1",
1786-
"line": 677,
1786+
"line": 701,
17871787
"column": 16,
17881788
"kind": "codec_read",
17891789
"api": "load_registry",
Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
"""History retains control evidence without probing every discarded artifact."""
2+
from __future__ import annotations
3+
4+
import hashlib
5+
import json
6+
import subprocess
7+
import sys
8+
from datetime import datetime, timedelta, timezone
9+
10+
import pytest
11+
12+
import loopx.history as history_module
13+
from loopx.history import collect_history, load_index_snapshot
14+
15+
16+
@pytest.fixture
17+
def history_case(tmp_path, monkeypatch):
18+
registry = tmp_path / "registry.json"
19+
registry.write_text("{}\n")
20+
runtime = tmp_path / "runtime"
21+
index = runtime / "goals" / "example" / "runs" / "index.jsonl"
22+
index.parent.mkdir(parents=True)
23+
artifacts = tmp_path / "artifacts"
24+
artifacts.mkdir()
25+
start = datetime(2026, 9, 1, tzinfo=timezone.utc)
26+
now = start + timedelta(seconds=1001)
27+
monkeypatch.setattr(
28+
"loopx.control_plane.runtime.run_context_retention.now_utc_iso",
29+
lambda: now.isoformat(),
30+
)
31+
rows = []
32+
for n in range(1000):
33+
rows.append({
34+
"generated_at": (start + timedelta(seconds=n)).isoformat(),
35+
"run_id": f"run-{n}", "classification": "progress",
36+
"agent_id": "worker", "json_path": f"artifacts/{n}.json",
37+
"markdown_path": f"artifacts/{n}.md",
38+
# Persisted booleans are deliberately wrong: each read observes files.
39+
"json_exists": True, "markdown_exists": True,
40+
})
41+
rows[0]["agent_vision"] = {"agent_id": "worker", "revision": "retained"}
42+
rows[1]["human_reward"] = {"decision": "revise", "reason_summary": "Keep evidence"}
43+
rows[990].update({
44+
"todo_id": "todo_blocked_work", "turn_instance_id": "turn-1",
45+
"delivery_outcome": "outcome_gap",
46+
"progress_observation": {"result_class": "blocked"},
47+
"blocked_retry": {
48+
"schema_version": "quota_blocked_retry_v0", "source": "turn_settlement",
49+
"todo_id": "todo_blocked_work", "observed_at": rows[990]["generated_at"],
50+
"due_at": (start + timedelta(seconds=1290)).isoformat(),
51+
"resume_when": f"resume_at:{(start + timedelta(seconds=1290)).isoformat()}",
52+
},
53+
})
54+
rows[995]["agent_id"] = "quiet-worker"
55+
# Recent neutral records must not hide the latest status-relevant run.
56+
rows[999]["classification"] = "quota_slot_spent"
57+
rows[998]["classification"] = "quota_slot_spent"
58+
# Last duplicate replaces only supplied fields, without another unique Run.
59+
duplicate = {**rows[0], "annotation": "merged duplicate"}
60+
index.write_text("\n".join(json.dumps(row) for row in [*rows, duplicate]) + "\nnot-json\n[]\n\n")
61+
for n in (0, 1, 990, 995, 997, 999):
62+
(artifacts / f"{n}.json").write_text("{}\n")
63+
return registry, runtime, index, artifacts
64+
65+
66+
def _collect(case, *, limit=1, agent_lane_id=None):
67+
registry, runtime, _, _ = case
68+
return collect_history(registry_path=registry, runtime_root=runtime,
69+
goal_id="example", limit=limit, agent_lane_id=agent_lane_id)
70+
71+
72+
@pytest.mark.parametrize("limit", [0, 1, 3])
73+
def test_history_observes_retained_evidence_after_full_reduction(history_case, monkeypatch, limit):
74+
_, _, index, _ = history_case
75+
calls = []
76+
original = history_module._indexed_artifact_exists
77+
78+
def observe(value, *, artifact_root):
79+
calls.append(value)
80+
return original(value, artifact_root=artifact_root)
81+
82+
monkeypatch.setattr(history_module, "_indexed_artifact_exists", observe)
83+
result = _collect(history_case, limit=limit)
84+
goal = result["goals"][0]
85+
assert result["run_count"] == goal["unique_runs"] == 1000
86+
assert goal["raw_index_records"] == 1003
87+
assert goal["index_digest"] == f"sha256:{hashlib.sha256(index.read_bytes()).hexdigest()}"
88+
assert len(result["runs"]) == len(goal["latest_runs"]) == limit
89+
semantic = goal["semantic_history"]
90+
vision = semantic["agents"][0]["latest_agent_vision_run"]
91+
correction = semantic["latest_owner_correction_run"]
92+
retry = semantic["active_blocked_retry_runs"][0]
93+
assert vision["run_id"] == "run-0"
94+
assert vision["annotation"] == "merged duplicate"
95+
assert correction["run_id"] == "run-1"
96+
assert retry["run_id"] == "run-990"
97+
assert goal["latest_status_run"]["run_id"] == "run-997"
98+
for row in [vision, correction, retry, goal["latest_status_run"]]:
99+
assert row["json_exists"] is True
100+
assert row["markdown_exists"] is False
101+
# Counts depend on retained output, never the 1,000-row historical population.
102+
expected = {0, 1, 990, 997, *range(1000 - limit, 1000)}
103+
assert set(calls) == {f"artifacts/{n}.{ext}" for n in expected for ext in ("json", "md")}
104+
assert len(calls) == len(set(calls))
105+
106+
107+
def test_lane_window_artifacts_remain_observed(history_case):
108+
result = _collect(history_case, agent_lane_id="quiet-worker")
109+
runs = result["goals"][0]["latest_runs"]
110+
assert [run["run_id"] for run in runs] == ["run-999", "run-995"]
111+
assert all(run["json_exists"] and not run["markdown_exists"] for run in runs)
112+
113+
114+
def test_unchanged_index_does_not_cache_artifact_existence(history_case):
115+
_, _, index, artifacts = history_case
116+
before = index.read_bytes()
117+
initial = _collect(history_case)
118+
assert initial["runs"][0]["json_exists"] is True
119+
assert initial["runs"][0]["markdown_exists"] is False
120+
(artifacts / "999.json").unlink()
121+
(artifacts / "999.md").write_text("# Delivered\n")
122+
(artifacts / "0.json").unlink()
123+
reread = _collect(history_case)
124+
assert reread["runs"][0]["json_exists"] is False
125+
assert reread["runs"][0]["markdown_exists"] is True
126+
assert reread["goals"][0]["semantic_history"]["agents"][0]["latest_agent_vision_run"]["json_exists"] is False
127+
assert index.read_bytes() == before
128+
assert reread["goals"][0]["index_digest"] == initial["goals"][0]["index_digest"]
129+
130+
131+
def test_full_index_reader_still_observes_every_row(history_case):
132+
registry, _, index, artifacts = history_case
133+
(artifacts / "500.md").write_text("# Older artifact\n")
134+
(artifacts / "501.json").symlink_to(artifacts / "absent.json")
135+
snapshot = load_index_snapshot(index, artifact_root=registry.parent)
136+
assert len(snapshot.records) == 1000
137+
assert snapshot.records[500]["markdown_exists"] is True
138+
assert snapshot.records[501]["json_exists"] is False
139+
assert snapshot.records[502]["json_exists"] is False
140+
141+
142+
@pytest.mark.parametrize("provider", ["file", "sqlite"])
143+
def test_public_history_and_status_keep_artifacts_with_canonical_todos(history_case, monkeypatch, tmp_path, provider):
144+
from tests.control_plane.canonical_authority_fixture import (
145+
initialize_canonical_authority, isolate_sqlite_runtime,
146+
)
147+
from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
148+
from loopx.control_plane.effect_runtime import restart_effect_runtime
149+
from loopx.control_plane.testing.canary_harness import write_fixture_registry
150+
151+
isolate_sqlite_runtime(tmp_path, monkeypatch)
152+
registry, runtime, _, _ = history_case
153+
state = tmp_path / "state.md"
154+
state.write_text("---\nstatus: active\n---\n# Example\n\n## Agent Todo\n")
155+
write_fixture_registry(project=tmp_path, registry_path=registry, runtime_root=runtime,
156+
goal_id="example", domain="engineering", state_file=str(state),
157+
adapter_kind="generic_project_goal_v0")
158+
initialize_canonical_authority(
159+
runtime, "example", build_todo_runtime_shadow_projection(goal_id="example", todos=[]),
160+
state_path=state, provider=provider,
161+
)
162+
try:
163+
for command in ("history", "status"):
164+
result = subprocess.run(
165+
[sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry),
166+
"--runtime-root", str(runtime), "--format", "json", command,
167+
"--goal-id", "example", "--limit", "1"],
168+
capture_output=True, text=True, timeout=60,
169+
)
170+
assert result.returncode == 0, result.stdout + result.stderr
171+
payload = json.loads(result.stdout)
172+
history = payload if command == "history" else payload["run_history"]
173+
goal = history["goals"][0]
174+
assert goal["unique_runs"] == 1000
175+
assert goal["latest_status_run"]["run_id"] == "run-997"
176+
assert goal["latest_status_run"]["json_exists"] is True
177+
assert goal["latest_status_run"]["markdown_exists"] is False
178+
finally:
179+
restart_effect_runtime()

0 commit comments

Comments
 (0)