Meta-Transfer

Why this subsystem exists

This is a bootstrap surface for policy transfer between domains, not a production adaptive controller. Its role is to make replay history reusable by giving operators an interpretable first proposal from prior domains.

In enterprise contexts, this reduces “blank start” risk for new deployments: teams can start from documented historical baselines rather than writing new policy defaults from scratch.

Decision point for operators

Meta-transfer is a bootstrap, not a controller. Its value is highest during:

• first-day onboarding of a new domainpack,
• recovery after a topology drift event,
• and cross-domain comparison when handoff teams need a first proposal quickly.

Because every output is review-only, teams preserve human authority over policy promotion while gaining a deterministic starting point from prior audited history.

Governance rule

All proposals from this surface remain advisory until the normal policy stack approves them. Evidence packages should be treated as hypothesis-generating artefacts and validated through the same deterministic replay channels as any other control-change candidate.

The meta-transfer subsystem provides a deterministic first slice for cross-domain policy bootstrapping. It reads replay or audit-derived records, embeds domain metrics into a shared feature vector, and proposes initial supervisor knobs from nearest historical neighbours.

This is not an online autonomous trainer. Proposals are reviewable starting points for policy authors, and every proposal exposes neighbour evidence and a serialisable audit record.

from scpn_phase_orchestrator.meta import CrossDomainMetaTransfer, MetaPolicyRecord

records = (
    MetaPolicyRecord("power_grid", {"R_global": 0.4}, {"K": 0.08}),
    MetaPolicyRecord("cardiac", {"R_global": 0.8}, {"zeta": 0.05}),
)
model = CrossDomainMetaTransfer.fit(records)
proposal = model.propose({"R_global": 0.5})

audit_payload = proposal.to_audit_record()

