Something went wrong. Try again.
Architecture research toward an agent-native operating system; rigorous, pre-prototype, and intentionally self-aware.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285#!/usr/bin/env python3"""Narrow external capture boundary for D-004 (standard library + Git only)."""import argparseimport hashlibimport jsonimport osfrom pathlib import Pathimport reimport subprocessimport sysimport tarfileimport tempfile
try: from . import d004_manifest as Vexcept ImportError: import d004_manifest as V
TEST_PATHS = list(V.TEST_PATHS)TEST_MODULES = list(V.TEST_MODULES)EXPECTED_TESTS = 45DEFAULT_REQUIRED_PATHS = list(V.REQUIRED_PACKAGE_PATHS)_HEX40 = re.compile(r"[0-9a-f]{40}")_COMMITTER = re.compile(rb"committer [^\n]* <[^<>\n]+> ([0-9]+) ([+-](?:0[0-9]|1[0-4])[0-5][0-9])$")_RESULT_PREFIX = b"D004_RUNTIME_RESULT="# Audited fixed bootstrap. -S leaves the standard library paths in sys.path;# retain only base-prefix stdlib paths, remove cwd/user/site paths, then insert# the explicit archive export root as the sole application import root.BOOTSTRAP = b'''import os,sys\nroot=os.path.realpath(sys.argv[1])\nmods=sys.argv[2:]\nbase=os.path.realpath(sys.base_prefix)+os.sep\nkeep=[]\nfor p in sys.path:\n q=os.path.realpath(p) if p else ""\n if q.startswith(base) and "site-packages" not in q and "dist-packages" not in q: keep.append(p)\nsys.path[:]=[root]+keep\nimport io,json,unittest\nstream=io.StringIO()\nsuite=unittest.defaultTestLoader.loadTestsFromNames(mods)\nresult=unittest.TextTestRunner(stream=stream,verbosity=0).run(suite)\nrecord={"errors":len(result.errors),"executed":result.testsRun,"failures":len(result.failures),"success":result.wasSuccessful()}\nsys.stdout.write("D004_RUNTIME_RESULT="+json.dumps(record,sort_keys=True,separators=(",",":"))+"\\n")\nraise SystemExit(0 if result.wasSuccessful() else 1)\n'''BOOTSTRAP_ARTIFACT = V.artifact_identity("PYTHON_BOOTSTRAP", BOOTSTRAP)
class CaptureError(ValueError): pass
def _run(args, *, cwd=None, data=None, check=True, env=None): p = subprocess.run([os.fspath(x) for x in args], cwd=cwd, input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) if check and p.returncode: raise CaptureError("command failed: " + " ".join(map(os.fspath, args)) + ": " + p.stderr.decode("utf-8", "replace").strip()) return p
def _git(repo, *args, data=None, check=True): return _run(["git", "-C", repo, *args], data=data, check=check)
def _sha(data): return hashlib.sha256(data).hexdigest()
def artifact(kind, data): return {"kind": kind, "sha256": _sha(data), "byte_length": len(data)}
def _require_sha1(repo): fmt = _git(repo, "rev-parse", "--show-object-format").stdout.strip() if fmt != b"sha1": raise CaptureError("Git object format must be sha1")
def _safe_inventory(expected): if not isinstance(expected, dict) or not expected: raise CaptureError("required path inventory") try: normalized = V._normalized_map(expected) except Exception as e: raise CaptureError("required path inventory") from e if normalized != expected: raise CaptureError("required path inventory normalization") return normalized
def _tree_entries(repo, commit): raw = _git(repo, "ls-tree", "-rz", "--full-tree", "-r", commit).stdout result = {} for record in raw.split(b"\0"): if not record: continue head, path = record.split(b"\t", 1) mode, typ, oid = head.split(b" ") try: name = path.decode("utf-8") except UnicodeDecodeError as e: raise CaptureError("non-UTF-8 committed path") from e if name in result: raise CaptureError("duplicate committed path") result[name] = (mode.decode(), typ.decode(), oid.decode()) return result
def _controlled_clean(repo, commit): head = _git(repo, "rev-parse", "HEAD").stdout.decode().strip() status = _git(repo, "status", "--porcelain=v1", "-z", "--untracked-files=all").stdout if head != commit or status: raise CaptureError("controlled capture requires clean current HEAD") # Ignored files are deliberately out of scope: they are absent from both # the committed Git tree and `git archive`, and therefore cannot affect it.
def capture_package_commit(repo, commit, expected_paths): """Bind ten selected package files and every regular blob in a monorepo commit.""" repo = Path(repo).resolve(); expected = _safe_inventory(expected_paths) if list(expected) != DEFAULT_REQUIRED_PATHS: raise CaptureError("exact ordered DEFAULT_REQUIRED_PATHS required") _require_sha1(repo) if not _HEX40.fullmatch(commit): raise CaptureError("commit must be a full lowercase SHA-1") typ = _git(repo, "cat-file", "-t", commit, check=False) if typ.returncode or typ.stdout != b"commit\n": raise CaptureError("object does not exist or is not a commit") payload = _git(repo, "cat-file", "commit", commit).stdout tree = _git(repo, "rev-parse", f"{commit}^{{tree}}").stdout.decode().strip() parsed = V.parse_git_commit(payload, tree) if parsed["commit"] != commit: raise CaptureError("commit object hash mismatch") matches = [_COMMITTER.fullmatch(x) for x in payload.split(b"\n") if x.startswith(b"committer ")] matches = [x for x in matches if x] if len(matches) != 1: raise CaptureError("unique raw committer required") epoch, zone = (x.decode() for x in matches[0].groups()) entries = _tree_entries(repo, commit); all_bytes = {}; leaves = [] for path in sorted(entries): mode, object_type, oid = entries[path] if mode != "100644" or object_type != "blob": raise CaptureError("full tree contains unrepresentable mode/type") data = _git(repo, "cat-file", "blob", oid).stdout if V.git_blob(data) != oid: raise CaptureError("committed blob OID mismatch") all_bytes[path] = data; leaf = V.package_leaf(path, data) if leaf["blob_oid"] != oid: raise CaptureError("package leaf OID mismatch") leaves.append(leaf) if not set(expected).issubset(all_bytes): raise CaptureError("missing selected required path") for path, wanted in expected.items(): if all_bytes[path] != wanted: raise CaptureError("selected required bytes mismatch") _controlled_clean(repo, commit) return {"repo":str(repo),"commit":commit,"tree":tree,"commit_payload":payload,"timestamp_token":f"GIT_COMMITTER_TIMESTAMP:{epoch}:{zone}","selected":dict(expected),"all_bytes":all_bytes,"full_tree":leaves}
def _test_only_validate_probe_values(exe, returncode, stdout, stderr): """Parse probe values for tests; this helper cannot run or yield a capture.""" try: values = json.loads(stdout) except Exception as e: raise CaptureError("Python probe failed") from e if returncode or stderr or not isinstance(values, list) or len(values) != 3 or values[:2] != ["CPython", "3.11.14"]: raise CaptureError("CPython 3.11.14 required") reported = Path(values[2]).resolve() if reported != exe: raise CaptureError("Python probe executable mismatch")
def probe_cpython(executable): exe = Path(executable).resolve() if not exe.is_file(): raise CaptureError("explicit Python executable required") code = "import json,platform,sys;print(json.dumps([platform.python_implementation(),platform.python_version(),sys.executable]))" p = _run([exe, "-I", "-c", code], check=False) _test_only_validate_probe_values(exe, p.returncode, p.stdout, p.stderr) raw = exe.read_bytes() return exe, artifact("PYTHON_INTERPRETER", raw)
def _export_commit(repo, commit, destination, expected_full_tree): archive = _git(repo, "archive", "--format=tar", commit).stdout destination.mkdir(); expected = {x["path"]: V._decode_raw(x["content"], "expected tree") for x in expected_full_tree}; observed = {} with tempfile.NamedTemporaryFile() as f: f.write(archive); f.flush() with tarfile.open(f.name, "r:") as tf: for member in tf.getmembers(): if member.isdir() or member.type in (tarfile.XHDTYPE, tarfile.XGLTYPE): continue if not member.isfile() or (member.mode & 0o111): raise CaptureError("archive mode/type mismatch") member.mode = 0o644 try: path = member.name.encode("utf-8").decode("utf-8") except UnicodeError as e: raise CaptureError("archive non-UTF-8 path") from e if path in observed: raise CaptureError("duplicate archive path") stream = tf.extractfile(member); observed[path] = stream.read() if stream else b"" if observed != expected: raise CaptureError("archive tree bytes mismatch") tf.extractall(destination, filter="data") for path, data in expected.items(): q = destination / path if not q.is_file() or q.read_bytes() != data or (q.stat().st_mode & 0o777) != 0o644: raise CaptureError("exported tree mismatch")
def _minimal_python_env(): env = {"PATH":os.environ.get("PATH", ""), "LANG":"C.UTF-8", "LC_ALL":"C.UTF-8"} for key in ("PYTHONPATH","PYTHONHOME","PYTHONSTARTUP","PYTHONUSERBASE","PYTHONINSPECT","PYTHONWARNINGS"): env.pop(key, None) return env
def runtime_test_capture(package, executable): exe, interpreter = probe_cpython(executable) selected = package["selected"] if list(selected) != DEFAULT_REQUIRED_PATHS: raise CaptureError("runtime exact selected inventory") closure = {"package_commit":package["commit"],"package_tree":package["tree"],"tested_files":[{"path":p,"sha256":_sha(selected[p]),"byte_length":len(selected[p])} for p in DEFAULT_REQUIRED_PATHS]} with tempfile.TemporaryDirectory() as td: root = Path(td,"export"); _export_commit(package["repo"],package["commit"],root,package["full_tree"]) command=[str(exe),"-I","-S","-c",BOOTSTRAP.decode("utf-8"),str(root),*TEST_MODULES] p=_run(command,cwd=Path(td),check=False,env=_minimal_python_env()) rows=[line for line in p.stdout.splitlines() if line.startswith(_RESULT_PREFIX)] if len(rows)!=1 or p.stderr or any(line and not line.startswith(_RESULT_PREFIX) for line in p.stdout.splitlines()): raise CaptureError("runtime output record") try: record=json.loads(rows[0][len(_RESULT_PREFIX):]) except Exception as e: raise CaptureError("runtime result JSON") from e expected_record={"errors":0,"executed":45,"failures":0,"success":True} if p.returncode!=0 or record!=expected_record: raise CaptureError("runtime exact result") normalized=["CPYTHON","-I","-S","-c",BOOTSTRAP_ARTIFACT["sha256"],"<EXPORT_ROOT>",*TEST_MODULES] result={"schema":"D004-RUNTIME-TEST-CAPTURE-2","implementation":"CPython","version":"3.11.14","interpreter":interpreter,"bootstrap":BOOTSTRAP_ARTIFACT,"package_commit":package["commit"],"package_tree":package["tree"],"test_paths":TEST_PATHS[:],"actual_normalized_command":normalized,"expected_count":45,"executed_count":45,"exit_status":0,"tested_closure":V.artifact_identity("TESTED_CLOSURE",V.canonical(closure))} V._runtime_test_capture(result,selected,package["commit"],package["tree"]) return result
def package_capture(package, runtime_capture, executor_role, checker_role, packet_raw, manifest_raw): """Construct and cross-validate the exact corrected package handoff.""" eb,cb=executor_role["bundle"],checker_role["bundle"] for role_result in (executor_role,checker_role): if role_result.get("derived_package_commit")!=package["commit"] or role_result.get("derived_package_tree")!=package["tree"]: raise CaptureError("role result not derived from package") if role_result["bundle_raw"]!=V.raw_bytes(role_result["bundle_bytes"]) or role_result["bundle_artifact"]!=V.bundle_identity(role_result["bundle"]["role"],role_result["bundle_bytes"]): raise CaptureError("role bundle material") verify_role_bundle(role_result["bundle_bytes"],role_result["bundle"]) result={"schema":"D004-PACKAGE-CAPTURE-2","commit":package["commit"],"tree":package["tree"],"timestamp_token":package["timestamp_token"],"commit_payload":V.raw_bytes(package["commit_payload"]),"full_tree":package["full_tree"],"selected_required":[{"path":p,"content":V.raw_bytes(package["selected"][p])} for p in DEFAULT_REQUIRED_PATHS],"executor_bundle":executor_role["bundle_raw"],"checker_bundle":checker_role["bundle_raw"],"executor_bundle_artifact":executor_role["bundle_artifact"],"checker_bundle_artifact":checker_role["bundle_artifact"],"runtime_test_capture":runtime_capture} V._package_capture(result,packet_raw,manifest_raw,eb,cb) return result
def _write_files(root, files): for path, data in files.items(): q = root / path; q.parent.mkdir(parents=True, exist_ok=True); q.write_bytes(data); q.chmod(0o644)
def capture_role_bundle(role, sources, dependencies, locks, o_bytes, bundle_output_dir, *, epoch=1700000000, zone="+0000"): """Build, verify and externally bundle one strict isolated zero-parent role commit.""" contract = V.provenance_contract(); root_name = contract["source_roots"].get(role) if root_name is None: raise CaptureError("role") sets = [set(sources), set(dependencies), set(locks)] if any(sets[i] & sets[j] for i in range(3) for j in range(i + 1, 3)): raise CaptureError("overlapping closure maps") tracked = {**sources, **dependencies, **locks} if any(Path(p).name.startswith("test") or "/test" in p for p in tracked): raise CaptureError("test files excluded from role closure") pure_tree = V.build_git_tree(tracked)["root"] output = Path(bundle_output_dir).resolve(); output.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory() as td: repo = Path(td, "role"); repo.mkdir(); _git(repo, "init", "-q"); _require_sha1(repo); _write_files(repo, tracked) _git(repo, "add", "--", *sorted(tracked)); tree = _git(repo, "write-tree").stdout.decode().strip() if tree != pure_tree: raise CaptureError("Git/pure tree mismatch") identity_line = f"D004 {role.title()} <d004-{role.lower()}@invalid> {epoch} {zone}".encode() payload = b"tree " + tree.encode() + b"\nauthor " + identity_line + b"\ncommitter " + identity_line + b"\n\nD-004 isolated " + role.encode() + b" capture\n" commit = _git(repo, "hash-object", "-t", "commit", "-w", "--stdin", data=payload).stdout.decode().strip() if V.git_commit(payload, tree) != commit or V.parse_git_commit(payload, tree)["parents"]: raise CaptureError("strict zero-parent commit mismatch") _git(repo, "update-ref", f"refs/d004/{role.lower()}", commit) _git(repo, "reset", "-q", "--hard", commit) if _git(repo, "status", "--porcelain=v1", "--untracked-files=all").stdout: raise CaptureError("role repo not clean") if set(_tree_entries(repo, commit)) != set(tracked): raise CaptureError("shared or monorepo role commit") _git(repo, "fsck", "--full", "--strict") bundle_path = output / f"d004-{role.lower()}-{commit}.bundle" _git(repo, "bundle", "create", bundle_path, f"refs/d004/{role.lower()}") _git(repo, "bundle", "verify", bundle_path) bundle_bytes = bundle_path.read_bytes() clean = {"head_commit":commit,"index_tree":tree,"worktree_tree":tree,"untracked_paths":[],"modified_paths":[],"tracked_paths":sorted(tracked),"tracked_paths_sha256":V.sha256_bytes(V.canonical(sorted(tracked)))} identity = V.producer_identity(role,sources,dependencies,locks,tracked,payload,clean,o_bytes,contract) bundle = {"identity":identity,"role":role,"sources":sources,"dependencies":dependencies,"locks":locks,"tracked":tracked,"commit_payload":payload,"clean_attestation":clean,"o_bytes":o_bytes} return {"bundle":bundle,"bundle_bytes":bundle_bytes,"bundle_raw":V.raw_bytes(bundle_bytes),"bundle_artifact":V.bundle_identity(role,bundle_bytes),"bundle_path":str(bundle_path),"commit":commit,"tree":tree,"ref":f"refs/d004/{role.lower()}"}
def capture_roles_from_package(package, o_bytes, bundle_output_dir): """Derive both role closures only from captured selected package bytes.""" selected=package["selected"]; locks={p:selected[p] for p in V.provenance_contract()["mandatory_locks"]}; out={} for role in ("EXECUTOR","CHECKER"): prefix="future/d004_"+role.lower()+"/" closure={p:selected[p] for p in DEFAULT_REQUIRED_PATHS if p.startswith(prefix) and not p.rsplit("/",1)[-1].startswith("test")} out[role.lower()]=capture_role_bundle(role,closure,{},locks,o_bytes,bundle_output_dir) out[role.lower()]["derived_package_commit"]=package["commit"] out[role.lower()]["derived_package_tree"]=package["tree"] return out
def verify_role_bundle(bundle_bytes, expected_bundle): """Reconstruct a role bundle and bind every Git byte/object to the pure bundle.""" if not isinstance(bundle_bytes, bytes) or not bundle_bytes: raise CaptureError("bundle bytes") role=expected_bundle["role"]; identity=expected_bundle["identity"]; commit=identity["git"]["commit"]["commit"] with tempfile.TemporaryDirectory() as td: path=Path(td,"capture.bundle"); path.write_bytes(bundle_bytes) repo=Path(td,"repo"); repo.mkdir(); _git(repo,"init","-q"); _git(repo,"bundle","verify",path); _git(repo,"fetch",path,commit); _git(repo,"update-ref",f"refs/d004/{role.lower()}",commit) _git(repo,"fsck","--full","--strict") payload=_git(repo,"cat-file","commit",commit).stdout if payload != expected_bundle["commit_payload"]: raise CaptureError("bundle commit payload") entries=_tree_entries(repo,commit); tracked=expected_bundle["tracked"] if set(entries)!=set(tracked): raise CaptureError("bundle closure inventory") for p,b in tracked.items(): mode,typ,oid=entries[p] if mode!="100644" or typ!="blob" or _git(repo,"cat-file","blob",oid).stdout!=b: raise CaptureError("bundle closure bytes") V.validate_producer_identity(identity,role,expected_bundle["sources"],expected_bundle["dependencies"],expected_bundle["locks"],tracked,expected_bundle["commit_payload"],expected_bundle["clean_attestation"],expected_bundle["o_bytes"],V.provenance_contract()) return True
def main(argv=None): p=argparse.ArgumentParser(); p.add_argument("repo"); p.add_argument("python"); p.add_argument("--inventory",required=True,help="JSON object in exact required-path order mapping paths to SHA-256"); p.add_argument("--bundle-dir",required=True); p.add_argument("--o-bytes",required=True) a=p.parse_args(argv); commit=_git(a.repo,"rev-parse","HEAD").stdout.decode().strip(); spec=json.loads(Path(a.inventory).read_text(),object_pairs_hook=dict); expected={} if list(spec)!=DEFAULT_REQUIRED_PATHS: raise CaptureError("inventory must be exact ordered required paths") for path,digest in spec.items(): raw=_git(a.repo,"show",f"{commit}:{path}").stdout if _sha(raw)!=digest: raise CaptureError("inventory digest mismatch") expected[path]=raw package=capture_package_commit(a.repo,commit,expected); runtime=runtime_test_capture(package,a.python); roles=capture_roles_from_package(package,Path(a.o_bytes).read_bytes(),a.bundle_dir) result=package_capture(package,runtime,roles["executor"],roles["checker"],expected[DEFAULT_REQUIRED_PATHS[0]],expected[DEFAULT_REQUIRED_PATHS[1]]) metadata={"schema":result["schema"],"commit":result["commit"],"tree":result["tree"],"timestamp_token":result["timestamp_token"],"executor_bundle_artifact":result["executor_bundle_artifact"],"checker_bundle_artifact":result["checker_bundle_artifact"],"runtime_test_capture":result["runtime_test_capture"]} print(V.canonical(metadata).decode(),end=""); return 0
if __name__ == "__main__": try: raise SystemExit(main()) except CaptureError as e: print(f"capture error: {e}",file=sys.stderr); raise SystemExit(2)