diff --git a/experiments/perception/OBSERVATORY_RECORDED_PROGRESS_2026-09-03.md b/experiments/perception/OBSERVATORY_RECORDED_PROGRESS_2026-09-03.md new file mode 100644 index 0000000..b0b9a1c --- /dev/null +++ b/experiments/perception/OBSERVATORY_RECORDED_PROGRESS_2026-09-03.md @@ -0,0 +1,115 @@ +# Recorded Observatory — attempt-bound progress and first real run + +Date: 2026-09-03. Current implementation: `bee8552`. +This is an observation/control-plane increment of stage 2, not onboard or +full recorded-analysis acceptance. No Synology release, physical acquisition, +actuation, model replacement or GPU benchmark is involved. + +## Implemented and activated + +- Worker counters are sampled independently of execution and sent through the + authenticated Worker gateway. Claim owner, generation, sequence and monotonic + counters are checked transactionally. Observation neither renews a lease nor + changes job identity, terminal state, artifact integrity or publication. +- One bounded snapshot per job is stored in a three-column SQLite table; + snapshots are at most 2048 bytes. Existing job JSON and six previous tables + are unchanged. The existing 128 MiB queue quota is not raised. +- A source/profile/definition/attempt-bound read endpoint supplies the existing + Observatory activity indicator and visible domain copy. No new per-LAB app, + generic visual control, completion badge or completion button was introduced. +- Phases: input delivery, input preparation, computation, result assembly and + output delivery; publication uses the existing queue state. M49 computation + counts complete ordered rows from the real TGS timing output. LAB V1 exposes + completed package operations, not invented per-frame inference counts. +- The two existing control agents receive an eight-file offline child layer. + Compute package declarations, compiled TGS runner, models, profile definition, + GPU allocation and resource limits are preserved. Previous agents remain + stopped with `restart=no`; no old and new owner are intentionally run together. + +## Validation before the real run + +- Frontend: architecture tests, typecheck, **726 unit tests**, production build. + Tests and build ran sequentially; memory pressure remained level 1 and swap + was approximately 2594 MiB. The existing large-chunk Vite warning remains. +- Backend/control tests: **128 focused tests** including payload admission, + owner/generation fencing, duplicate/regressive snapshots, sender connection + failure, queue persistence, HTTP authentication, old/new transport compatibility, + actual subprocess invocation and timeout child cleanup. A separate **40 tests** + cover portable runtime, M49 entrypoint/queue/LAB V1 wiring and minimal imports. + Ruff and focused mypy pass. The existing TestClient deprecation warning remains. +- Canonical8000 was replaced only while the queue was idle. Previous UI output + and a SQLite backup were retained. All rows in the six pre-existing tables + matched their pre-update hashes; the new progress table was empty. +- Browser: actual source/profile selected, normal and expanded window checked, + dropdown Escape checked, no Calculate action while that profile runs. Reload + and reopening/selecting M49 recover the same job and generation, not a new job. + The workspace maximize action itself is not an Escape-to-restore contract. + +## Exact control-agent installation + +Plan SHA-256: +`108897767de320562b4f75297e1c49d0ae5c44770f9c820d92d127f0fdad9216`. +Installer SHA-256: +`13edd1dd4e2579de788721de5ff397f57a5e1af0e4c59593c4af5a29f1fc2a77`. +Started `2026-09-03T08:06:19.107635Z`, finished +`2026-09-03T08:06:48.126401Z`. Receipt retains UTC and monotonic timestamps. + +| Control agent | New immutable image | Retained predecessor | +|---|---|---| +| `ndc-observatory-m49-worker-agent` | `927c3c4f5b00ae6c084d1f5a8bc77f7b262cfc5e83cf4be1c48f615c12c06e80` | `ndc-observatory-m49-worker-agent-pre-progress-7a3e07929c82` | +| `ndc-observatory-installed-lab-worker-agent` | `052af3ccd10e11b162c163943f5427af2dc95b09b93481dd85e954e494ba1107` | `ndc-observatory-installed-lab-worker-agent-pre-progress-a8e8353d4eb9` | + +Durable release: +`D:\NDC_MISSIONCORE\runtime\services\observatory-progress-20260903-DbxaPA\release`. +Its exact `*-declaration.json` files supersede the previous claim-v3 control-agent +declarations. Package manifests remain the compute source of truth. The installer +rechecks the exact container and complete Config/HostConfig hash immediately +before cutover. The loaded Python file hashes and running state are verified +after replacement. Only temporary offline layer/helper containers are removed. + +Rollback is explicit, idle-only and per agent: prove no open recorded/live work, +retain current declarations/evidence, stop the exact replacement, then restore +the corresponding predecessor's name and previous restart policy. Do not start +predecessors alongside replacements, use the old package launcher to overwrite +the control layer, or restore an old SQLite backup over newer work. No reboot +was performed; installed restart declarations, not reboot recovery, were checked. + +## Real run — status at 08:20 UTC + +One ordinary UI Calculate action submitted: + +- Source: RAVNOVES004TREE / `20260828T130511Z_viewer_live`, source duration 13:29. +- Setup: `m49-tgs-portable-v2`, definition + `f56d6321bd794ccdfb7d2e3b05d044b11f616ffb81ee29517386cc253046d4eb`. +- Job: `observatory-run-b230216709dc4c59bc56c98c7e329bf1`, generation 1. +- Input inventory: 6837 members. Delivery completed after approximately 298 s; + input preparation is still running. There is no result/cache/view acceptance yet. +- Queue: 10 retained failures, one retained historical success/not-required, + one running job, one current progress row. No second model/profile was started. + +An early diagnosis based on a stationary 556912640-byte temporary camera archive +did **not** prove a network stall. That equals the sealed archive's complete +length; the job subsequently advanced normally. Do not report it as a proven +transport outage or benchmark. + +## Remaining boundaries — not hidden by the progress feature + +1. This does not introduce `recorded-analysis` into the retained live runtime. + No live freshness, latest-wins, ownership or recovery rule is weakened. +2. Existing M49 still delivers the whole admitted input before TGS. The current + LiDAR builder performs full capture passes **before** its replay-pack cache + lookup (`compute/lidar_replay.py`, `build_lidar_replay_pack_v2`). Eliminating + this barrier needs its own versioned input/runtime change and equivalence proof. +3. Camera archive download/verification and initial LiDAR decode have no internal + byte/message counter yet. File/frame counts may stay unchanged within those + operations. Unknown totals remain unknown; no made-up percentage or ETA. +4. Phase snapshots are not a complete performance trace. Warmup/stage timings, + throughput, all required-frame accounting and onboard estimates need separate + evidence. These UI counts are not FPS or real-time qualification. +5. Stage 2B remains open until this run publishes an exact, reviewable result and + cached reopening is verified. LAB V1 and the other recording follow sequentially. + +Private evidence, not Git: `.runtime/observatory-progress-DbxaPA/` contains the +SQLite backup and row hashes, payload/plan/receipt/declarations, build/test logs, +submitted job and bounded read-only progress observations. Model assets and +recorded camera/LiDAR payloads are not copied into this report. diff --git a/experiments/perception/worker/observatory_portable/install_recorded_progress.py b/experiments/perception/worker/observatory_portable/install_recorded_progress.py new file mode 100644 index 0000000..51cc1b4 --- /dev/null +++ b/experiments/perception/worker/observatory_portable/install_recorded_progress.py @@ -0,0 +1,308 @@ +"""Hash-gated observation-only layer for the two existing recorded agents. + +Pack locally, plan read-only on Worker, then apply the exact plan. No downloads, +model changes, resource changes or live jobs. Stopped predecessors are retained. +""" + +from __future__ import annotations + +import argparse +import copy +import io +import json +import tarfile +import time +from datetime import UTC, datetime +from pathlib import Path +from urllib.parse import urlencode + +from migrate_claim_transport_v3 import ( + READINESS, + Engine, + canonical, + require_idle, + save, + sha, +) + +SOURCE_ROOT = "/opt/nodedc/installed-lab/src/k1link/observatory" +BEFORE = { + "worker_agent.py": "cf81131746ff4c4a3af6b186a852238f2cdde7e66c70aaa8ac1e9a3dc66bde7c", + "worker_http_transport.py": "81bdc27f374cd91358f0eb58603e3e1f53e1f3017bf158d269e75affa4567293", + "m49_portable_executor.py": "19a1b5bdb9c8dfc6a91af42ebc648d7f525cce95637f8a368c7679148b9666f0", + "m49_portable_source.py": "ab178740e5e577d00863cca6d1d1963acad766b3aa6d7a19f9de870953131fca", + "portable_worker_runtime.py": ( + "1c62636a6da24cae0b9e632ccd34108043dab7d2b0aefba7868376e3ca3feaf7" + ), + "installed_lab_package_runner.py": ( + "43195cc53e524ff57307ead691441cdd7c2ba09949bbdae6fe77920a63263eac" + ), + "recorded_progress.py": None, + "m49_timing_progress.py": None, +} +TARGETS = { + "ndc-observatory-m49-worker-agent": ( + "7aa6ccd2ddba4ebb0c07d793e5331539a2e13b07961c269439d5b79762328194" + ), + "ndc-observatory-installed-lab-worker-agent": ( + "1b1e335916c1c3d77888b7537331e9d775d078957c4ac0f82e91852301725390" + ), +} + + +def probe() -> str: + return f"""import hashlib, json, pathlib +from k1link.observatory import worker_agent +root=pathlib.Path(worker_agent.__file__).parent +assert str(root) == {SOURCE_ROOT!r} +print(json.dumps({{name: hashlib.sha256((root/name).read_bytes()).hexdigest() + if (root/name).is_file() else None for name in {list(BEFORE)!r}}})) +""" + + +def pack(repository: Path, output: Path) -> None: + output.mkdir(parents=False, exist_ok=False) + files = {} + for name in BEFORE: + payload = (repository / "src/k1link/observatory" / name).read_bytes() + compile(payload, name, "exec") + (output / name).write_bytes(payload) + files[name] = sha(payload) + save(output / "payload.json", {"schema_version": 1, "files": files}) + + +def payload_files(root: Path) -> dict[str, bytes]: + manifest = json.loads((root / "payload.json").read_bytes()) + if set(manifest) != {"schema_version", "files"} or manifest["schema_version"] != 1: + raise ValueError("invalid progress payload manifest") + if set(manifest["files"]) != set(BEFORE): + raise ValueError("progress file set changed") + files = {} + for name, digest in manifest["files"].items(): + path = root / name + if path.is_symlink() or not path.is_file() or path.stat().st_size > 256_000: + raise ValueError("unsafe progress payload") + value = path.read_bytes() + if sha(value) != digest: + raise ValueError("progress payload hash mismatch") + compile(value, name, "exec") + files[name] = value + return files + + +def validate_target(name: str, row: dict) -> None: + host, config = row["HostConfig"], row["Config"] + if row["Name"] != "/" + name or row["Image"] != "sha256:" + TARGETS[name]: + raise ValueError("agent identity changed") + if not row["State"]["Running"] or not host["ReadonlyRootfs"]: + raise ValueError("agent is not in its expected running/read-only state") + if host["NetworkMode"] != "bridge" or host.get("DeviceRequests") or host.get("Privileged"): + raise ValueError("agent GPU/network/privilege boundary changed") + if config["Labels"].get("com.nodedc.authority") != "observation-only": + raise ValueError("agent authority changed") + for entry in config["Env"]: + key = entry.split("=", 1)[0].upper() + if any(word in key for word in ("TOKEN", "PASSWORD", "SECRET")) and not key.endswith( + "_FILE" + ): + raise ValueError("inline secret in declaration") + + +def create_hash(row: dict) -> str: + return sha(canonical({"Config": row["Config"], "HostConfig": row["HostConfig"]})) + + +def validate_fence(target: dict, row: dict) -> None: + validate_target(target["name"], row) + if row["Id"] != target["id"] or create_hash(row) != target["create_sha256"]: + raise ValueError("agent or declaration changed since plan") + + +def plan(engine: Engine, root: Path) -> dict: + files = payload_files(root) + targets = [] + for name in TARGETS: + row = engine.inspect(name) + validate_target(name, row) + if engine.execute_json(name, probe()) != BEFORE: + raise ValueError("imported agent code is not the reviewed baseline") + require_idle(engine.execute_json(name, READINESS)) + targets.append( + { + "name": name, + "id": row["Id"], + "parent": TARGETS[name], + "create_sha256": create_hash(row), + } + ) + return { + "schema_version": "missioncore.recorded-progress-install-plan/v1", + "targets": targets, + "files": {name: sha(value) for name, value in files.items()}, + "installer_sha256": sha(Path(__file__).read_bytes()), + "engine_helper_sha256": sha( + Path(__file__).with_name("migrate_claim_transport_v3.py").read_bytes() + ), + "compute_packages_changed": False, + } + + +def build(engine: Engine, target: dict, files: dict[str, bytes], plan_sha: str) -> str: + created = engine.request( + "POST", + "/containers/create", + { + "Image": "sha256:" + target["parent"], + "Entrypoint": ["/bin/true"], + "Cmd": [], + "HostConfig": { + "NetworkMode": "none", + "CapDrop": ["ALL"], + "PidsLimit": 32, + "SecurityOpt": ["no-new-privileges"], + }, + }, + )["Id"] + try: + engine.request("POST", f"/containers/{created}/start") + if engine.request("POST", f"/containers/{created}/wait")["StatusCode"] != 0: + raise ValueError("offline layer initialization failed") + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode="w") as archive: + for name, payload in files.items(): + member = tarfile.TarInfo(name) + member.size, member.mode, member.mtime = len(payload), 0o644, int(time.time()) + archive.addfile(member, io.BytesIO(payload)) + engine.request( + "PUT", + f"/containers/{created}/archive?" + urlencode({"path": SOURCE_ROOT}), + buffer.getvalue(), + ) + changes = engine.request("GET", f"/containers/{created}/changes") + allowed = {str(Path(SOURCE_ROOT) / name) for name in files} + parents = {str(p) for p in Path(SOURCE_ROOT).parents} | {SOURCE_ROOT} + if not changes or any( + item["Path"] not in allowed | parents or item["Kind"] not in (0, 1) for item in changes + ): + raise ValueError("unrelated filesystem changes in progress layer") + if not allowed.issubset({item["Path"] for item in changes}): + raise ValueError("a progress file was omitted from the image layer") + parent = engine.request("GET", f"/images/sha256:{target['parent']}/json") + config = copy.deepcopy(parent["Config"]) + config.setdefault("Labels", {}).update( + { + "com.nodedc.recorded-progress.plan-sha256": plan_sha, + "com.nodedc.recorded-progress.parent-sha256": target["parent"], + } + ) + image = engine.request( + "POST", + "/commit?" + + urlencode( + { + "container": created, + "repo": target["name"] + "-progress", + "tag": "v1", + } + ), + config, + )["Id"] + result = engine.request("GET", f"/images/{image}/json") + if result["RootFS"]["Layers"][:-1] != parent["RootFS"]["Layers"]: + raise ValueError("parent image layers changed") + return image + finally: + engine.request("DELETE", f"/containers/{created}") + + +def apply(engine: Engine, root: Path, expected: str, evidence: Path) -> dict: + started_at = datetime.now(UTC).isoformat() + started_mono = time.monotonic_ns() + proposal = plan(engine, root) + if sha(canonical(proposal)) != expected: + raise ValueError("progress install plan changed") + evidence.mkdir(parents=False, exist_ok=False) + save(evidence / "plan.json", proposal) + files = payload_files(root) + results = [] + for target in proposal["targets"]: + name = target["name"] + before = engine.inspect(name) + validate_fence(target, before) + image = build(engine, target, files, expected) + require_idle(engine.execute_json(name, READINESS)) + validate_fence(target, engine.inspect(name)) + body = copy.deepcopy(before["Config"]) + body["Image"] = image + body["Labels"]["com.nodedc.recorded-progress.plan-sha256"] = expected + body["HostConfig"] = copy.deepcopy(before["HostConfig"]) + backup = name + "-pre-progress-" + before["Id"][:12] + declaration = { + "name": name, + "create_body": body, + "rollback_name": backup, + "rollback_container_id": before["Id"], + "parent": target["parent"], + } + save(evidence / (name + "-declaration.json"), declaration) + engine.request("POST", f"/containers/{before['Id']}/stop?t=15") + engine.request( + "POST", f"/containers/{before['Id']}/update", {"RestartPolicy": {"Name": "no"}} + ) + engine.request("POST", f"/containers/{before['Id']}/rename?" + urlencode({"name": backup})) + created = engine.request("POST", "/containers/create?" + urlencode({"name": name}), body)[ + "Id" + ] + engine.request("POST", f"/containers/{created}/start") + # A started replacement may already own operator work; never auto-delete it. + time.sleep(3) + after = engine.inspect(name) + if not after["State"]["Running"] or after["RestartCount"] != 0: + raise ValueError("replacement needs reconciliation; predecessor retained") + if engine.execute_json(name, probe()) != proposal["files"]: + raise ValueError("replacement imported another progress payload") + results.append( + { + "name": name, + "id": created, + "image": image, + "rollback": backup, + "readiness": engine.execute_json(name, READINESS), + } + ) + save(evidence / (name + "-acceptance.json"), results[-1]) + receipt = { + "plan_sha256": expected, + "agents": results, + "compute_packages_changed": False, + "started_at_utc": started_at, + "finished_at_utc": datetime.now(UTC).isoformat(), + "started_monotonic_ns": started_mono, + "finished_monotonic_ns": time.monotonic_ns(), + } + save(evidence / "receipt.json", receipt) + return receipt + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repository", type=Path) + parser.add_argument("--pack", type=Path) + parser.add_argument("--payload", type=Path) + parser.add_argument("--apply-plan-sha256") + parser.add_argument("--evidence", type=Path) + args = parser.parse_args() + if args.pack: + pack(args.repository, args.pack) + return + engine = Engine() + if args.apply_plan_sha256: + result = apply(engine, args.payload, args.apply_plan_sha256, args.evidence) + else: + proposal = plan(engine, args.payload) + result = {"plan": proposal, "plan_sha256": sha(canonical(proposal))} + print(json.dumps(result, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_observatory_progress_install.py b/tests/test_observatory_progress_install.py new file mode 100644 index 0000000..9f3a4b3 --- /dev/null +++ b/tests/test_observatory_progress_install.py @@ -0,0 +1,62 @@ +"""Control-agent layer admission; no Docker or model execution.""" + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +SCRIPTS = Path(__file__).parents[1] / "experiments/perception/worker/observatory_portable" + + +@pytest.fixture +def installer(monkeypatch): + monkeypatch.syspath_prepend(str(SCRIPTS)) + spec = importlib.util.spec_from_file_location( + "recorded_progress_installer", SCRIPTS / "install_recorded_progress.py" + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, spec.name, module) + spec.loader.exec_module(module) + return module + + +def test_pack_is_exact_bounded_code_only_and_hash_gated(installer, tmp_path): + output = tmp_path / "payload" + installer.pack(Path(__file__).parents[1], output) + assert set(installer.payload_files(output)) == set(installer.BEFORE) + with pytest.raises(FileExistsError): + installer.pack(Path(__file__).parents[1], output) + file = output / "worker_agent.py" + file.write_bytes(file.read_bytes() + b"\n# modified\n") + with pytest.raises(ValueError, match="hash mismatch"): + installer.payload_files(output) + + +def test_manifest_cannot_add_unreviewed_paths(installer, tmp_path): + output = tmp_path / "payload" + installer.pack(Path(__file__).parents[1], output) + manifest = json.loads((output / "payload.json").read_text()) + manifest["files"]["../../outside.py"] = "0" * 64 + (output / "payload.json").write_text(json.dumps(manifest)) + with pytest.raises(ValueError, match="file set changed"): + installer.payload_files(output) + + +def test_cutover_rechecks_exact_container_and_declaration(installer): + name = next(iter(installer.TARGETS)) + row = { + "Id": "a" * 64, + "Name": "/" + name, + "Image": "sha256:" + installer.TARGETS[name], + "State": {"Running": True}, + "HostConfig": {"ReadonlyRootfs": True, "NetworkMode": "bridge"}, + "Config": {"Labels": {"com.nodedc.authority": "observation-only"}, "Env": []}, + } + target = {"name": name, "id": row["Id"], "create_sha256": installer.create_hash(row)} + installer.validate_fence(target, row) + row["HostConfig"]["Memory"] = 1024 + with pytest.raises(ValueError, match="changed since plan"): + installer.validate_fence(target, row)