Larger replay corpora can be loaded from explicit audit JSONL file lists with CrossDomainMetaTransfer.fit_audit_history() or from nested audit directories with CrossDomainMetaTransfer.fit_audit_directory(). Directory loading uses records_from_audit_directory() and discovers **/*.jsonl by default, so multi-domain replay corpora can be trained without hand-listing every audit file. The fitted model exposes an audit-ready training_summary with record count, domain count, feature keys, knob keys, and reward range. Use to_json_package() and from_json_package() to save and restore a deterministic review package for proposal jobs. Audit JSONL ingestion and JSON package import reject non-finite constants, duplicate object keys, and non-object package payloads before records enter the nearest-neighbour proposal surface. Domain labels must be canonical strings; feature, knob, action, and reward evidence must be finite real numbers, with booleans, complex values, and numeric strings rejected rather than coerced. Validated feature and knob mappings are copied into immutable snapshots, so later caller mutation cannot rewrite the replay corpus or proposal evidence. to_package_manifest() emits a packaging-readiness manifest for the optional scpn-meta surface: it binds the deterministic JSON package SHA-256, public import target, console-script name, and training summary while keeping execution_permitted=false. It does not build, install, run, or upload a package.

The same review-only manifest can be emitted from the CLI for release and operator review jobs:

spo meta-transfer-manifest audit_grid.jsonl audit_cardiac.jsonl --min-records 2
spo meta-transfer-manifest --audit-directory audit_history --min-records 10

Both forms print manifest JSON to stdout unless --output is provided. The command accepts explicit audit JSONL files or one nested audit directory, never both, and still keeps execution_permitted=false; it does not build, install, upload, or execute scpn-meta.

Installed packages also expose the same review-only command as scpn-meta. This console script is intentionally narrow: it points to the manifest exporter, not the full SPO runtime CLI, so packaging metadata matches the manifest without adding a live training or execution surface.

How teams typically use it

The operational path is usually:

1. Collect comparable replay corpus (or nested history directory),
2. Fit and inspect training_summary,
3. Generate proposals and review neighbour evidence,
4. Export a manifest for reproducible transfer handoff.

That sequence keeps transfer evidence, not just transfer parameters, part of the release documentation.

Leave-one-domain-out transfer sweep

Proposing knobs is one thing; claiming a detector generalises across domains is a far stronger, and more easily overstated, claim. meta.leave_one_domain_out runs the honest test: hold out each domain in turn, transfer the pooled remainder onto it, and aggregate the per-fold verdicts under a rule that never upgrades. A single domain that is detectable within-domain yet receives no transfer skill (transfer_negative) refutes generality decisively — the recorded CHB-MIT cross-subject negative must surface as lodo_negative, never a laundered aggregate positive. Only an unbroken sweep of positive folds earns lodo_generalises; a sweep with no detectable target at all is lodo_untestable; anything in between is lodo_inconclusive.

from scpn_phase_orchestrator.meta import (
    LeaveOneDomainOutFold,
    leave_one_domain_out_transfer,
)

report = leave_one_domain_out_transfer(
    [
        LeaveOneDomainOutFold("grid", transfer, within, controls),
        LeaveOneDomainOutFold("chbmit", chb_transfer, chb_within, chb_controls),
    ]
)
verdict = report.verdict  # e.g. "lodo_negative"
audit_payload = report.to_record()

Every arm is scored by the caller and audited through the same honest audit_cross_domain_transfer calibration, so the sweep stays a pure, deterministic aggregation with no hidden training step. The public verdict classifier also rejects empty sweeps, unknown fold verdicts, and impossible testable-fold counts, so malformed evidence cannot accidentally earn lodo_generalises.

transfer

Replay-backed cross-domain policy proposal utilities.

Classes

MetaPolicyRecord dataclass

MetaPolicyRecord(
    domain: str,
    features: Mapping[str, float],
    knobs: Mapping[str, float],
    reward: float = 1.0,
)

One replay-derived domain policy example.

MetaTransferProposal dataclass

MetaTransferProposal(
    knobs: Mapping[str, float],
    confidence: float,
    neighbours: tuple[tuple[str, float], ...],
    feature_keys: tuple[str, ...],
)

Initial policy proposal for a new domain signature.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a serialisable proposal record.

Returns

dict[str, object] A serialisable proposal record.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable proposal record.

    Returns
    -------
    dict[str, object]
        A serialisable proposal record.
    """
    return {
        "knobs": dict(self.knobs),
        "confidence": self.confidence,
        "neighbours": [
            {"domain": domain, "similarity": similarity}
            for domain, similarity in self.neighbours
        ],
        "feature_keys": list(self.feature_keys),
        "method": "cosine_nearest_policy_transfer",
    }

MetaTrainingSummary dataclass

MetaTrainingSummary(
    record_count: int,
    domain_count: int,
    domains: tuple[str, ...],
    feature_keys: tuple[str, ...],
    knob_keys: tuple[str, ...],
    reward_mean: float,
    reward_min: float,
    reward_max: float,
)

Audit-ready summary of the replay corpus used for meta-transfer.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe training corpus summary.

Returns

dict[str, object] A JSON-safe training corpus summary.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe training corpus summary.

    Returns
    -------
    dict[str, object]
        A JSON-safe training corpus summary.
    """
    return {
        "record_count": self.record_count,
        "domain_count": self.domain_count,
        "domains": list(self.domains),
        "feature_keys": list(self.feature_keys),
        "knob_keys": list(self.knob_keys),
        "reward_mean": self.reward_mean,
        "reward_min": self.reward_min,
        "reward_max": self.reward_max,
    }

MetaPackageManifest dataclass

MetaPackageManifest(
    package_name: str,
    import_target: str,
    console_script: str,
    package_sha256: str,
    training_summary: MetaTrainingSummary,
    execution_permitted: bool = False,
)

Packaging-readiness manifest for the optional scpn-meta surface.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe packaging-readiness manifest.

Returns

dict[str, object] A JSON-safe packaging-readiness manifest.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe packaging-readiness manifest.

    Returns
    -------
    dict[str, object]
        A JSON-safe packaging-readiness manifest.
    """
    return {
        "schema": "scpn_meta_package_manifest_v1",
        "package_name": self.package_name,
        "import_target": self.import_target,
        "console_script": self.console_script,
        "package_sha256": self.package_sha256,
        "training_summary": self.training_summary.to_audit_record(),
        "execution_permitted": self.execution_permitted,
    }

CrossDomainMetaTransfer

CrossDomainMetaTransfer(
    records: tuple[MetaPolicyRecord, ...],
)

Nearest-neighbour policy transfer over replay-derived embeddings.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def __init__(self, records: tuple[MetaPolicyRecord, ...]) -> None:
    if not records:
        raise ValueError("at least one meta-policy record is required")
    self.records = records
    self.feature_keys = _feature_keys(records)
    self._matrix = np.vstack(
        [_feature_vector(record.features, self.feature_keys) for record in records]
    )
    self.training_summary = _training_summary(records, self.feature_keys)

Methods:

fit classmethod

fit(
    records: list[MetaPolicyRecord]
    | tuple[MetaPolicyRecord, ...],
) -> CrossDomainMetaTransfer

Construct a meta-transfer model from replay-derived records.

Parameters

records : list[MetaPolicyRecord] | tuple[MetaPolicyRecord, ...] The records to summarise.

Returns

CrossDomainMetaTransfer A meta-transfer model from replay-derived records.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

@classmethod
def fit(
    cls, records: list[MetaPolicyRecord] | tuple[MetaPolicyRecord, ...]
) -> CrossDomainMetaTransfer:
    """Construct a meta-transfer model from replay-derived records.

    Parameters
    ----------
    records : list[MetaPolicyRecord] | tuple[MetaPolicyRecord, ...]
        The records to summarise.

    Returns
    -------
    CrossDomainMetaTransfer
        A meta-transfer model from replay-derived records.
    """
    return cls(tuple(records))

fit_audit_history classmethod

fit_audit_history(
    paths: list[str | Path] | tuple[str | Path, ...],
    *,
    min_records: int = 1,
) -> CrossDomainMetaTransfer

Fit a model from one or more audit JSONL files.

Parameters

paths : list[str | Path] | tuple[str | Path, ...] Filesystem paths. min_records : int Minimum number of records required.

Returns

CrossDomainMetaTransfer Fit a model from one or more audit JSONL files.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

@classmethod
def fit_audit_history(
    cls,
    paths: list[str | Path] | tuple[str | Path, ...],
    *,
    min_records: int = 1,
) -> CrossDomainMetaTransfer:
    """Fit a model from one or more audit JSONL files.

    Parameters
    ----------
    paths : list[str | Path] | tuple[str | Path, ...]
        Filesystem paths.
    min_records : int
        Minimum number of records required.

    Returns
    -------
    CrossDomainMetaTransfer
        Fit a model from one or more audit JSONL files.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if min_records < 1:
        raise ValueError("min_records must be at least 1")
    records: list[MetaPolicyRecord] = []
    for path in paths:
        records.extend(records_from_audit_jsonl(path))
    if len(records) < min_records:
        raise ValueError(
            f"audit history yielded {len(records)} records; "
            f"min_records={min_records}"
        )
    return cls.fit(tuple(records))

fit_audit_directory classmethod

fit_audit_directory(
    root: str | Path,
    *,
    pattern: str = "**/*.jsonl",
    min_records: int = 1,
) -> CrossDomainMetaTransfer

