Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
13 changes: 10 additions & 3 deletions examples/control_plane/status-collection-readmodel-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
sys.path.insert(0, str(ROOT))

from loopx import status as status_module # noqa: E402
from loopx.control_plane.goals import coordination_observation # noqa: E402
from loopx.control_plane.status import collection as collection_read_model # noqa: E402


Expand Down Expand Up @@ -104,9 +105,15 @@ def assert_wrapper_parity(registry_path: Path, runtime_root: Path) -> None:
"goal_id": GOAL_ID,
"available_capabilities": ["network", "material_lifecycle"],
}
# Compare one observation time without hiding freshness or coverage fields.
with patch.object(
collection_read_model, "now_utc_iso", return_value="2026-01-01T00:00:00Z"
# Both adapters sample time independently; keep complete freshness/provider
# comparisons deterministic even when the two reads cross a wall-clock second.
with (
patch.object(
collection_read_model, "now_utc_iso", return_value="2026-01-01T00:00:00Z"
),
patch.object(
coordination_observation, "now_local_iso", return_value="2026-01-01T00:00:00Z"
),
):
wrapper = status_module.collect_status(**kwargs)
direct = collection_read_model.collect_status(
Expand Down
4 changes: 3 additions & 1 deletion examples/project/configure-goal-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,9 @@ def goal_from_registry(registry_path: Path) -> dict:

def main() -> int:
with tempfile.TemporaryDirectory(prefix="loopx-configure-goal-smoke-") as tmp:
root = Path(tmp)
# Canonical import binds the physical project path, including when the
# temporary directory is reached through a filesystem alias.
root = Path(tmp).resolve()
registry_path = write_registry(root)
original = registry_path.read_text(encoding="utf-8")
agent_profile = {
Expand Down
Loading