Workflow helpers — skills.*

These two modules are the helper layer the sv-* workflow uses to scaffold, version, branch and ground a study. They are not part of the simulation engine — they operate on the study directory — but they are worth knowing if you drive the pipeline yourself instead of through the skills, or if you script batch experiments over many studies.

Both are stdlib-only and safe to import without a configured LLM.

skills.sv_workspace — study directories, versions and branches

Scaffolding (scaffold, study_paths), the catalog that sv-init routes against (discover_studies, catalog_view), study-level forking (fork_study — copies an existing study under a new study_id, rewrites that id throughout the copied artifacts, forces reference: false, and deliberately carries no code and no run outputs), and the capability registry view (load_capabilities, capability_view).

The version machinery behind /sv-iterate, in the order a run touches it:

function	what it does
init_manifest(study_dir, note)	registers the live directory as v1 with kind: "version" and parent: null. No-op once a manifest exists
create_version(study_dir, note, base=None, warm_start=None, resume_from_stage=None)	freezes the current version into versions/<id>/, optionally restores an older base into the live dir, drops the live run outputs, then appends the new entry and moves current. Records kind — "branch" when a warm_start is passed, "version" otherwise — and fork_step
create_branch(study_dir, note, *, fork_step, source=None, resume_from_stage="sv-build-environment")	the proper entry point for a branch. source defaults to the current version; naming an older one restores its artifacts first. fork_step < 1 raises ValueError (step 0 is the materialized initial state — there is nothing to inherit). Internally records warm_start={"source": src, "resume_from": fork_step - 1}
check_branch_invariant(study_dir, *, version_id=None, downgrade=True)	verifies the E-only claim against the parent's snapshot and returns {kind, ok, violations, downgraded, source, fork_step, error}. On a violation it downgrades the entry (kind → "version", warm_start → None, fork_step → None, reason into downgraded_from_branch) instead of raising, so the run continues — cold, and still correct. downgrade=False checks without writing
resolve_warm_start(study_dir)	turns the manifest's replay provenance into runnable engine params {source_version, source_trajectory, resume_from}. It calls check_branch_invariant first, so a downgraded branch resolves to None and sv-run starts cold from step 0
version_entry(study_dir, version_id=None), version_kind(entry), version_fork_step(entry)	read one manifest entry and classify it. The latter two fall back to warm_start provenance, so manifests written before kind and fork_step existed classify correctly without a migration

skills.sv_workspace

Shared helper for the SocioVerse2 workflow skills.

Scaffolds a standardized study workspace and provides the strict-contract load/save helpers (thin wrappers over socioverse.validation) the skills must use at every boundary.

scaffold

scaffold(studies_root: str | Path, study_id: str) -> Path

Create the standardized study workspace and return its path.

Source code in skills/sv_workspace.py

def scaffold(studies_root: str | Path, study_id: str) -> Path:
    """Create the standardized study workspace and return its path."""
    root = Path(studies_root) / study_id
    for sub in STUDY_SUBDIRS:
        (root / sub).mkdir(parents=True, exist_ok=True)
    return root

write_roster

write_roster(path: str | Path, rows) -> Path

Write the materialized initial roster (population/roster.jsonl): one JSON line per agent, the t=0 panel row (agent_id, step, state, action_kind, action_payload). This is the durable 'instantiated agents' artifact sv-build-population produces so the dashboard's agent inspector can show every initialized agent BEFORE any run — the same shape the live panel_live.jsonl uses, so the dashboard reads it with no special casing. rows = TrajectoryRecords (or dicts with those keys). Lives under population/, so version snapshots keep it.

Source code in skills/sv_workspace.py

def write_roster(path: str | Path, rows) -> Path:
    """Write the materialized initial roster (population/roster.jsonl): one JSON line per agent,
    the t=0 panel row (agent_id, step, state, action_kind, action_payload). This is the durable
    'instantiated agents' artifact sv-build-population produces so the dashboard's agent inspector
    can show every initialized agent BEFORE any run — the same shape the live panel_live.jsonl
    uses, so the dashboard reads it with no special casing. ``rows`` = TrajectoryRecords (or dicts
    with those keys). Lives under population/, so version snapshots keep it."""
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)

    def _get(r, k, default=None):
        return getattr(r, k, default) if not isinstance(r, dict) else r.get(k, default)

    with p.open("w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps({
                "agent_id": _get(r, "agent_id"),
                "step": _get(r, "step", 0),
                "state": _get(r, "state", {}),
                "action_kind": _get(r, "action_kind"),
                "action_payload": _get(r, "action_payload", {}),
            }, ensure_ascii=False) + "\n")
    return p

save_study_yaml

save_study_yaml(path: str | Path, study_spec) -> Path

StudySpec is written as JSON-in-.yaml (valid YAML, simple + re-loadable).

Source code in skills/sv_workspace.py

def save_study_yaml(path: str | Path, study_spec) -> Path:
    """StudySpec is written as JSON-in-.yaml (valid YAML, simple + re-loadable)."""
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(study_spec.model_dump_json(indent=2), encoding="utf-8")
    return p

fork_study

fork_study(
    studies_root: str | Path,
    source_study_id: str,
    new_study_id: str,
    *,
    status: str = "draft",
) -> Path

Path-A reuse: copy an existing study into a NEW study_id so the source stays intact.