Fit a model from a recursively discovered audit JSONL corpus.

Parameters

root : str | Path Root directory to search. pattern : str Glob pattern for discovery. min_records : int Minimum number of records required.

Returns

CrossDomainMetaTransfer Fit a model from a recursively discovered audit JSONL corpus.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

@classmethod
def fit_audit_directory(
    cls,
    root: str | Path,
    *,
    pattern: str = "**/*.jsonl",
    min_records: int = 1,
) -> CrossDomainMetaTransfer:
    """Fit a model from a recursively discovered audit JSONL corpus.

    Parameters
    ----------
    root : str | Path
        Root directory to search.
    pattern : str
        Glob pattern for discovery.
    min_records : int
        Minimum number of records required.

    Returns
    -------
    CrossDomainMetaTransfer
        Fit a model from a recursively discovered audit JSONL corpus.
    """
    records = records_from_audit_directory(
        root,
        pattern=pattern,
        min_records=min_records,
    )
    return cls.fit(records)

propose

propose(
    features: dict[str, float], *, k_neighbours: int = 3
) -> MetaTransferProposal

Propose initial policy knobs for a new domain signature.

Parameters

features : dict[str, float] Input feature array. k_neighbours : int Number of nearest neighbours.

Returns

MetaTransferProposal Propose initial policy knobs for a new domain signature.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def propose(
    self,
    features: dict[str, float],
    *,
    k_neighbours: int = 3,
) -> MetaTransferProposal:
    """Propose initial policy knobs for a new domain signature.

    Parameters
    ----------
    features : dict[str, float]
        Input feature array.
    k_neighbours : int
        Number of nearest neighbours.

    Returns
    -------
    MetaTransferProposal
        Propose initial policy knobs for a new domain signature.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    _validate_float_mapping(features, "features", allow_empty=False)
    if k_neighbours < 1:
        raise ValueError("k_neighbours must be at least 1")
    query = _feature_vector(features, self.feature_keys)
    similarities = np.array(
        [_cosine_similarity(query, row) for row in self._matrix],
        dtype=np.float64,
    )
    order: IntArray = np.asarray(
        np.argsort(similarities)[::-1][: min(k_neighbours, len(self.records))],
        dtype=np.intp,
    )
    weights = _proposal_weights(similarities[order], self.records, order)
    knobs = _weighted_knobs(self.records, order, weights)
    confidence = float(np.clip(np.mean(similarities[order]), 0.0, 1.0))
    neighbours = tuple(
        (self.records[index].domain, float(similarities[index])) for index in order
    )
    return MetaTransferProposal(
        knobs=knobs,
        confidence=confidence,
        neighbours=neighbours,
        feature_keys=self.feature_keys,
    )

to_json_package

to_json_package() -> str

Serialise records and training summary for reviewable reuse.

Returns

str Serialise records and training summary for reviewable reuse.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def to_json_package(self) -> str:
    """Serialise records and training summary for reviewable reuse.

    Returns
    -------
    str
        Serialise records and training summary for reviewable reuse.
    """
    package = {
        "schema": "scpn_meta_transfer_package_v1",
        "training_summary": self.training_summary.to_audit_record(),
        "records": [
            {
                "domain": record.domain,
                "features": dict(record.features),
                "knobs": dict(record.knobs),
                "reward": record.reward,
            }
            for record in self.records
        ],
    }
    return json.dumps(package, indent=2, sort_keys=True) + "\n"

to_package_manifest

to_package_manifest(
    *,
    package_name: str = "scpn-meta",
    import_target: str = "scpn_phase_orchestrator.meta",
    console_script: str = "scpn-meta",
) -> MetaPackageManifest

Build a deterministic packaging-readiness manifest.

The manifest binds the JSON package hash and public import/console metadata for review jobs. It does not create distributions, install commands, run proposal jobs, or upload artefacts.

Parameters

package_name : str Name for the emitted package. import_target : str Import target for the generated package. console_script : str Console-script entry-point name.

Returns

MetaPackageManifest A deterministic packaging-readiness manifest.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def to_package_manifest(
    self,
    *,
    package_name: str = "scpn-meta",
    import_target: str = "scpn_phase_orchestrator.meta",
    console_script: str = "scpn-meta",
) -> MetaPackageManifest:
    """Build a deterministic packaging-readiness manifest.

    The manifest binds the JSON package hash and public import/console
    metadata for review jobs. It does not create distributions, install
    commands, run proposal jobs, or upload artefacts.

    Parameters
    ----------
    package_name : str
        Name for the emitted package.
    import_target : str
        Import target for the generated package.
    console_script : str
        Console-script entry-point name.

    Returns
    -------
    MetaPackageManifest
        A deterministic packaging-readiness manifest.
    """
    _validate_package_identifier(package_name, "package_name")
    _validate_non_empty_text(import_target, "import_target")
    _validate_non_empty_text(console_script, "console_script")
    return MetaPackageManifest(
        package_name=package_name,
        import_target=import_target,
        console_script=console_script,
        package_sha256=sha256(self.to_json_package().encode("utf-8")).hexdigest(),
        training_summary=self.training_summary,
        execution_permitted=False,
    )

from_json_package classmethod

from_json_package(payload: str) -> CrossDomainMetaTransfer

Restore a packaged meta-transfer model.

Parameters

payload : str The payload mapping or bytes.

Returns

CrossDomainMetaTransfer Restore a packaged meta-transfer model.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

@classmethod
def from_json_package(cls, payload: str) -> CrossDomainMetaTransfer:
    """Restore a packaged meta-transfer model.

    Parameters
    ----------
    payload : str
        The payload mapping or bytes.

    Returns
    -------
    CrossDomainMetaTransfer
        Restore a packaged meta-transfer model.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    data = _loads_meta_json(payload)
    if not isinstance(data, dict):
        raise ValueError("package must be a JSON object")
    if data.get("schema") != "scpn_meta_transfer_package_v1":
        raise ValueError("unsupported meta-transfer package schema")
    records_payload = data.get("records")
    if not isinstance(records_payload, list):
        raise ValueError("package records must be a list")
    records = [
        _record_from_payload(record, index)
        for index, record in enumerate(records_payload, start=1)
        if isinstance(record, dict)
    ]
    if len(records) != len(records_payload):
        raise ValueError("package records must be objects")
    return cls.fit(tuple(records))

