482 lines
17 KiB
Python
482 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Production full-generation peer conversation over a transport directory.
|
|
|
|
The directory is the object-level seam implemented by the MTP adapter in a
|
|
deployed bundle and directly supplied by isolated tests. Business documents are
|
|
always produced and consumed by the existing generation services.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from contextlib import closing, redirect_stdout
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import sync_generation_exchange as generation
|
|
|
|
CAPS = {
|
|
"generation-manifest-v1",
|
|
"generation-ack-v1",
|
|
"causal-delete-v1",
|
|
"mobile-history-v4",
|
|
"execution-draft-v1",
|
|
"generation-archive-v1",
|
|
"session-preparations-v2",
|
|
}
|
|
|
|
|
|
def emit(run_id: str, phase: str, **values: object) -> None:
|
|
value = {
|
|
"format": "trainlog-sync-progress",
|
|
"version": 1,
|
|
"run_id": run_id,
|
|
"phase": phase,
|
|
} | values
|
|
print(json.dumps(value, sort_keys=True, separators=(",", ":")), flush=True)
|
|
|
|
|
|
def bounded(path: Path, limit: int = 65536) -> bytes:
|
|
with path.open("rb") as stream:
|
|
raw = stream.read(limit + 1)
|
|
if len(raw) > limit:
|
|
raise RuntimeError(f"{path.name} exceeds bound")
|
|
return raw
|
|
|
|
|
|
def publish_json(path: Path, value: dict) -> None:
|
|
"""Publish a coordination object only after all bytes are durable."""
|
|
raw = generation.canonical(value)
|
|
if len(raw) > 65536:
|
|
raise RuntimeError(f"{path.name} exceeds bound")
|
|
temporary = path.with_name("." + path.name + ".tmp-" + str(os.getpid()))
|
|
with temporary.open("wb") as stream:
|
|
stream.write(raw)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.chmod(temporary, 0o600)
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def run_adapter(
|
|
adapter: Path,
|
|
operation: str,
|
|
peer: str,
|
|
root: Path,
|
|
deadline: float,
|
|
allow_missing_peer: bool = False,
|
|
) -> bool:
|
|
remaining = max(1, int(deadline - time.monotonic()))
|
|
timeout = min(remaining, 30)
|
|
try:
|
|
result = subprocess.run(
|
|
[str(adapter), operation, peer, str(root)],
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired as error:
|
|
# CONTRACT: an adapter timeout is an ambiguous transport failure. It
|
|
# never implies rollback, replay, or permission to retry a mutation.
|
|
raise RuntimeError(
|
|
f"transport_timeout: MTP {operation} did not finish within {timeout} seconds"
|
|
) from error
|
|
if result.returncode != 0:
|
|
diagnostic = (
|
|
result.stderr.decode("utf-8", "replace")[:1024] or "MTP adapter failed"
|
|
).strip()
|
|
missing_peer = diagnostic in {
|
|
"double failed 6 expected Android peer not found",
|
|
"MTP adapter failed status=6 diagnostic=expected Android peer not found",
|
|
}
|
|
if allow_missing_peer and missing_peer:
|
|
return False
|
|
raise RuntimeError(diagnostic)
|
|
return True
|
|
|
|
|
|
def push_adapter(
|
|
adapter: Path,
|
|
peer: str,
|
|
root: Path,
|
|
relative_paths: list[str],
|
|
deadline: float,
|
|
) -> None:
|
|
"""Publish only the phase-owned bounded MTP objects.
|
|
|
|
WHY: the durable transport root also retains inbound objects, legacy files,
|
|
and staging evidence; recursively uploading that root makes each exchange
|
|
grow with history and can time out before Android sees the request.
|
|
CONTRACT: callers name fixed relative files/directories already durably
|
|
published below ``root``. The adapter receives a private disposable view.
|
|
INVARIANT: no retained generation, ACK, tombstone, or source object is
|
|
deleted or modified while constructing the view.
|
|
"""
|
|
with tempfile.TemporaryDirectory(prefix="trainlog-mtp-outbox-", dir=root.parent) as directory:
|
|
outbox = Path(directory)
|
|
os.chmod(outbox, 0o700)
|
|
for relative in relative_paths:
|
|
source = root / relative
|
|
destination = outbox / relative
|
|
if (
|
|
source.is_symlink()
|
|
or not source.exists()
|
|
or not source.resolve().is_relative_to(root.resolve())
|
|
):
|
|
raise RuntimeError(f"invalid MTP outbox source: {relative}")
|
|
if source.is_dir() and any(path.is_symlink() for path in source.rglob("*")):
|
|
raise RuntimeError(f"MTP outbox source contains a symbolic link: {relative}")
|
|
destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
if source.is_dir():
|
|
shutil.copytree(source, destination, copy_function=shutil.copy2)
|
|
elif source.is_file():
|
|
shutil.copy2(source, destination)
|
|
else:
|
|
raise RuntimeError(f"invalid MTP outbox object: {relative}")
|
|
run_adapter(adapter, "push", peer, outbox, deadline)
|
|
|
|
|
|
def push_desktop_generation(
|
|
adapter: Path,
|
|
peer: str,
|
|
root: Path,
|
|
relative_path: str,
|
|
deadline: float,
|
|
) -> None:
|
|
"""Commit immutable bytes before making their correlated reference visible."""
|
|
push_adapter(adapter, peer, root, [relative_path], deadline)
|
|
push_adapter(
|
|
adapter,
|
|
peer,
|
|
root,
|
|
[
|
|
"request-v1.json",
|
|
"desktop-archive-acknowledgements-v1.json",
|
|
"desktop-consumption-ack-v1.json",
|
|
"desktop-generation-v1.json",
|
|
],
|
|
deadline,
|
|
)
|
|
|
|
|
|
def wait_file(path: Path, deadline: float, pump=None) -> None:
|
|
while not path.is_file():
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError(f"timeout waiting for {path.name}")
|
|
if pump is not None:
|
|
pump()
|
|
time.sleep(0.05)
|
|
|
|
|
|
def wait_json(path: Path, deadline: float, predicate, pump=None) -> dict:
|
|
while True:
|
|
wait_file(path, deadline, pump)
|
|
try:
|
|
value = json.loads(bounded(path))
|
|
except (OSError, json.JSONDecodeError):
|
|
value = None
|
|
if isinstance(value, dict) and predicate(value):
|
|
return value
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError(f"timeout waiting for correlated {path.name}")
|
|
if pump is not None:
|
|
pump()
|
|
time.sleep(0.05)
|
|
|
|
|
|
def refresh_mtp_peer(
|
|
adapter: Path,
|
|
peer: str,
|
|
root: Path,
|
|
deadline: float,
|
|
) -> None:
|
|
"""Require a successful device pull before trusting a retained peer file.
|
|
|
|
WHY: the durable transport root legitimately retains coordination objects
|
|
from earlier runs. A local peer advertisement can therefore predate an
|
|
Android upgrade that adds a required capability.
|
|
CONTRACT: MTP mode validates only an advertisement obtained after at least
|
|
one successful pull of the expected physical peer during this run.
|
|
INVARIANT: an unavailable peer never makes a stale local advertisement
|
|
authoritative and remains a bounded wait state.
|
|
"""
|
|
peer_path = root / "android-peer-v1.json"
|
|
while True:
|
|
pulled = run_adapter(
|
|
adapter,
|
|
"pull",
|
|
peer,
|
|
root,
|
|
deadline,
|
|
True,
|
|
)
|
|
if pulled and peer_path.is_file():
|
|
return
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError("timeout waiting for android-peer-v1.json")
|
|
time.sleep(0.05)
|
|
|
|
|
|
def correlated_peer_error(root: Path, run_id: str) -> None:
|
|
path = root / "android-generation-error-v1.json"
|
|
if not path.is_file():
|
|
return
|
|
try:
|
|
value = json.loads(bounded(path))
|
|
except (OSError, json.JSONDecodeError):
|
|
return
|
|
if (
|
|
isinstance(value, dict)
|
|
and value.get("format") == "trainlog-sync-generation-error"
|
|
and value.get("version") == 1
|
|
and value.get("run_id") == run_id
|
|
and value.get("code") == "peer_capacity_exhausted"
|
|
):
|
|
raise RuntimeError(
|
|
"peer_capacity_exhausted: Android active generation capacity is exhausted; "
|
|
"archive acknowledged generations, then retry explicitly"
|
|
)
|
|
|
|
|
|
def load_peer(root: Path, expected: str) -> dict:
|
|
value = json.loads(bounded(root / "android-peer-v1.json"))
|
|
if (
|
|
set(value) != {"format", "version", "peer_id", "capabilities"}
|
|
or value["format"] != "trainlog-sync-peer"
|
|
or type(value["version"]) is not int
|
|
or value["version"] != 1
|
|
or value["peer_id"] != expected
|
|
or not isinstance(value["capabilities"], list)
|
|
or not CAPS.issubset(set(value["capabilities"]))
|
|
):
|
|
raise RuntimeError("incompatible or unpaired Android peer advertisement")
|
|
return value
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--database", required=True, type=Path)
|
|
parser.add_argument("--transport-root", required=True, type=Path)
|
|
parser.add_argument("--owned-root", required=True, type=Path)
|
|
parser.add_argument("--run-id", required=True)
|
|
parser.add_argument("--expected-peer", required=True)
|
|
parser.add_argument("--timeout", required=True, type=int)
|
|
parser.add_argument("--mode", choices=("directory", "mtp"), default="directory")
|
|
parser.add_argument("--mtp-adapter", type=Path)
|
|
args = parser.parse_args()
|
|
deadline = time.monotonic() + args.timeout
|
|
pump = None
|
|
if args.mode == "mtp":
|
|
if args.mtp_adapter is None:
|
|
raise RuntimeError("MTP adapter is required")
|
|
pump = lambda: run_adapter(
|
|
args.mtp_adapter,
|
|
"pull",
|
|
args.expected_peer,
|
|
args.transport_root,
|
|
deadline,
|
|
True,
|
|
)
|
|
refresh_mtp_peer(
|
|
args.mtp_adapter,
|
|
args.expected_peer,
|
|
args.transport_root,
|
|
deadline,
|
|
)
|
|
peer = load_peer(args.transport_root, args.expected_peer)
|
|
with closing(generation.connect_database(args.database)) as db:
|
|
generation.require_schema(db)
|
|
desktop_peer = generation.peer_identity(db, "desktop")
|
|
db.commit()
|
|
request = {
|
|
"format": "trainlog-sync-generation-request",
|
|
"version": 1,
|
|
"run_id": args.run_id,
|
|
"desktop_peer_id": desktop_peer,
|
|
"android_peer_id": peer["peer_id"],
|
|
"capabilities": sorted(CAPS),
|
|
}
|
|
request_path = args.transport_root / "request-v1.json"
|
|
publish_json(request_path, request)
|
|
with closing(generation.connect_database(args.database)) as db:
|
|
acknowledgement_rows = db.execute(
|
|
"SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? "
|
|
"AND consumer_peer_id=? AND result='consumed' ORDER BY consumed_at,generation_id LIMIT 32",
|
|
(peer["peer_id"], desktop_peer),
|
|
).fetchall()
|
|
archive_acknowledgements = {
|
|
"format": "trainlog-sync-archive-acknowledgements",
|
|
"version": 1,
|
|
"run_id": args.run_id,
|
|
"android_peer_id": peer["peer_id"],
|
|
"desktop_peer_id": desktop_peer,
|
|
"acknowledgements": [json.loads(row[0]) for row in acknowledgement_rows],
|
|
}
|
|
archive_ack_path = args.transport_root / "desktop-archive-acknowledgements-v1.json"
|
|
publish_json(archive_ack_path, archive_acknowledgements)
|
|
if args.mode == "mtp":
|
|
push_adapter(
|
|
args.mtp_adapter,
|
|
args.expected_peer,
|
|
args.transport_root,
|
|
["request-v1.json", "desktop-archive-acknowledgements-v1.json"],
|
|
deadline,
|
|
)
|
|
emit(
|
|
args.run_id,
|
|
"waiting_android_publication",
|
|
producer_peer_id=peer["peer_id"],
|
|
consumer_peer_id=desktop_peer,
|
|
)
|
|
inbound_ref = args.transport_root / "android-generation-v1.json"
|
|
def pump_inbound():
|
|
if pump is not None:
|
|
pump()
|
|
correlated_peer_error(args.transport_root, args.run_id)
|
|
|
|
inbound = wait_json(
|
|
inbound_ref, deadline, lambda value: value.get("run_id") == args.run_id, pump_inbound
|
|
)
|
|
inbound_dir = args.transport_root / inbound["relative_path"]
|
|
ack = generation.consume_desktop(args.database, inbound_dir)
|
|
ack_path = args.transport_root / "desktop-consumption-ack-v1.json"
|
|
publish_json(ack_path, ack)
|
|
if args.mode == "mtp":
|
|
push_adapter(
|
|
args.mtp_adapter,
|
|
args.expected_peer,
|
|
args.transport_root,
|
|
["request-v1.json", "desktop-archive-acknowledgements-v1.json", "desktop-consumption-ack-v1.json"],
|
|
deadline,
|
|
)
|
|
if ack["result"] != "consumed":
|
|
raise RuntimeError("desktop rejected Android generation")
|
|
emit(
|
|
args.run_id,
|
|
"local_import_committed",
|
|
inbound_generation_id=ack["generation_id"],
|
|
manifest_sha256=ack["manifest_sha256"],
|
|
)
|
|
outgoing_id = "gen_" + str(uuid.uuid4())
|
|
# Export codecs retain human diagnostics on stdout when invoked as CLIs.
|
|
# The worker stdout contract is typed NDJSON only, so route those bounded
|
|
# diagnostics to the separately bounded stderr channel.
|
|
with redirect_stdout(sys.stderr):
|
|
stage, manifest, digest = generation.capture_desktop(
|
|
args.database, args.owned_root, peer["peer_id"], args.run_id, outgoing_id
|
|
)
|
|
del stage
|
|
published = generation.publish(
|
|
args.database, outgoing_id, args.transport_root / "desktop-objects"
|
|
)
|
|
ref = {
|
|
"format": "trainlog-sync-generation-reference",
|
|
"version": 1,
|
|
"run_id": args.run_id,
|
|
"generation_id": outgoing_id,
|
|
"manifest_sha256": digest,
|
|
"relative_path": str(published.relative_to(args.transport_root)),
|
|
}
|
|
publish_json(args.transport_root / "desktop-generation-v1.json", ref)
|
|
if args.mode == "mtp":
|
|
# CONTRACT: the immutable generation must be fully visible before its
|
|
# mutable run-correlated reference. Publishing both in one MTP walk
|
|
# lets Android observe the reference while artifacts are still being
|
|
# transferred, turning a complete generation into a false rejection.
|
|
push_desktop_generation(
|
|
args.mtp_adapter,
|
|
args.expected_peer,
|
|
args.transport_root,
|
|
ref["relative_path"],
|
|
deadline,
|
|
)
|
|
emit(
|
|
args.run_id,
|
|
"published",
|
|
outbound_generation_id=outgoing_id,
|
|
manifest_sha256=digest,
|
|
)
|
|
emit(args.run_id, "waiting_acknowledgement", outbound_generation_id=outgoing_id)
|
|
android_ack = args.transport_root / "android-consumption-ack-v1.json"
|
|
android_ack_value = wait_json(
|
|
android_ack,
|
|
deadline,
|
|
lambda value: value.get("run_id") == args.run_id
|
|
and value.get("generation_id") == outgoing_id,
|
|
pump,
|
|
)
|
|
result = generation.accept_ack(args.database, android_ack)
|
|
if result not in ("acknowledged", "unchanged"):
|
|
raise RuntimeError(
|
|
f"Android did not durably consume desktop generation: {result}: {android_ack_value.get('diagnostic','')}"
|
|
)
|
|
with closing(generation.connect_database(args.database)) as db:
|
|
inbound_row = db.execute(
|
|
"SELECT result FROM sync_consumed_generations WHERE generation_id=?",
|
|
(ack["generation_id"],),
|
|
).fetchone()
|
|
outbound_row = db.execute(
|
|
"SELECT status FROM sync_generations WHERE generation_id=?", (outgoing_id,)
|
|
).fetchone()
|
|
drafts = [
|
|
{
|
|
"draft_id": row[0],
|
|
"state": row[1],
|
|
"session_type": json.loads(row[2]).get("session_type", "training"),
|
|
"occurrence_count": len(json.loads(row[2]).get("exercises", [])),
|
|
}
|
|
for row in db.execute(
|
|
"SELECT session_id,state,payload_json FROM execution_drafts ORDER BY session_id LIMIT 32"
|
|
)
|
|
]
|
|
if inbound_row != ("consumed",) or outbound_row != ("acknowledged",):
|
|
raise RuntimeError("generation evidence did not corroborate completion")
|
|
emit(args.run_id, "peer_consumed", outbound_generation_id=outgoing_id)
|
|
report = {
|
|
"format": "trainlog-sync-worker-report",
|
|
"version": 1,
|
|
"run_id": args.run_id,
|
|
"producer_peer_id": peer["peer_id"],
|
|
"consumer_peer_id": desktop_peer,
|
|
"inbound_generation_id": ack["generation_id"],
|
|
"outbound_generation_id": outgoing_id,
|
|
"manifest_sha256": digest,
|
|
"result": "completed",
|
|
"sessions_reconciled": 0,
|
|
"domains": {
|
|
name: True
|
|
for name in (
|
|
"history-v4",
|
|
"execution-draft-v1",
|
|
"causal-delete-v1",
|
|
"equipment",
|
|
"aliases",
|
|
"profiles",
|
|
"body-zones",
|
|
"feedback",
|
|
)
|
|
},
|
|
"drafts": drafts,
|
|
"ai_midpoint": {"result": "not_configured"},
|
|
"ai_post_sync": {"result": "not_configured"},
|
|
}
|
|
print(json.dumps(report, sort_keys=True, separators=(",", ":")), flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except RuntimeError as error:
|
|
print(f"sync peer failed: {error}", file=sys.stderr)
|
|
raise SystemExit(1) from None
|