"""Synthetic offline fixture generator, not a source of real platform records.""" import argparse import copy import hashlib import json from pathlib import Path from tools.business_export import encoded, export, reconstruct CANARY = "hf_SYNTHETIC_CANARY_SECRET_DO_NOT_EXPORT_123456" ORCID = "0000-0002-1825-0097" SUBJECT = "synthetic-stable-subject" TASKS = ("propertyEstimation", "task_future_prediction", "motionTransfer", "videoGeneration_static") TIMESTAMP = "2026-10-11T00:00:00Z" def write(root, relative, value): path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(encoded(value)) def fixture(root, counts=(3, 3, 3, 3)): root = Path(root) root.mkdir(parents=True, exist_ok=False) profile = dict(name="Synthetic Researcher", role="Independent Researcher", affiliation="Synthetic Institute", email="synthetic@example.org", supervisor_name="", hf_username="synthetic-user", hf_subject=SUBJECT, orcid_id=ORCID, researcher_identity_id=ORCID, submitter_id=ORCID, accepted_at=TIMESTAMP, rules_version="1", identity_created_at=TIMESTAMP, orcid_connected_at=TIMESTAMP, profile_updated_at=TIMESTAMP, original_profile=dict(name="Original Researcher", role="Independent Researcher", affiliation="", email="synthetic@example.org", supervisor_name="", saved_at=TIMESTAMP, access_token=CANARY), profile_revisions=[dict(timestamp=TIMESTAMP, actor_hf_username="synthetic-user", researcher_identity_id=ORCID, changes=dict(name=dict(old_value="Original Researcher", new_value="Synthetic Researcher", token=CANARY)))], rules_acceptances=[dict(researcher_identity_id=ORCID, rules_version="1", accepted_at=TIMESTAMP, actor_hf_username="synthetic-user")]) pending_id, running_id = "0000000005", "0000000006" registry = dict(submitters={"synthetic-user": profile}, reservations={pending_id: dict(orcid_id=ORCID, reserved_at=TIMESTAMP), running_id: dict(orcid_id=ORCID, reserved_at=TIMESTAMP)}, hf_accounts={SUBJECT: dict(username="synthetic-user")}, publication_sequences={ORCID: {TASKS[0]: 1}}, orcid_states={CANARY: dict(hf_username="synthetic-user", created_at=TIMESTAMP)}, orcid_verifications={"synthetic-user": dict(orcid_id=ORCID, code=CANARY)}) write(root, "submission_policy/registry.json", registry) migration = copy.deepcopy(registry) migration["reservations"] = {} migration_id = hashlib.sha256(encoded(migration)).hexdigest() write(root, f"submission_policy/identity_migration_snapshots/{migration_id}.json", migration) for index, (task, count) in enumerate(zip(TASKS, counts), 1): submission_id = f"{index:010x}" primary = "rendering_pmf" if index == 1 else "pmf_novel" if index == 2 else "pmf" metrics = {primary: 0.5} snapshot = dict(hf_username_snapshot="synthetic-user", orcid=ORCID, researcher_identity={key: profile[key] for key in ("name", "role", "affiliation", "email", "supervisor_name")}, display_name=f"Synthetic {task}", task_name=task, track_id=task, user_dataset="synthetic-user/predictions", repo_subdir="run", submitted_at=TIMESTAMP, private_token_provided=True, private_token_consent=dict(policy_version="private-repository-token-v1", accepted_at=TIMESTAMP), paper_url="https://example.org/paper?version=2&token=" + CANARY, code_url="https://github.com/synthetic/repo?branch=main&api_key=" + CANARY, primary_metric=primary, rules_version="1", session=dict(secret=CANARY)) provenance = dict(schema_version=1, input=dict(repository_id="synthetic-user/predictions", repository_type="dataset", subdirectory="run", revision="d" * 40, resolved_at=TIMESTAMP), evaluator=dict(git_commit="a" * 40, git_commit_source="build_environment", source_digest="b" * 64, protocol_version="scene-evaluation-v1", manifest_version=1, manifest_digest="c" * 64, metric_configuration_version=1, metric_configuration_digest="e" * 64, metric_configuration=dict(requested_device="cuda", effective_device="cuda", VIDEO_GENERATION_FRAME_BATCH_SIZE=12, packages=dict(torch="synthetic-version"), api_key=CANARY))) record = dict(submission_id=submission_id, researcher_identity_id=ORCID, hf_username_snapshot="synthetic-user", display_name=f"Synthetic {task}", task_name=task, user_dataset="synthetic-user/predictions", repo_subdir="run", submitted_at=TIMESTAMP, started_at=TIMESTAMP, completed_at=TIMESTAMP, status="done", scenes_total=count, scenes_done=count, scenes_failed=0, score=0.5, metrics=metrics, primary_metric=primary, quota_consumed_at=TIMESTAMP, evaluation_revision="d" * 40, evaluation_provenance=provenance, creation_snapshot=snapshot, worker_id="synthetic-worker", worker_attempt=1, evaluator_config_version=1, submission_events=[dict(event_id=submission_id + ":completed", event="completed", timestamp=TIMESTAMP, actor_worker_id="synthetic-worker")]) progress = dict(submission_id=submission_id, scenes_total=count, scene_results={}, scene_failures={}, scene_retries={}, infrastructure_failures={}, case_started_at={}, last_worker="synthetic-worker", last_updated_at=TIMESTAMP, submission_events=[]) for case_index in range(count): case = f"SyntheticScene__bg100__{case_index:06d}_trajectory" progress["scene_results"][case] = dict(metrics=metrics, ts=TIMESTAMP) progress["case_started_at"][case] = TIMESTAMP for event in ("case_started", "case_completed"): progress["submission_events"].append(dict(event_id=f"{submission_id}:{case}:{event}", event=event, case=case, timestamp=TIMESTAMP, submission_id=submission_id, attempt=2 if event == "case_completed" and case_index == 0 else 1, actor_worker_id="synthetic-worker")) if count: progress["scene_retries"]["SyntheticScene__bg100__000000_trajectory"] = dict(attempts=1, status="recovered", infrastructure=True, error=f"RuntimeError: Bearer {CANARY}; password=PLAIN_SYNTHETIC_CANARY; https://example.org/file?signature=URL_SYNTHETIC_CANARY&part=2") write(root, f"queue/complete/{submission_id}.json", record) write(root, f"queue/progress/{submission_id}.json", progress) history = dict(record, submission_events=record["submission_events"] + progress["submission_events"], scene_failures={}) history_path = root / "results" / task / "_history.2026-10.jsonl" history_path.parent.mkdir(parents=True, exist_ok=True) history_path.write_text(json.dumps(history) + "\n", encoding="utf-8") write(root, f"presentation_metadata/{task}/{submission_id}.json", dict(researcher_identity_id=ORCID, display_name=record["display_name"], paper_url=snapshot["paper_url"], code_url=snapshot["code_url"], metadata_updated_at=TIMESTAMP, revisions=[dict(timestamp=TIMESTAMP, field="code_url", previous_value=None, new_value=snapshot["code_url"], event_id="synthetic-edit", actor_hf_username="synthetic-user", submission_id=submission_id)])) for submission_id, status in ((pending_id, "queued"), (running_id, "running")): record = dict(submission_id=submission_id, researcher_identity_id=ORCID, hf_username_snapshot="synthetic-user", task_name=TASKS[0], display_name="Synthetic active", status=status, submitted_at=TIMESTAMP, user_dataset="synthetic-user/predictions", access_token_encrypted=CANARY, lease_expires_at=TIMESTAMP, evaluator_dispatch_ready=False) write(root, f"queue/{'pending' if status == 'queued' else 'processing'}/{submission_id}.json", record) cancelled_id = "0000000007" cancelled = dict(submission_id=cancelled_id, researcher_identity_id=ORCID, hf_username_snapshot="synthetic-user", task_name=TASKS[0], status="cancelled", score=None, metrics={}, scenes_done=0, scenes_failed=0, completed_at=TIMESTAMP, quota_consumed_at=TIMESTAMP, quota_refunded_at=TIMESTAMP, quota_refund_reason="api_key=PLAIN_SYNTHETIC_CANARY") write(root, f"archive/complete-2026-09/{cancelled_id}.json", cancelled) with (root / f"results/{TASKS[0]}/_history.2026-10.jsonl").open("a", encoding="utf-8") as handle: handle.write(json.dumps(cancelled) + "\n") legacy = dict(submission_id="0000000008", team_id="synthetic-user", display_name="Legacy synthetic", task_name=TASKS[2], primary_metric="pmf", score=0.75, metrics=dict(pmf=0.75), scenes_count=3, updated_at=TIMESTAMP) write(root, f"results/{TASKS[2]}/synthetic-user_best.json", legacy) tombstone = dict(submission_id="0000000002", task_name=TASKS[1], researcher_identity_id=ORCID, deleted_at=TIMESTAMP, request_id="f" * 16, event_id="f" * 16, event="soft_deleted", actor_hf_username="synthetic-user", reason="deleted by submitter") (root / f"results/{TASKS[1]}/_deleted_submissions.jsonl").write_text(json.dumps(tombstone) + "\n", encoding="utf-8") write(root, "queue/deletions/complete/" + "f" * 16 + ".json", dict(request_id="f" * 16, task_name=TASKS[1], submission_id="0000000002", researcher_identity_id=ORCID, requested_by="synthetic-user", status="done", requested_at=TIMESTAMP, completed_at=TIMESTAMP, message="submission deleted")) entry = dict(researcher_identity_id=ORCID, track_id=TASKS[0], task_name=TASKS[0], submission_id="0000000001", published_by="synthetic-user", published_at=TIMESTAMP) request_id = "a" * 24 write(root, "submission_policy/publications.json", dict(entries=[entry], versions={ORCID: {TASKS[0]: dict(sequence=1, request_id=request_id, action="publish", submission_id="0000000001", replace_submission_id=None)}}, history=[dict(event_id=request_id, request_id=request_id, actor_hf_username="synthetic-user", timestamp=TIMESTAMP, action="publish", requested_action="publish", researcher_identity_id=ORCID, track_id=TASKS[0], previous_publication=None, new_publication=entry)])) write(root, f"submission_policy/publication_requests/complete/{request_id}.json", dict(request_id=request_id, action="publish", task_name=TASKS[0], track_id=TASKS[0], submission_id="0000000001", requested_by="synthetic-user", hf_subject=SUBJECT, researcher_identity_id=ORCID, publication_sequence=1, requested_at=TIMESTAMP, completed_at=TIMESTAMP, status="done", message="Publication updated")) audit_path = root / "logs/exec-2026-10.jsonl" audit_path.parent.mkdir(parents=True, exist_ok=True) audit_path.write_text(json.dumps(dict(submission_id="0000000001", timestamp=TIMESTAMP, action="history_reconciled", actor="synthetic-worker", previous_state="done", current_state="done", reason="token=PLAIN_SYNTHETIC_CANARY", result="repaired")) + "\n", encoding="utf-8") (root / "logs/motionTransfer.child.log").write_text(CANARY, encoding="utf-8") return root def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", required=True) arguments = parser.parse_args() output = Path(arguments.output) if output.exists(): parser.error("output must not exist") output.mkdir(parents=True) fixture(output / "synthetic-source") manifest = export(output / "synthetic-source", output / "export", source_id="synthetic-example-v1") summary = reconstruct(output / "export", output / "isolated") print(json.dumps(dict(files=manifest["file_count"], validation=manifest["validation"], reconstruction=summary), indent=2)) if __name__ == "__main__": main()