Functions:

records_from_audit_jsonl

records_from_audit_jsonl(
    path: str | Path,
) -> tuple[MetaPolicyRecord, ...]

Load meta-policy records from audit-style JSONL lines.

Each line may provide either explicit features and knobs mappings or the common SPO audit shape with metrics plus actions.

Parameters

path : str | Path Filesystem path to the target file.

Returns

tuple[MetaPolicyRecord, ...] Load meta-policy records from audit-style JSONL lines.

Raises

ValueError If a line is not a canonical JSON object or contains invalid evidence. OSError If the audit file cannot be opened or read.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def records_from_audit_jsonl(path: str | Path) -> tuple[MetaPolicyRecord, ...]:
    """Load meta-policy records from audit-style JSONL lines.

    Each line may provide either explicit ``features`` and ``knobs`` mappings
    or the common SPO audit shape with ``metrics`` plus ``actions``.

    Parameters
    ----------
    path : str | Path
        Filesystem path to the target file.

    Returns
    -------
    tuple[MetaPolicyRecord, ...]
        Load meta-policy records from audit-style JSONL lines.

    Raises
    ------
    ValueError
        If a line is not a canonical JSON object or contains invalid evidence.
    OSError
        If the audit file cannot be opened or read.
    """
    records: list[MetaPolicyRecord] = []
    with Path(path).open("r", encoding="utf-8") as handle:
        for line_number, line in enumerate(handle, start=1):
            stripped = line.strip()
            if not stripped:
                continue
            payload = _loads_meta_json(stripped)
            if not isinstance(payload, dict):
                raise ValueError(
                    f"line {line_number}: audit record must be a JSON object"
                )
            records.append(_record_from_payload(payload, line_number))
    return tuple(records)