SocioVerse2's reuse rule is never edit a matched or reference study in place — fork it. Any study whose study.yaml carries reference: true (the bundled templates), and any already-adapted study, is a read-only baseline; a new query becomes its own study. This copies the source's authored artifacts (study.yaml + the resources/environment/population/simulation JSON that exist, plus grounding/grounding.json — the fork inherits the source's real-world anchors and refreshes only the entries it changes), rewrites the fork's identity (study_id across study.yaml AND every copied artifact that embeds it — resources/environment/population/simulation — plus status / created_by), forces reference: false (a working fork is never a read-only template), and DROPS stale run outputs (trajectory/*, reports/*) so the fork re-runs clean. No code is copied or written: the fork's bundles keep pointing at the source's registered provider_refs (that is what makes this the "no new code" path). Returns the new study root. Refuses to overwrite an existing study_id.

Source code in skills/sv_workspace.py

def fork_study(
    studies_root: str | Path,
    source_study_id: str,
    new_study_id: str,
    *,
    status: str = "draft",
) -> Path:
    """Path-A reuse: copy an existing study into a NEW ``study_id`` so the source stays intact.

    SocioVerse2's reuse rule is **never edit a matched or reference study in place — fork it.** Any
    study whose ``study.yaml`` carries ``reference: true`` (the bundled templates), and any
    already-adapted study, is a read-only baseline; a new query becomes its own study. This copies the source's
    authored artifacts (``study.yaml`` + the resources/environment/population/simulation JSON
    that exist, plus ``grounding/grounding.json`` — the fork inherits the source's real-world
    anchors and refreshes only the entries it changes), rewrites the fork's identity
    (``study_id`` across study.yaml AND every copied artifact that embeds it —
    resources/environment/population/simulation — plus ``status`` / ``created_by``), forces
    ``reference: false`` (a working fork is never a read-only template), and
    DROPS stale run outputs (``trajectory/*``, ``reports/*``) so the fork re-runs clean. No code
    is copied or written: the fork's bundles keep pointing at the source's registered
    ``provider_refs`` (that is what makes this the "no new code" path). Returns the new study
    root. Refuses to overwrite an existing ``study_id``.
    """
    src = Path(studies_root) / source_study_id
    if not (src / "study.yaml").exists():
        raise FileNotFoundError(f"no study to fork at {src}")
    dst = Path(studies_root) / new_study_id
    if dst.exists():
        raise FileExistsError(f"refusing to overwrite existing study {dst}")

    scaffold(studies_root, new_study_id)
    src_paths, dst_paths = study_paths(src), study_paths(dst)
    for key in ("study", "resources", "environment", "population", "simulation", "grounding"):
        s = src_paths[key]
        if s.exists():
            dst_paths[key].parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(s, dst_paths[key])

    spec = load_study_yaml(dst_paths["study"])
    spec.study_id = new_study_id
    spec.status = status
    spec.created_by = f"sv-init (fork of {source_study_id})"
    spec.reference = False   # a working fork is NEVER a read-only template — else it masquerades as
                             # one (blocks in-place edits, shows a REF badge, can be picked as a build template)
    save_study_yaml(dst_paths["study"], spec)

    # Rewrite the copied artifacts' embedded study_id too. resources/environment/population/simulation
    # each carry `study_id`; a stale SOURCE id mis-tags the fork's trajectory store (DuckDB `study_id`),
    # binding the fork's results onto the source study in the dashboard. fork_study historically only
    # rewrote study.yaml, so every caller had to hand-patch these (the recurring fork id-drift bug).
    for key in ("resources", "environment", "population", "simulation"):
        p = dst_paths[key]
        if not p.exists():
            continue
        try:
            obj = json.loads(p.read_text())
        except (json.JSONDecodeError, OSError):
            continue
        if isinstance(obj, dict) and obj.get("study_id") not in (None, new_study_id):
            obj["study_id"] = new_study_id
            p.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
    return dst

current_version

current_version(study_dir: str | Path) -> str

The version id the live dir currently carries (implicit v1 pre-manifest).

Source code in skills/sv_workspace.py

def current_version(study_dir: str | Path) -> str:
    """The version id the live dir currently carries (implicit ``v1`` pre-manifest)."""
    man = load_manifest(study_dir)
    return (man or {}).get("current") or "v1"

init_manifest

init_manifest(
    study_dir: str | Path, note: str = "initial build"
) -> dict

Register the live dir as v1. sv-init calls this at scaffold/fork time so the dashboard shows a default version from day one; pre-versioning studies get it lazily on their first /sv-iterate. No-op when a manifest already exists.

Source code in skills/sv_workspace.py

def init_manifest(study_dir: str | Path, note: str = "initial build") -> dict:
    """Register the live dir as ``v1``. sv-init calls this at scaffold/fork time so the
    dashboard shows a default version from day one; pre-versioning studies get it
    lazily on their first /sv-iterate. No-op when a manifest already exists."""
    existing = load_manifest(study_dir)
    if existing:
        return existing
    manifest = {
        "current": "v1",
        "versions": [{"id": "v1", "parent": None, "note": note, "kind": "version",
                      "created_at": time.time(), "snapshot": None, "warm_start": None,
                      "fork_step": None}],
    }
    _write_manifest(study_dir, manifest)
    return manifest

version_trajectory_db

version_trajectory_db(
    study_dir: str | Path, version_id: str
) -> Path

Path to an archived version's trajectory store — what warm-start replays from. create_version snapshots the live dir into versions/<id>/ and study.duckdb is NOT in SNAPSHOT_EXCLUDE, so a finished version's panel is preserved here for the next version to inherit. (Only defined for versions that have a snapshot; the live/current version's store is at trajectory/study.duckdb.)

Source code in skills/sv_workspace.py

def version_trajectory_db(study_dir: str | Path, version_id: str) -> Path:
    """Path to an archived version's trajectory store — what warm-start replays from.
    ``create_version`` snapshots the live dir into ``versions/<id>/`` and ``study.duckdb`` is NOT
    in SNAPSHOT_EXCLUDE, so a finished version's panel is preserved here for the next version to
    inherit. (Only defined for versions that have a snapshot; the live/current version's store is
    at ``trajectory/study.duckdb``.)"""
    return Path(study_dir) / VERSIONS_DIR / version_id / "trajectory" / "study.duckdb"

create_version

create_version(
    study_dir: str | Path,
    note: str,
    base: str | None = None,
    warm_start: dict | None = None,
    resume_from_stage: str | None = None,
) -> str

Iterate the study to a NEW version, preserving the current one as a snapshot.

This is the version-iterate path of /sv-iterate (the in-place path simply skips this call). Order is crash-safe: (1) freeze live → versions/v<current>/; (2) if base names an older version, restore its snapshot into the live dir (copy2 keeps mtimes, so untouched artifacts read as carried-from-parent); (3) drop the live run outputs (kept in the snapshot) so the new version starts at "built, not yet run"; (4) append the new entry + move current (atomic, last — a half-done copy is invisible to the dashboard until the manifest says otherwise).

Every entry records a kind: a branch (warm_start given — inherits the parent's realized history by replay and changes only the environment from its fork step on) or a version (anything else — any artifact may change, the run starts cold). See create_branch / check_branch_invariant for the branch contract.

note is the one-line human description shown in the dashboard's version box. base defaults to the current version; passing an older id starts the new version from that archived one instead (cross-version iteration). warm_start (optional) records the inherit-steps provenance on the NEW version entry — {"source": <parent v>, "resume_from": K} — set by /sv-iterate when the user opts to inherit the parent's already-run steps (the parent's snapshot is what sv-run then replays). resume_from_stage (optional) records the earliest sv-* stage this iteration re-enters (e.g. "sv-build-environment"); the dashboard reads it so the new version's progress pointer parks on that stage — rather than reading the parent's carried artifacts as this version's finished work — until Claude actually re-authors it. Returns the new version id. The caller (skill contract) must not invoke this while a run is writing the trajectory store.

Source code in skills/sv_workspace.py

def create_version(study_dir: str | Path, note: str, base: str | None = None,
                   warm_start: dict | None = None, resume_from_stage: str | None = None) -> str:
    """Iterate the study to a NEW version, preserving the current one as a snapshot.

    This is the **version-iterate** path of /sv-iterate (the in-place path simply skips
    this call). Order is crash-safe: (1) freeze live → ``versions/v<current>/``;
    (2) if ``base`` names an older version, restore its snapshot into the live dir
    (``copy2`` keeps mtimes, so untouched artifacts read as carried-from-parent);
    (3) drop the live run outputs (kept in the snapshot) so the new version starts at
    "built, not yet run"; (4) append the new entry + move ``current`` (atomic, last —
    a half-done copy is invisible to the dashboard until the manifest says otherwise).

    Every entry records a ``kind``: a **branch** (``warm_start`` given — inherits the parent's
    realized history by replay and changes only the environment from its fork step on) or a
    **version** (anything else — any artifact may change, the run starts cold). See
    ``create_branch`` / ``check_branch_invariant`` for the branch contract.

    ``note`` is the one-line human description shown in the dashboard's version box.
    ``base`` defaults to the current version; passing an older id starts the new version from
    that archived one instead (cross-version iteration). ``warm_start`` (optional) records the inherit-steps provenance
    on the NEW version entry — ``{"source": <parent v>, "resume_from": K}`` — set by /sv-iterate
    when the user opts to inherit the parent's already-run steps (the parent's snapshot is what
    `sv-run` then replays). ``resume_from_stage`` (optional) records the earliest sv-* stage this
    iteration re-enters (e.g. ``"sv-build-environment"``); the dashboard reads it so the new
    version's progress pointer parks on that stage — rather than reading the parent's carried
    artifacts as this version's finished work — until Claude actually re-authors it. Returns the
    new version id. The caller (skill contract) must not invoke this while a run is writing the
    trajectory store.
    """
    d = Path(study_dir)
    if not (d / "study.yaml").exists():
        raise FileNotFoundError(f"no study at {d}")
    manifest = load_manifest(d) or init_manifest(d)
    cur = manifest["current"]
    entries = {v["id"]: v for v in manifest["versions"]}
    base = base or cur
    if base not in entries:
        raise ValueError(f"unknown base version {base!r}; have {sorted(entries)}")
    # Starting a new VERSION from an older one (base != cur) is a big context switch → default the
    # re-entry to sv-init so the dashboard's review gate (see compute_stages) makes Claude re-review
    # EVERY stage from the start. (A BRANCH is different: create_branch passes sv-build-environment,
    # because E is the only artifact it re-authors.) A same-line new version leaves resume_from_stage
    # to the caller (the earliest affected stage); if that too is omitted, compute_stages fails safe
    # to the first carried stage, never run.
    if resume_from_stage is None and base != cur:
        resume_from_stage = "sv-init"

    # 1) freeze the live dir as the current version's archive
    entries[cur]["snapshot"] = _snapshot_live(d, cur)

    # 2) branching from an older version: make live == that snapshot (minus outputs)
    if base != cur:
        base_snap = d / (entries[base].get("snapshot") or "")
        if not base_snap.is_dir():
            raise FileNotFoundError(f"base version {base!r} has no snapshot at {base_snap}")
        for child in d.iterdir():
            if child.name in RESTORE_KEEP:
                continue
            shutil.rmtree(child) if child.is_dir() else child.unlink()
        for child in base_snap.iterdir():
            if child.name in OUTPUT_SUBDIRS:
                continue  # outputs stay archived; the new version re-runs
            dst = d / child.name
            if child.is_dir():
                shutil.copytree(child, dst)
            else:
                shutil.copy2(child, dst)

    # 3) clear live run outputs (preserved in the snapshot) + restore the empty layout
    for sub in OUTPUT_SUBDIRS:
        if (d / sub).exists():
            shutil.rmtree(d / sub)
    for sub in STUDY_SUBDIRS:
        (d / sub).mkdir(parents=True, exist_ok=True)

    # 4) register the new version and point `current` at it (atomic, last)
    nums = [int(v["id"][1:]) for v in manifest["versions"]
            if v["id"].startswith("v") and v["id"][1:].isdigit()]
    new_id = f"v{(max(nums) if nums else len(manifest['versions'])) + 1}"
    manifest["versions"].append({"id": new_id, "parent": base, "note": note,
                                 # branch = inherits the parent's realized history by replay and
                                 # changes only E from the fork step on; version = anything else.
                                 "kind": "branch" if warm_start else "version",
                                 "created_at": time.time(), "snapshot": None,
                                 "warm_start": warm_start,   # {source, resume_from} | None
                                 "fork_step": _fork_step_of(warm_start),   # first self-computed step | None
                                 "resume_from_stage": resume_from_stage})   # earliest re-entered stage | None
    manifest["current"] = new_id
    _write_manifest(d, manifest)
    return new_id

version_entry

version_entry(
    study_dir: str | Path, version_id: str | None = None
) -> dict | None

The manifest entry for version_id (default: the current version).

Source code in skills/sv_workspace.py

def version_entry(study_dir: str | Path, version_id: str | None = None) -> dict | None:
    """The manifest entry for ``version_id`` (default: the current version)."""
    man = load_manifest(study_dir)
    if not man:
        return None
    vid = version_id or man.get("current") or "v1"
    return next((v for v in man.get("versions", []) if v.get("id") == vid), None)

version_kind

version_kind(entry: dict | None) -> str

"branch" or "version". Manifests written before kind existed are read by their warm-start provenance, so old studies classify correctly without a migration.

Source code in skills/sv_workspace.py

def version_kind(entry: dict | None) -> str:
    """``"branch"`` or ``"version"``. Manifests written before ``kind`` existed are read by their
    warm-start provenance, so old studies classify correctly without a migration."""
    if not entry:
        return "version"
    kind = entry.get("kind")
    if kind in ("branch", "version"):
        return kind
    return "branch" if (entry.get("warm_start") or {}).get("source") else "version"

version_fork_step

version_fork_step(entry: dict | None) -> int | None

A branch's fork step, derived from warm-start provenance on pre-fork_step manifests.

Source code in skills/sv_workspace.py

def version_fork_step(entry: dict | None) -> int | None:
    """A branch's fork step, derived from warm-start provenance on pre-``fork_step`` manifests."""
    if not entry:
        return None
    fs = entry.get("fork_step")
    if isinstance(fs, int):
        return fs
    return _fork_step_of(entry.get("warm_start"))

create_branch

create_branch(
    study_dir: str | Path,
    note: str,
    *,
    fork_step: int,
    source: str | None = None,
    resume_from_stage: str = "sv-build-environment",
) -> str

Branch the study at fork_step — the counterfactual path of /sv-iterate.

The new version inherits source's realized history for steps 0..fork_step-1 (sv-run replays that version's stored actions, no LLM, so the inherited steps reproduce the parent exactly) and computes fork_step onward for itself. Only the environment may differ, and only from fork_step on: the intervention is a ScheduledEvent or Broadcast authored at a step >= fork_step. check_branch_invariant enforces that before the run.

source defaults to the current version. Naming an older one restores that version's artifacts into the live dir first (as create_version(base=...) does), so the branch really departs from that version's design. Re-entry defaults to sv-build-environment because the environment bundle is the only artifact a branch re-authors. Returns the new version id.

Source code in skills/sv_workspace.py

def create_branch(study_dir: str | Path, note: str, *, fork_step: int,
                  source: str | None = None,
                  resume_from_stage: str = "sv-build-environment") -> str:
    """Branch the study at ``fork_step`` — the counterfactual path of /sv-iterate.

    The new version inherits ``source``'s realized history for steps ``0..fork_step-1`` (sv-run
    replays that version's stored actions, no LLM, so the inherited steps reproduce the parent
    exactly) and computes ``fork_step`` onward for itself. Only the environment may differ, and
    only from ``fork_step`` on: the intervention is a ScheduledEvent or Broadcast authored at a
    step >= ``fork_step``. ``check_branch_invariant`` enforces that before the run.

    ``source`` defaults to the current version. Naming an older one restores that version's
    artifacts into the live dir first (as ``create_version(base=...)`` does), so the branch really
    departs from that version's design. Re-entry defaults to ``sv-build-environment`` because the
    environment bundle is the only artifact a branch re-authors. Returns the new version id.
    """
    d = Path(study_dir)
    fork_step = int(fork_step)
    if fork_step < 1:
        raise ValueError(f"fork_step must be >= 1 (step 0 is the materialized initial state, "
                         f"nothing to inherit); got {fork_step}")
    src = source or current_version(d)
    return create_version(d, note, base=None if src == current_version(d) else src,
                          warm_start={"source": src, "resume_from": fork_step - 1},
                          resume_from_stage=resume_from_stage)

check_branch_invariant

check_branch_invariant(
    study_dir: str | Path,
    *,
    version_id: str | None = None,
    downgrade: bool = True,
) -> dict

Verify that the current version really is a branch, and downgrade it to a version if not.

A branch claims that everything differing from its parent after the fork step is caused by the declared intervention. That claim only holds while the population, the study's code, the run settings and the pre-fork schedule are the parent's. This compares the live workspace against the parent's frozen snapshot and reports every artifact that breaks the claim.

On a violation the version is downgraded, not rejected: kind becomes "version", the warm-start provenance is dropped (so sv-run does a cold run from step 0 and the results stay correct, just more expensive) and the reason is recorded on the entry as downgraded_from_branch for the audit trail. The caller is expected to tell the user.

Returns {kind, ok, violations, downgraded, source, fork_step, error}. A version (or a study with no manifest) is trivially ok — there is nothing to inherit.

Source code in skills/sv_workspace.py

def check_branch_invariant(study_dir: str | Path, *, version_id: str | None = None,
                           downgrade: bool = True) -> dict:
    """Verify that the current version really is a branch, and downgrade it to a version if not.

    A branch claims that everything differing from its parent after the fork step is caused by the
    declared intervention. That claim only holds while the population, the study's code, the run
    settings and the pre-fork schedule are the parent's. This compares the live workspace against
    the parent's frozen snapshot and reports every artifact that breaks the claim.

    On a violation the version is **downgraded, not rejected**: ``kind`` becomes ``"version"``,
    the warm-start provenance is dropped (so sv-run does a cold run from step 0 and the results
    stay correct, just more expensive) and the reason is recorded on the entry as
    ``downgraded_from_branch`` for the audit trail. The caller is expected to tell the user.

    Returns ``{kind, ok, violations, downgraded, source, fork_step, error}``. A version (or a
    study with no manifest) is trivially ``ok`` — there is nothing to inherit.
    """
    d = Path(study_dir)
    report = {"kind": "version", "ok": True, "violations": [], "downgraded": False,
              "source": None, "fork_step": None, "error": None}
    man = load_manifest(d)
    if not man:
        return report
    vid = version_id or man.get("current") or "v1"
    entry = next((v for v in man.get("versions", []) if v.get("id") == vid), None)
    if version_kind(entry) != "branch":
        return report

    ws = (entry or {}).get("warm_start") or {}
    source, fork_step = ws.get("source"), version_fork_step(entry)
    report.update(kind="branch", source=source, fork_step=fork_step)
    violations: list[str] = []
    try:
        snap = d / VERSIONS_DIR / str(source)
        if not snap.is_dir():
            violations.append(f"the snapshot of parent version {source} does not exist, so its history cannot be inherited")
        else:
            for rel, label in BRANCH_INVARIANT_FILES:
                live_p, snap_p = d / rel, snap / rel
                if live_p.suffix == ".json":
                    live_v = _strip_volatile(_load_json(live_p)) if live_p.is_file() else None
                    snap_v = _strip_volatile(_load_json(snap_p)) if snap_p.is_file() else None
                    same = _stable(live_v) == _stable(snap_v)
                else:
                    same = _digest(live_p) == _digest(snap_p)
                if not same:
                    violations.append(f"{label} ({rel}) differs from the parent version — a branch must not change the population P")
            if _code_digest(d) != _code_digest(snap):
                violations.append("model code (model.py / adapter/*.py) differs from the parent version "
                                  "— a branch must not change the behavior function f")
            live_sim, snap_sim = _load_json(d / "simulation" / "simulation.json"), \
                _load_json(snap / "simulation" / "simulation.json")
            if live_sim is not None and snap_sim is not None:
                changed = [f for f in BRANCH_INVARIANT_SIM_FIELDS
                           if _stable(live_sim.get(f)) != _stable(snap_sim.get(f))]
                if changed:
                    violations.append("run settings " + ", ".join(changed) +
                                      " differ from the parent version — a branch may only lengthen n_steps")
            live_env, snap_env = _load_json(d / "environment" / "environment.json"), \
                _load_json(snap / "environment" / "environment.json")
            if fork_step is not None and live_env is not None and snap_env is not None:
                if _stable(_env_history(live_env, fork_step)) != \
                        _stable(_env_history(snap_env, fork_step)):
                    violations.append(f"the environment before step {fork_step} (provider / scheduled events / broadcasts) "
                                      "differs from the parent version — a branch may only add interventions at or after the fork step")
                live_layers = {l.get("name"): l for l in (live_env.get("layers") or [])}
                for layer in snap_env.get("layers") or []:
                    name = layer.get("name")
                    if name not in live_layers:
                        violations.append(f"the parent version's environment layer {name!r} was deleted in the branch")
                    elif _stable(live_layers[name]) != _stable(layer):
                        violations.append(f"environment layer {name!r} was rewritten — a branch may only add layers, never change existing ones")
    except Exception as exc:                     # verification must never break a run
        report["error"] = f"{type(exc).__name__}: {exc}"
        return report

    if not violations:
        return report
    report.update(ok=False, violations=violations)
    if downgrade and entry is not None:
        entry["kind"] = "version"
        entry["warm_start"] = None
        entry["fork_step"] = None
        entry["downgraded_from_branch"] = {"source": source, "fork_step": fork_step,
                                           "violations": violations, "at": time.time()}
        _write_manifest(d, man)
        report["downgraded"] = True
        report["kind"] = "version"
    return report

resolve_warm_start

resolve_warm_start(study_dir: str | Path) -> dict | None

Turn the current version's manifest warm_start provenance into runnable engine params.

/sv-iterate records {"source": <parent v>, "resume_from": K} on the new version entry; sv-run calls this to build a WarmStartSpec — it resolves source to the parent's archived store (versions/<source>/trajectory/study.duckdb) and returns {source_version, source_trajectory, resume_from}, or None when there is no warm-start or the parent snapshot is missing (→ fall back to a cold run from step 0). Keeps the manifest the single source of truth (sv-run doesn't re-derive the resume point).

Source code in skills/sv_workspace.py

def resolve_warm_start(study_dir: str | Path) -> dict | None:
    """Turn the current version's manifest ``warm_start`` provenance into runnable engine params.

    /sv-iterate records ``{"source": <parent v>, "resume_from": K}`` on the new version entry;
    sv-run calls this to build a ``WarmStartSpec`` — it resolves ``source`` to the parent's
    archived store (``versions/<source>/trajectory/study.duckdb``) and returns
    ``{source_version, source_trajectory, resume_from}``, or ``None`` when there is no warm-start
    or the parent snapshot is missing (→ fall back to a cold run from step 0). Keeps the manifest
    the single source of truth (sv-run doesn't re-derive the resume point)."""
    # The branch contract is verified here, not only in the skill: every path to a run goes
    # through this call. A branch whose workspace no longer satisfies it was downgraded to a
    # plain version by check_branch_invariant, so there is no warm start left to resolve.
    try:
        if not check_branch_invariant(study_dir).get("ok", True):
            return None
    except Exception:
        pass
    man = load_manifest(study_dir)
    if not man:
        return None
    cur = man.get("current")
    entry = next((v for v in man.get("versions", []) if v.get("id") == cur), None)
    ws = (entry or {}).get("warm_start")
    if not ws or not ws.get("source"):
        return None
    db = version_trajectory_db(study_dir, ws["source"])
    if not db.exists():
        return None
    return {"source_version": ws["source"], "source_trajectory": str(db),
            "resume_from": int(ws.get("resume_from", 0))}

discover_studies

discover_studies(
    studies_root: str | Path = "studies",
) -> list

The runtime 'catalog': glob studies/*/study.yaml -> list[StudySpec].

sv-init Step 0 calls this to see what has already been adapted, then matches a new query against the discovery fields (domain / tags / legacy_simulator / metrics / adjustable_params) to choose Path A (reuse) vs Path B (build new). Decentralised by design — each study carries its own card in its own study.yaml, so collaborators add studies without editing a shared index (no merge conflicts). Malformed/partial study.yaml files are skipped rather than raising.

Source code in skills/sv_workspace.py

def discover_studies(studies_root: str | Path = "studies") -> list:
    """The runtime 'catalog': glob ``studies/*/study.yaml`` -> list[StudySpec].

    sv-init Step 0 calls this to see what has already been adapted, then matches a new
    query against the discovery fields (domain / tags / legacy_simulator / metrics /
    adjustable_params) to choose Path A (reuse) vs Path B (build new). Decentralised by
    design — each study carries its own card in its own study.yaml, so collaborators add
    studies without editing a shared index (no merge conflicts). Malformed/partial
    study.yaml files are skipped rather than raising.
    """
    out = []
    for p in sorted(Path(studies_root).glob("*/study.yaml")):
        try:
            out.append(load_study_yaml(p))
        except Exception:
            continue
    return out

sibling_root

sibling_root(name: str) -> Path

Where the companion repo name is looked for: $ (process env, else .env), else ../.

Source code in skills/sv_workspace.py

def sibling_root(name: str) -> Path:
    """Where the companion repo `name` is looked for: $<its env var> (process env, else
    `.env`), else ../<name>."""
    from socioverse.external_events import env_setting

    info = SIBLING_REPOS.get(name) or {}
    env = env_setting(info["env"]) if info.get("env") else None
    return Path(env) if env else _REPO_ROOT.parent / name

requirement_status

requirement_status(req: str) -> dict

Resolve one requires entry -> {requirement, ok, where, setup}.

Supported kinds

sibling:<Name> a companion repo on disk (see SIBLING_REPOS); extra:<name> the Python packages a seam imports, installed by pip install -e ".[<name>]" (abm, chicago, workbench); legacy:chicago the legacy Chicago model's data, located exactly as the chicago seam does ($SV_CHICAGO_LEGACY, $SV_ABM_ROOT, ../SocioVerse-ABM).

Unknown kinds are reported as not satisfied so a typo never silently marks a study runnable.

Source code in skills/sv_workspace.py

def requirement_status(req: str) -> dict:
    """Resolve one `requires` entry -> {requirement, ok, where, setup}.

    Supported kinds:
      ``sibling:<Name>``   a companion repo on disk (see SIBLING_REPOS);
      ``extra:<name>``     the Python packages a seam imports, installed by
                           ``pip install -e ".[<name>]"`` (``abm``, ``chicago``,
                           ``workbench``);
      ``legacy:chicago``   the legacy Chicago model's data, located exactly as the chicago
                           seam does ($SV_CHICAGO_LEGACY, $SV_ABM_ROOT, ../SocioVerse-ABM).
    Unknown kinds are reported as not satisfied so a typo never silently marks a study runnable."""
    kind, _, name = req.partition(":")
    if kind == "sibling" and name:
        root = sibling_root(name)
        markers = (SIBLING_REPOS.get(name) or {}).get("markers") or []
        ok = root.is_dir() and all((root / m).exists() for m in markers)
        url = (SIBLING_REPOS.get(name) or {}).get("url")
        setup = (f"git clone {url} ../{name}" if url else f"clone {name} next to this repo")
        env = (SIBLING_REPOS.get(name) or {}).get("env")
        if env:
            setup += f"  (or set {env})"
        return {"requirement": req, "ok": ok, "where": str(root), "setup": setup}
    if kind == "extra" and name:
        try:
            missing = _extra_missing(name)
        except Exception as exc:                      # a broken seam import counts as unmet
            missing = [f"seam import failed: {exc}"]
        if missing is not None:
            return {"requirement": req, "ok": not missing, "where": "",
                    "setup": f'pip install -e ".[{name}]"  (missing: {", ".join(missing)})'}
    if kind == "legacy" and name == "chicago":
        url = SIBLING_REPOS["SocioVerse-ABM"]["url"]
        setup = (f"git clone {url} ../SocioVerse-ABM  (or set SV_CHICAGO_LEGACY to the legacy "
                 "dir, or SV_ABM_ROOT to a SocioVerse-ABM checkout)")
        try:
            seam = _seam_module("chicago")
            root = seam.resolve_chicago_legacy()
            ok, where = seam.chicago_data_present(root), str(root)
        except Exception as exc:                      # a broken seam import counts as unmet
            ok, where = False, f"seam import failed: {exc}"
        return {"requirement": req, "ok": ok, "where": where, "setup": setup}
    return {"requirement": req, "ok": False, "where": "",
            "setup": f"unknown requirement {req!r}"}

study_availability

study_availability(spec) -> dict

{requires, available, missing, setup, rerunnable} for one StudySpec.

Source code in skills/sv_workspace.py

def study_availability(spec) -> dict:
    """{requires, available, missing, setup, rerunnable} for one StudySpec."""
    reqs = list(spec.requires or [])
    status = [requirement_status(r) for r in reqs]
    missing = [s["requirement"] for s in status if not s["ok"]]
    return {
        "requires": reqs,
        "available": not missing,
        "missing": missing,
        "setup": [s["setup"] for s in status if not s["ok"]],
        "rerunnable": bool(spec.rerunnable),
    }

catalog_view

catalog_view(
    studies_root: str | Path = "studies",
) -> list[dict]

Compact, matching-relevant projection of discover_studies() for display/routing.

Besides the discovery fields, every row carries availability: requires (the study.yaml list), available (every requirement is present on disk), missing (the unmet ones), setup (how to satisfy each missing one) and rerunnable (False for imported references). sv-init offers a Path-A fork or a run only when available and rerunnable; otherwise it shows the study as "needs " with setup.

Source code in skills/sv_workspace.py

def catalog_view(studies_root: str | Path = "studies") -> list[dict]:
    """Compact, matching-relevant projection of discover_studies() for display/routing.

    Besides the discovery fields, every row carries availability: ``requires`` (the
    study.yaml list), ``available`` (every requirement is present on disk), ``missing``
    (the unmet ones), ``setup`` (how to satisfy each missing one) and ``rerunnable``
    (False for imported references). sv-init offers a Path-A fork or a run only when
    ``available and rerunnable``; otherwise it shows the study as "needs <X>" with ``setup``.
    """
    fields = ("study_id", "title", "domain", "tags", "study_type", "metrics",
              "legacy_simulator", "provider_refs", "adjustable_params", "status",
              "demonstrates", "teaches", "reference")
    return [{**{f: getattr(s, f) for f in fields}, **study_availability(s)}
            for s in discover_studies(studies_root)]

load_capabilities

load_capabilities(
    path: str | Path | None = None,
) -> list[dict]

Raw registry entries (list of dicts); [] when the registry is absent or empty.

Lazy-imports yaml so environments without PyYAML can still use every other helper.

Source code in skills/sv_workspace.py

def load_capabilities(path: str | Path | None = None) -> list[dict]:
    """Raw registry entries (list of dicts); [] when the registry is absent or empty.

    Lazy-imports yaml so environments without PyYAML can still use every other helper."""
    import yaml

    p = Path(path) if path is not None else CAPABILITIES_PATH
    if not p.exists():
        return []
    data = yaml.safe_load(p.read_text(encoding="utf-8")) or []
    return [e for e in data if isinstance(e, dict) and e.get("name")]

capability_view

capability_view(
    stage: str | None = None,
    path: str | Path | None = None,
    mcp_json: str | Path | None = None,
) -> list[dict]

catalog_view's sibling for external SERVICES — what each build skill's generic "capability check" hook lists.

stage filters to the entries anchored at that skill (their stages field); None returns all. Each entry is returned verbatim plus available / availability_note resolved from local config. Adding, removing, or changing a service is an edit to the registry only — the skills' hook text never changes.

Source code in skills/sv_workspace.py

def capability_view(stage: str | None = None, path: str | Path | None = None,
                    mcp_json: str | Path | None = None) -> list[dict]:
    """``catalog_view``'s sibling for external SERVICES — what each build skill's generic
    "capability check" hook lists.

    ``stage`` filters to the entries anchored at that skill (their ``stages`` field);
    ``None`` returns all. Each entry is returned verbatim plus ``available`` /
    ``availability_note`` resolved from local config. Adding, removing, or changing a
    service is an edit to the registry only — the skills' hook text never changes.
    """
    mj = Path(mcp_json) if mcp_json is not None else MCP_JSON_PATH
    out = []
    for e in load_capabilities(path):
        if stage and stage not in (e.get("stages") or []):
            continue
        avail, note = _capability_available(e, mj)
        out.append({**e, "available": avail, "availability_note": note})
    return out

skills.sv_grounding — the provenance sidecar

Read/merge access to studies/<id>/grounding/grounding.json. merge upserts facts and assumptions by id, so every build stage can extend the ledger without clobbering an earlier stage's entries, and writes atomically.

Deliberately not a Pydantic handoff: the ledger is append-friendly documentation for review, not a runtime-validated contract. See Grounding & Provenance.

skills.sv_grounding

Grounding sidecar for SocioVerse2 studies (stdlib only — no socioverse import).

studies/<id>/grounding/grounding.json records the study's real-world anchors: facts with provenance, implementation_refs (how similar phenomena are modeled), and declared assumptions. Deliberately NOT a pydantic handoff artifact — a flat, merge-friendly JSON the sv-* stages append to ("ground before you invent") and the dashboard renders read-only. Downloaded reference tables (small, MB-scale) live next to it under grounding/data/ and are pointed at by source.local_path.

Document shape (convention, not schema):

{"study_id": "...", "query": "...", "updated_at": "...", "method_notes": "...",
 "implementation_refs": [{"title", "url", "takeaway", "accessed"}],
 "facts": [{"id", "claim", "value", "unit", "as_of",
            "basis": "sourced" | "proxy" | "assumed",
            "source": {"title", "url",
                       "via": "event_service" | "web_search" | "provider" | "user",
                       "accessed", "local_path"?},
            "applies_to": ["environment.provider_args.base", ...], "note"}],
 "assumptions": [{"id", "claim", "rationale"}]}

method_notes="stylized" + empty facts records the decision that a study has no real-world anchors (abstract/toy models) — distinct from grounding merely missing.

load

load(study_dir: str | Path) -> dict | None

Parsed grounding.json, or None when missing/corrupt (callers treat as empty).

Source code in skills/sv_grounding.py

def load(study_dir: str | Path) -> dict | None:
    """Parsed grounding.json, or None when missing/corrupt (callers treat as empty)."""
    p = grounding_path(study_dir)
    if not p.exists():
        return None
    try:
        return json.loads(p.read_text(encoding="utf-8"))
    except Exception:
        return None

merge

merge(
    study_dir: str | Path,
    *,
    facts=(),
    implementation_refs=(),
    assumptions=(),
    method_notes: str | None = None,
    query: str | None = None,
) -> Path

Merge entries into grounding.json (creating it, and grounding/, if absent).

Facts/assumptions upsert by id (order preserved; missing ids assigned f<N>/ a<N> past collisions); refs dedupe by url; method_notes/query overwrite only when not None. updated_at bumps on every merge. Write is atomic (tmp + rename, like sv_workspace._write_manifest) — the dashboard poller must never read a torn file. Returns the file path.

Source code in skills/sv_grounding.py

def merge(study_dir: str | Path, *, facts=(), implementation_refs=(), assumptions=(),
          method_notes: str | None = None, query: str | None = None) -> Path:
    """Merge entries into grounding.json (creating it, and ``grounding/``, if absent).

    Facts/assumptions upsert by ``id`` (order preserved; missing ids assigned ``f<N>``/
    ``a<N>`` past collisions); refs dedupe by url; ``method_notes``/``query`` overwrite
    only when not None. ``updated_at`` bumps on every merge. Write is atomic (tmp +
    rename, like sv_workspace._write_manifest) — the dashboard poller must never read
    a torn file. Returns the file path.
    """
    doc = load(study_dir) or _skeleton(study_dir)
    doc["facts"] = _merge_by_id(doc.get("facts") or [], facts, "f")
    doc["assumptions"] = _merge_by_id(doc.get("assumptions") or [], assumptions, "a")
    doc["implementation_refs"] = _merge_refs(doc.get("implementation_refs") or [],
                                             implementation_refs)
    if method_notes is not None:
        doc["method_notes"] = method_notes
    if query is not None:
        doc["query"] = query
    doc["updated_at"] = _now_iso()

    p = grounding_path(study_dir)
    p.parent.mkdir(parents=True, exist_ok=True)  # tolerate pre-grounding scaffolds
    tmp = p.with_name(p.name + ".tmp")
    tmp.write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp.replace(p)
    return p

summary

summary(study_dir: str | Path) -> str

One-liner for sv_emit narratives / stage reports.

'12 facts: 8 sourced / 2 proxy / 2 assumed · 3 refs · 2 assumptions' 'stylized model, no real-world anchors' (method_notes=="stylized", no facts) 'no grounding recorded' (file missing/corrupt)

Source code in skills/sv_grounding.py

def summary(study_dir: str | Path) -> str:
    """One-liner for sv_emit narratives / stage reports.

    '12 facts: 8 sourced / 2 proxy / 2 assumed · 3 refs · 2 assumptions'
    'stylized model, no real-world anchors'  (method_notes=="stylized", no facts)
    'no grounding recorded'                  (file missing/corrupt)
    """
    doc = load(study_dir)
    if doc is None:
        return "no grounding recorded"
    facts = doc.get("facts") or []
    if not facts and doc.get("method_notes") == "stylized":
        return "stylized model, no real-world anchors"
    counts = {b: 0 for b in BASES}
    for f in facts:
        if f.get("basis") in counts:
            counts[f["basis"]] += 1
    parts = [f"{len(facts)} facts: " + " / ".join(f"{counts[b]} {b}" for b in BASES)]
    if doc.get("implementation_refs"):
        parts.append(f"{len(doc['implementation_refs'])} refs")
    if doc.get("assumptions"):
        parts.append(f"{len(doc['assumptions'])} assumptions")
    return " · ".join(parts)