tools/export_fixture.py

"""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()