records_from_audit_directory

records_from_audit_directory(
    root: str | Path,
    *,
    pattern: str = "**/*.jsonl",
    min_records: int = 1,
) -> tuple[MetaPolicyRecord, ...]

Load replay records from a nested audit-history directory.

Parameters

root : str | Path Root directory to search. pattern : str Glob pattern for discovery. min_records : int Minimum number of records required.

Returns

tuple[MetaPolicyRecord, ...] Load replay records from a nested audit-history directory.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/meta/transfer.py

def records_from_audit_directory(
    root: str | Path,
    *,
    pattern: str = "**/*.jsonl",
    min_records: int = 1,
) -> tuple[MetaPolicyRecord, ...]:
    """Load replay records from a nested audit-history directory.

    Parameters
    ----------
    root : str | Path
        Root directory to search.
    pattern : str
        Glob pattern for discovery.
    min_records : int
        Minimum number of records required.

    Returns
    -------
    tuple[MetaPolicyRecord, ...]
        Load replay records from a nested audit-history directory.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if min_records < 1:
        raise ValueError("min_records must be at least 1")
    base = Path(root)
    if not base.exists() or not base.is_dir():
        raise ValueError("audit directory must exist")
    paths = tuple(sorted(path for path in base.glob(pattern) if path.is_file()))
    if not paths:
        raise ValueError("audit directory yielded no JSONL files")
    records: list[MetaPolicyRecord] = []
    for path in paths:
        records.extend(records_from_audit_jsonl(path))
    if len(records) < min_records:
        raise ValueError(
            f"audit directory yielded {len(records)} records; min_records={min_records}"
        )
    return tuple(records)

leave_one_domain_out

Aggregate cross-domain transfer honestly across a leave-one-domain-out sweep.

A single audit_cross_domain_transfer call answers does a detector trained elsewhere transfer to this one target?. A claim of domain-general transfer is a stronger, and far more easily overstated, thing: it needs every domain to survive being held out while the pooled remainder is transferred onto it. This module runs that leave-one-domain-out (LODO) sweep and aggregates the per-fold verdicts under a rule that, like the single-pair auditor it builds on, never upgrades — it fails toward caution and stays able to return a decisive negative.

Each :class:LeaveOneDomainOutFold carries the already-scored arms for one held-out target domain: the pooled-source detector's scores on that target, a target-trained detector's scores on the same segments (the within-domain detectability ceiling), and a shuffled-source floor ensemble. Scoring the arms is the caller's job — exactly as it is for the single-pair auditor — so this harness stays a pure, deterministic aggregation with no hidden training step.

:func:leave_one_domain_out_transfer audits every fold and hands the per-fold verdicts to :func:classify_lodo_verdict, which decides the sweep verdict:

• :data:LODO_NEGATIVE — at least one fold is a decisive :data:~scpn_phase_orchestrator.evaluation.cross_domain_transfer.TRANSFER_NEGATIVE (the target is detectable within-domain yet the transfer carried no skill). One such domain refutes generality; this is the verdict the recorded CHB-MIT cross-subject negative must produce, never a laundered aggregate positive.
• :data:LODO_GENERALISES — every fold is a positive transfer. Only an unbroken sweep of positives earns the general claim.
• :data:LODO_UNTESTABLE — no fold was detectable even within-domain, so transfer is untestable across the whole sweep, not refuted.
• :data:LODO_INCONCLUSIVE — anything in between: a mix of positives and nulls with no decisive negative and no clean sweep. The evidence neither supports nor refutes domain-general transfer.

Classes

LeaveOneDomainOutFold dataclass

LeaveOneDomainOutFold(
    target_domain: str,
    transfer: ScorePair,
    within_domain: ScorePair,
    shuffled_source: tuple[ScorePair, ...],
)

The precomputed transfer arms for one held-out target domain.

Attributes

target_domain : str Label of the domain held out on this fold; the pooled remainder of the sweep is the transfer source. transfer : ScorePair The pooled-source detector's scores on the held-out target's event and null segments. within_domain : ScorePair A target-trained detector's scores on the same target segments — the detectability ceiling that makes a transfer failure decisive. shuffled_source : tuple[ScorePair, ...] Controls in which the source signal is scrambled before scoring the target; their skill margins form the floor a genuine transfer must beat. At least one is required, normalised to a tuple at construction.

LeaveOneDomainOutReport dataclass

LeaveOneDomainOutReport(
    verdict: str,
    folds: tuple[CrossDomainTransferAudit, ...],
    n_domains: int,
    n_testable: int,
    n_positive: int,
    alpha: float,
)

The aggregated verdict of a leave-one-domain-out transfer sweep.

Attributes

verdict : str One of :data:LODO_NEGATIVE, :data:LODO_GENERALISES, :data:LODO_UNTESTABLE, :data:LODO_INCONCLUSIVE. folds : tuple[CrossDomainTransferAudit, ...] The per-fold transfer audits, in the order the folds were supplied. n_domains : int Number of held-out domains in the sweep. n_testable : int Number of folds whose held-out target is detectable within-domain — the folds on which a transfer failure would be decisive. n_positive : int Number of folds whose transfer verdict is a positive transfer. alpha : float Significance level at which every fold gate was decided.

Attributes

domain_verdicts property

domain_verdicts: dict[str, str]

Map each held-out target domain to its single-fold transfer verdict.

verdict_counts property

verdict_counts: dict[str, int]

Return the count of each single-fold transfer verdict in the sweep.

Methods:

to_record

to_record() -> dict[str, object]

Return a JSON-safe mapping of the leave-one-domain-out verdict.

Returns

dict[str, object] The sweep verdict, the per-domain fold verdicts, the verdict counts, the testable and positive fold tallies, the significance level, and the full per-fold transfer audit records.

Source code in src/scpn_phase_orchestrator/meta/leave_one_domain_out.py

def to_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the leave-one-domain-out verdict.

    Returns
    -------
    dict[str, object]
        The sweep verdict, the per-domain fold verdicts, the verdict counts,
        the testable and positive fold tallies, the significance level, and the
        full per-fold transfer audit records.
    """
    return {
        "schema": "scpn_leave_one_domain_out_transfer_v1",
        "verdict": self.verdict,
        "n_domains": self.n_domains,
        "n_testable": self.n_testable,
        "n_positive": self.n_positive,
        "alpha": self.alpha,
        "domain_verdicts": self.domain_verdicts,
        "verdict_counts": self.verdict_counts,
        "folds": [fold.to_record() for fold in self.folds],
    }

