Skip to content

test(checkpoint): initialize canonical source before its first read #7513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangruiteng merged 1 commit into from
Oct 27, 2026
+5 −7
Merged

test(checkpoint): initialize canonical source before its first read #7513

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 5 additions & 7 deletions benchmark/tests/test_native_checkpoint_handoff.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -380,8 +380,7 @@ def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, exp
from pathlib import Path
from benchmark.runtime.worker import run_once
from tests.control_plane import test_quota_settlement_cli as cli
from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority
from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture

project, runtime, registry = cli._write_fixture(tmp_path / "authority")
document = json.loads(registry.read_text())
Expand All @@ -408,11 +407,10 @@ def run(*args):
assert code == 0, json.dumps(result, indent=2)
return result

listed = run("todo", "list", "--goal-id", cli.GOAL_ID)
initialize_canonical_authority(runtime, cli.GOAL_ID,
build_todo_runtime_shadow_projection(goal_id=cli.GOAL_ID, todos=listed["todos"],
handoff_mode="hard_lease", leases=[]),
state_path=state_path, provider="sqlite")
initialize_canonical_source_fixture(
runtime, cli.GOAL_ID, state_path=state_path,
provider="sqlite", handoff_mode="hard_lease",
)
run("task-lease", "acquire", "--goal-id", cli.GOAL_ID, "--todo-id", cli.TODO_ID,
"--owner", cli.AGENT_ID, "--idempotency-key", "source-work")
configured = run("configure-goal", "--goal-id", cli.GOAL_ID,
Expand Down
Loading