Functions:

classify_lodo_verdict

classify_lodo_verdict(
    fold_verdicts: Sequence[str], *, n_testable: int
) -> str

Aggregate per-fold transfer verdicts into a leave-one-domain-out verdict.

The rule never upgrades: a single decisive negative refutes generality, and only an unbroken sweep of positives earns the general claim.

Parameters

fold_verdicts : sequence of str The single-fold transfer verdicts, one per held-out domain. Must be non-empty; the harness enforces at least two folds. n_testable : int Number of folds whose held-out target is detectable within-domain.

Returns

str :data:LODO_NEGATIVE if any fold is a decisive transfer negative; :data:LODO_GENERALISES if every fold is a positive transfer; :data:LODO_UNTESTABLE if no fold was detectable within-domain; :data:LODO_INCONCLUSIVE otherwise.

Raises

ValueError If the verdict sequence is empty, contains an unsupported verdict, or conflicts with n_testable.

Source code in src/scpn_phase_orchestrator/meta/leave_one_domain_out.py

def classify_lodo_verdict(
    fold_verdicts: Sequence[str],
    *,
    n_testable: int,
) -> str:
    """Aggregate per-fold transfer verdicts into a leave-one-domain-out verdict.

    The rule never upgrades: a single decisive negative refutes generality, and only
    an unbroken sweep of positives earns the general claim.

    Parameters
    ----------
    fold_verdicts : sequence of str
        The single-fold transfer verdicts, one per held-out domain. Must be
        non-empty; the harness enforces at least two folds.
    n_testable : int
        Number of folds whose held-out target is detectable within-domain.

    Returns
    -------
    str
        :data:`LODO_NEGATIVE` if any fold is a decisive transfer negative;
        :data:`LODO_GENERALISES` if every fold is a positive transfer;
        :data:`LODO_UNTESTABLE` if no fold was detectable within-domain;
        :data:`LODO_INCONCLUSIVE` otherwise.

    Raises
    ------
    ValueError
        If the verdict sequence is empty, contains an unsupported verdict, or
        conflicts with ``n_testable``.
    """
    verdicts = tuple(fold_verdicts)
    if not verdicts:
        raise ValueError("at least one fold verdict is required")
    allowed = {TRANSFER_NEGATIVE, TRANSFER_NULL, TRANSFER_POSITIVE}
    unsupported = sorted(set(verdicts) - allowed)
    if unsupported:
        raise ValueError(f"unsupported fold verdict: {unsupported[0]!r}")
    if (
        isinstance(n_testable, bool)
        or not isinstance(n_testable, int)
        or not 0 <= n_testable <= len(verdicts)
    ):
        raise ValueError("n_testable must be an integer within the fold count")
    decisive = sum(verdict != TRANSFER_NULL for verdict in verdicts)
    if n_testable < decisive:
        raise ValueError("n_testable is inconsistent with decisive fold verdicts")
    if any(verdict == TRANSFER_NEGATIVE for verdict in verdicts):
        return LODO_NEGATIVE
    if all(verdict == TRANSFER_POSITIVE for verdict in verdicts):
        return LODO_GENERALISES
    if n_testable == 0:
        return LODO_UNTESTABLE
    return LODO_INCONCLUSIVE

leave_one_domain_out_transfer

leave_one_domain_out_transfer(
    folds: Sequence[LeaveOneDomainOutFold],
    *,
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> LeaveOneDomainOutReport

Run a leave-one-domain-out cross-domain transfer sweep and aggregate it.

Each fold is audited through audit_cross_domain_transfer with identical calibration, and the per-fold verdicts are aggregated by :func:classify_lodo_verdict. The source of every fold is labelled as the pooled remainder of the sweep (pooled-not-<target>).

Parameters

folds : sequence of LeaveOneDomainOutFold The precomputed transfer arms, one per held-out target domain. At least two distinct target domains are required. target_false_alarm : float The false-alarm rate every arm's threshold is calibrated to hold. n_permutations : int Random relabellings drawn for each arm's label-permutation p-value. seed : int Seed of the permutation resampling, so the sweep is reproducible. alpha : float Significance level at which each gate is decided.

Returns

LeaveOneDomainOutReport The aggregated sweep verdict with every per-fold transfer audit.

Raises

ValueError If fewer than two folds are supplied, or two folds name the same target domain.

Source code in src/scpn_phase_orchestrator/meta/leave_one_domain_out.py

def leave_one_domain_out_transfer(
    folds: Sequence[LeaveOneDomainOutFold],
    *,
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> LeaveOneDomainOutReport:
    """Run a leave-one-domain-out cross-domain transfer sweep and aggregate it.

    Each fold is audited through ``audit_cross_domain_transfer`` with identical
    calibration, and the per-fold verdicts are aggregated by
    :func:`classify_lodo_verdict`. The source of every fold is labelled as the pooled
    remainder of the sweep (``pooled-not-<target>``).

    Parameters
    ----------
    folds : sequence of LeaveOneDomainOutFold
        The precomputed transfer arms, one per held-out target domain. At least two
        distinct target domains are required.
    target_false_alarm : float
        The false-alarm rate every arm's threshold is calibrated to hold.
    n_permutations : int
        Random relabellings drawn for each arm's label-permutation p-value.
    seed : int
        Seed of the permutation resampling, so the sweep is reproducible.
    alpha : float
        Significance level at which each gate is decided.

    Returns
    -------
    LeaveOneDomainOutReport
        The aggregated sweep verdict with every per-fold transfer audit.

    Raises
    ------
    ValueError
        If fewer than two folds are supplied, or two folds name the same target
        domain.
    """
    if len(folds) < 2:
        raise ValueError("leave-one-domain-out requires at least two domains")
    targets = [fold.target_domain for fold in folds]
    if len(set(targets)) != len(targets):
        raise ValueError("each fold must hold out a distinct target domain")
    audits = tuple(
        audit_cross_domain_transfer(
            transfer=fold.transfer,
            within_domain=fold.within_domain,
            shuffled_source=fold.shuffled_source,
            source_domain=f"pooled-not-{fold.target_domain}",
            target_domain=fold.target_domain,
            target_false_alarm=target_false_alarm,
            n_permutations=n_permutations,
            seed=seed,
            alpha=alpha,
        )
        for fold in folds
    )
    n_testable = sum(1 for audit in audits if audit.within_domain.beats_chance)
    n_positive = sum(1 for audit in audits if audit.verdict == TRANSFER_POSITIVE)
    verdict = classify_lodo_verdict(
        [audit.verdict for audit in audits],
        n_testable=n_testable,
    )
    return LeaveOneDomainOutReport(
        verdict=verdict,
        folds=audits,
        n_domains=len(audits),
        n_testable=n_testable,
        n_positive=n_positive,
        alpha=alpha,
    )