feat(orchestration): compose security scenarios

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02929-244b-7391-b933-c04010e8eedb

tools/playbook-capability-contract/README.md

@@ -13,7 +13,7 @@ python3 tools/playbook-capability-contract/playbook_contract_validator.py \
   ../railiance-infra/capabilities/playbooks/railiance-infra.bootstrap-host.yaml
 ```
 
-Validate and compose a sample scenario:
+Validate and compose the legacy single-provider conformance sample:
 
 ```bash
 python3 tools/playbook-capability-contract/playbook_contract_validator.py \
@@ -21,6 +21,17 @@ python3 tools/playbook-capability-contract/playbook_contract_validator.py \
   --scenario examples/playbook-capability-contract/scenario-s1-host-bootstrap.yaml
 ```
 
+The `--scenario` path exists for v0.1 conformance compatibility. It refuses
+ambiguous providers and unselected overrides, but it does not perform trust
+sequencing or build a complete owner handoff. Use the canonical Security
+Scenario Composer for deterministic multi-provider plans:
+
+```bash
+python3 tools/security-scenario-composer/security_scenario_composer.py \
+  --scenario examples/security-scenarios/c0-local-identity.yaml \
+  capabilities/playbooks/net-kingdom.local-identity.yaml
+```
+
 Run tests:
 
 ```bash

tools/playbook-capability-contract/playbook_contract_validator.py

@@ -60,6 +60,9 @@ TRUST_STATES = {
     "tenant_onboarding_trust",
 }
 SCENARIO_AUTHORITIES = {"platform", "netkingdom", "tenant"}
+SECRET_REFERENCE_PATTERN = re.compile(
+    r"(?:[a-z][a-z0-9+.-]*://\S+|(?:/|\./|\.\./)\S+)"
+)
 
 
 @dataclass
@@ -127,6 +130,10 @@ def is_type(value: Any, declared_type: str) -> bool:
     return False
 
 
+def is_secret_reference(value: Any) -> bool:
+    return isinstance(value, str) and SECRET_REFERENCE_PATTERN.fullmatch(value) is not None
+
+
 def validate_constraints(value: Any, param: dict[str, Any], path: str) -> list[Issue]:
     issues: list[Issue] = []
     constraints = param.get("constraints", {})
@@ -277,6 +284,18 @@ def validate_parameters(spec: dict[str, Any]) -> list[Issue]:
         if item.get("sensitivity") in {"security_sensitive", "secret_reference"} and item.get("tuning_authority") == "tenant_tunable":
             issues.append(issue("ERROR", path, "security-sensitive parameters cannot be tenant_tunable"))
 
+        if item.get("sensitivity") == "secret_reference":
+            if declared_type != "string":
+                issues.append(issue("ERROR", f"{path}.type", "secret_reference parameters must use string type"))
+            if "default" in item and not is_secret_reference(item["default"]):
+                issues.append(
+                    issue(
+                        "ERROR",
+                        f"{path}.default",
+                        "secret_reference default must be a URI or explicit absolute/relative path",
+                    )
+                )
+
         if "default" in item and declared_type in PARAM_TYPES and not is_type(item["default"], declared_type):
             issues.append(issue("ERROR", f"{path}.default", f"default does not match type {declared_type!r}"))
         if declared_type in PARAM_TYPES:
@@ -405,6 +424,15 @@ def validate_override_allowed(param: dict[str, Any], value: Any, scenario_author
     if scenario_authority == "tenant" and sensitivity in {"security_sensitive", "secret_reference"}:
         issues.append(issue("ERROR", path, f"tenant authority cannot override {sensitivity} parameter {name!r}"))
 
+    if sensitivity == "secret_reference" and not is_secret_reference(value):
+        issues.append(
+            issue(
+                "ERROR",
+                path,
+                f"override for {name!r} must be a secret-reference URI or explicit absolute/relative path",
+            )
+        )
+
     if not is_type(value, declared_type):
         issues.append(issue("ERROR", path, f"override for {name!r} does not match type {declared_type!r}"))
     issues.extend(validate_constraints(value, param, path))
@@ -433,10 +461,23 @@ def compose_scenario(declarations: list[Declaration], scenario: dict[str, Any])
 
     selected: list[Declaration] = []
     for cap_id in required_caps:
-        matches = [declaration for declaration in declarations if cap_id in declaration.capabilities]
+        matches = sorted(
+            [declaration for declaration in declarations if cap_id in declaration.capabilities],
+            key=lambda declaration: declaration.id,
+        )
         if not matches:
             issues.append(issue("ERROR", "scenario.requires.capabilities", f"no declaration provides {cap_id!r}"))
             continue
+        if len(matches) > 1:
+            issues.append(
+                issue(
+                    "ERROR",
+                    "scenario.requires.capabilities",
+                    f"ambiguous providers {[declaration.id for declaration in matches]}; "
+                    "use the Security Scenario Composer to pin one explicitly",
+                )
+            )
+            continue
         selected.append(matches[0])
 
     # Preserve order while deduplicating declarations selected for several capabilities.
@@ -444,6 +485,16 @@ def compose_scenario(declarations: list[Declaration], scenario: dict[str, Any])
     for declaration in selected:
         selected_by_id.setdefault(declaration.id, declaration)
 
+    for declaration_id in overrides:
+        if declaration_id not in selected_by_id:
+            issues.append(
+                issue(
+                    "ERROR",
+                    f"scenario.parameter_overrides.{declaration_id}",
+                    "override targets an unselected declaration",
+                )
+            )
+
     composed = {
         "scenario": scenario.get("id", "scenario:unnamed"),
         "authority": authority,

tools/playbook-capability-contract/tests/test_playbook_contract_validator.py

@@ -142,6 +142,81 @@ def test_tenant_tunable_secret_reference_fails(tmp_path):
     assert any("security-sensitive parameters cannot be tenant_tunable" in msg for msg in error_messages(issues))
 
 
+def test_plaintext_secret_reference_default_fails(tmp_path):
+    data = valid_declaration()
+    data["spec"]["parameters"][2].update(
+        {
+            "type": "string",
+            "default": "plaintext-looking-value",
+            "sensitivity": "secret_reference",
+            "tuning_authority": "platform_only",
+        }
+    )
+    declaration = declaration_from(data, tmp_path)
+
+    issues = validator.validate_declaration(declaration)
+
+    assert any("secret_reference default must be" in msg for msg in error_messages(issues))
+
+
+def test_secret_reference_uri_override_passes(tmp_path):
+    data = valid_declaration()
+    data["spec"]["parameters"][2].update(
+        {
+            "name": "credential_ref",
+            "type": "string",
+            "default": "openbao://kv/platform/reference",
+            "sensitivity": "secret_reference",
+            "tuning_authority": "platform_only",
+        }
+    )
+    declaration = declaration_from(data, tmp_path)
+    scenario = {
+        "id": "scenario:secret-reference",
+        "authority": "platform",
+        "requires": {"capabilities": ["s1.os-baseline"]},
+        "parameter_overrides": {
+            "railiance-infra.bootstrap-host": {
+                "target_hosts": ["railiance01"],
+                "credential_ref": "kubernetes://platform/reference#token",
+            }
+        },
+    }
+
+    issues, _ = validator.compose_scenario([declaration], scenario)
+
+    assert error_messages(issues) == []
+
+
+def test_plaintext_secret_reference_override_fails(tmp_path):
+    data = valid_declaration()
+    data["spec"]["parameters"][2].update(
+        {
+            "name": "credential_ref",
+            "type": "string",
+            "default": "openbao://kv/platform/reference",
+            "sensitivity": "secret_reference",
+            "tuning_authority": "platform_only",
+        }
+    )
+    declaration = declaration_from(data, tmp_path)
+    scenario = {
+        "id": "scenario:bad-secret-reference",
+        "authority": "platform",
+        "requires": {"capabilities": ["s1.os-baseline"]},
+        "parameter_overrides": {
+            "railiance-infra.bootstrap-host": {
+                "target_hosts": ["railiance01"],
+                "credential_ref": "plaintext-looking-value",
+            }
+        },
+    }
+
+    issues, _ = validator.compose_scenario([declaration], scenario)
+
+    assert any("must be a secret-reference URI" in msg for msg in error_messages(issues))
+
+
 def test_scenario_composition_selects_and_overrides(tmp_path):
     declaration = declaration_from(valid_declaration(), tmp_path)
     scenario = {
@@ -196,3 +271,36 @@ def test_required_parameter_without_override_fails(tmp_path):
     issues, _ = validator.compose_scenario([declaration], scenario)
 
     assert any("required parameter has no default or override" in msg for msg in error_messages(issues))
+
+
+def test_demo_composition_refuses_ambiguous_provider(tmp_path):
+    first = declaration_from(valid_declaration(), tmp_path)
+    second_data = valid_declaration()
+    second_data["metadata"]["id"] = "railiance-infra.bootstrap-host-alternative"
+    second_path = tmp_path / "alternative.yaml"
+    second_path.write_text(yaml.safe_dump(second_data, sort_keys=False), encoding="utf-8")
+    second = validator.Declaration(path=second_path, data=second_data)
+    scenario = {
+        "id": "scenario:ambiguous-provider",
+        "authority": "platform",
+        "requires": {"capabilities": ["s1.os-baseline"]},
+        "parameter_overrides": {},
+    }
+
+    issues, _ = validator.compose_scenario([second, first], scenario)
+
+    assert any("ambiguous providers" in msg for msg in error_messages(issues))
+
+
+def test_demo_composition_refuses_unselected_override(tmp_path):
+    declaration = declaration_from(valid_declaration(), tmp_path)
+    scenario = {
+        "id": "scenario:unselected-override",
+        "authority": "platform",
+        "requires": {"capabilities": ["s1.os-baseline"]},
+        "parameter_overrides": {"owner.unselected": {"target_hosts": ["railiance01"]}},
+    }
+
+    issues, _ = validator.compose_scenario([declaration], scenario)
+
+    assert any("override targets an unselected declaration" in msg for msg in error_messages(issues))

tools/security-scenario-composer/README.md

@@ -0,0 +1,24 @@
+# Security Scenario Composer
+
+The composer is the executable, plan-only implementation of
+`canon/standards/security-scenario-composition_v0.1.md`. It validates
+Playbook Capability Contract v0.1 declarations, selects exact providers,
+applies authority-bound overrides, orders trust transitions, and emits an
+owner-routed JSON handoff.
+
+It never invokes an entry point and always emits
+`execution.permitted: false`.
+
+Compose the checked-in C0 reference:
+
+```bash
+python3 tools/security-scenario-composer/security_scenario_composer.py \
+  --scenario examples/security-scenarios/c0-local-identity.yaml \
+  capabilities/playbooks/net-kingdom.local-identity.yaml
+```
+
+Run tests:
+
+```bash
+python3 -m pytest tools/security-scenario-composer/tests
+```

tools/security-scenario-composer/security_scenario_composer.py

@@ -0,0 +1,445 @@
+#!/usr/bin/env python3
+"""Compose a deterministic, plan-only NetKingdom security scenario."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+TOOL_DIR = Path(__file__).resolve().parent
+CONTRACT_TOOL_DIR = TOOL_DIR.parent / "playbook-capability-contract"
+sys.path.insert(0, str(CONTRACT_TOOL_DIR))
+
+import playbook_contract_validator as contract  # noqa: E402
+
+
+PLAN_API_VERSION = "netkingdom.io/security-scenario-composition/v0.1"
+PLAN_KIND = "SecurityScenarioComposition"
+SCENARIO_ID = re.compile(r"^scenario:[a-z0-9][a-z0-9._:-]*$")
+SCENARIO_KEYS = {
+    "id",
+    "authority",
+    "initial_trust",
+    "requires",
+    "providers",
+    "parameter_overrides",
+}
+
+Issue = contract.Issue
+Declaration = contract.Declaration
+
+
+def error(path: str, message: str) -> Issue:
+    return Issue("ERROR", path, message)
+
+
+def has_errors(issues: list[Issue]) -> bool:
+    return any(item.level == "ERROR" for item in issues)
+
+
+def validate_scenario(scenario: dict[str, Any]) -> list[Issue]:
+    issues: list[Issue] = []
+    unknown_keys = sorted(set(scenario) - SCENARIO_KEYS)
+    if unknown_keys:
+        issues.append(error("scenario", f"unknown fields: {unknown_keys}"))
+
+    scenario_id = scenario.get("id")
+    if not isinstance(scenario_id, str) or not SCENARIO_ID.fullmatch(scenario_id):
+        issues.append(error("scenario.id", "must match 'scenario:<lowercase-id>'"))
+
+    authority = scenario.get("authority")
+    if authority not in contract.SCENARIO_AUTHORITIES:
+        issues.append(error("scenario.authority", f"unknown authority {authority!r}"))
+
+    requires = scenario.get("requires")
+    if not isinstance(requires, dict):
+        issues.append(error("scenario.requires", "must be an object"))
+        required_caps: list[Any] = []
+    else:
+        unknown_requires = sorted(set(requires) - {"capabilities"})
+        if unknown_requires:
+            issues.append(error("scenario.requires", f"unknown fields: {unknown_requires}"))
+        required_caps = requires.get("capabilities", [])
+        if not isinstance(required_caps, list) or not required_caps:
+            issues.append(error("scenario.requires.capabilities", "must be a non-empty list"))
+            required_caps = []
+
+    seen_caps: set[str] = set()
+    for index, cap_id in enumerate(required_caps):
+        path = f"scenario.requires.capabilities[{index}]"
+        if cap_id not in contract.CAPABILITIES:
+            issues.append(error(path, f"unknown capability id {cap_id!r}"))
+        if cap_id in seen_caps:
+            issues.append(error(path, f"duplicate capability id {cap_id!r}"))
+        if isinstance(cap_id, str):
+            seen_caps.add(cap_id)
+
+    initial_trust = scenario.get("initial_trust", [])
+    if not isinstance(initial_trust, list):
+        issues.append(error("scenario.initial_trust", "must be a list"))
+        initial_trust = []
+    seen_trust: set[str] = set()
+    for index, state in enumerate(initial_trust):
+        path = f"scenario.initial_trust[{index}]"
+        if state not in contract.TRUST_STATES:
+            issues.append(error(path, f"unknown trust state {state!r}"))
+        if state in seen_trust:
+            issues.append(error(path, f"duplicate trust state {state!r}"))
+        if isinstance(state, str):
+            seen_trust.add(state)
+
+    providers = scenario.get("providers", {})
+    if not isinstance(providers, dict):
+        issues.append(error("scenario.providers", "must be an object"))
+        providers = {}
+    for cap_id, declaration_id in providers.items():
+        path = f"scenario.providers.{cap_id}"
+        if cap_id not in seen_caps:
+            issues.append(error(path, "provider pin targets an unrequested capability"))
+        if not isinstance(declaration_id, str) or not declaration_id:
+            issues.append(error(path, "provider declaration id must be a non-empty string"))
+
+    overrides = scenario.get("parameter_overrides", {})
+    if not isinstance(overrides, dict):
+        issues.append(error("scenario.parameter_overrides", "must be an object"))
+    else:
+        for declaration_id, values in overrides.items():
+            if not isinstance(declaration_id, str) or not declaration_id:
+                issues.append(error("scenario.parameter_overrides", "declaration ids must be non-empty strings"))
+            if not isinstance(values, dict):
+                issues.append(error(f"scenario.parameter_overrides.{declaration_id}", "must be an object"))
+
+    return issues
+
+
+def validate_declarations(declarations: list[Declaration]) -> list[Issue]:
+    issues: list[Issue] = []
+    ids: dict[str, list[str]] = {}
+    for declaration in declarations:
+        for item in contract.validate_declaration(declaration):
+            issues.append(
+                Issue(
+                    item.level,
+                    f"declarations[{declaration.path}].{item.path}",
+                    item.message,
+                )
+            )
+        ids.setdefault(declaration.id, []).append(str(declaration.path))
+
+    for declaration_id, paths in sorted(ids.items()):
+        if not declaration_id:
+            continue
+        if len(paths) > 1:
+            issues.append(
+                error(
+                    "declarations",
+                    f"duplicate declaration id {declaration_id!r}: {sorted(paths)}",
+                )
+            )
+    return issues
+
+
+def declaration_trust(declaration: Declaration, section: str) -> list[dict[str, Any]]:
+    trust = declaration.data.get("spec", {}).get("trust", {})
+    values = trust.get(section, []) if isinstance(trust, dict) else []
+    return values if isinstance(values, list) else []
+
+
+def trust_states(declaration: Declaration, section: str) -> set[str]:
+    return {
+        str(item.get("state"))
+        for item in declaration_trust(declaration, section)
+        if isinstance(item, dict) and item.get("state")
+    }
+
+
+def stable_digest(scenario: dict[str, Any], selected: dict[str, Declaration]) -> str:
+    payload = {
+        "scenario": scenario,
+        "declarations": {
+            declaration_id: selected[declaration_id].data
+            for declaration_id in sorted(selected)
+        },
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
+
+
+def select_declarations(
+    declarations: list[Declaration],
+    scenario: dict[str, Any],
+) -> tuple[list[Issue], dict[str, Declaration], dict[str, str]]:
+    issues: list[Issue] = []
+    required_caps = scenario["requires"]["capabilities"]
+    pins = scenario.get("providers", {})
+    selected: dict[str, Declaration] = {}
+    capability_providers: dict[str, str] = {}
+
+    for cap_id in required_caps:
+        matches = sorted(
+            (declaration for declaration in declarations if cap_id in declaration.capabilities),
+            key=lambda declaration: declaration.id,
+        )
+        pin = pins.get(cap_id)
+        if pin is not None:
+            pinned = [declaration for declaration in matches if declaration.id == pin]
+            if not pinned:
+                available = [declaration.id for declaration in matches]
+                issues.append(
+                    error(
+                        f"scenario.providers.{cap_id}",
+                        f"pinned declaration {pin!r} does not provide {cap_id!r}; available={available}",
+                    )
+                )
+                continue
+            chosen = pinned[0]
+        elif not matches:
+            issues.append(
+                error(
+                    "scenario.requires.capabilities",
+                    f"no declaration provides {cap_id!r}",
+                )
+            )
+            continue
+        elif len(matches) > 1:
+            issues.append(
+                error(
+                    f"scenario.providers.{cap_id}",
+                    f"ambiguous providers {[item.id for item in matches]}; pin one explicitly",
+                )
+            )
+            continue
+        else:
+            chosen = matches[0]
+
+        selected[chosen.id] = chosen
+        capability_providers[str(cap_id)] = chosen.id
+
+    return issues, selected, capability_providers
+
+
+def effective_parameters(
+    declaration: Declaration,
+    overrides: dict[str, Any],
+    authority: str,
+) -> tuple[list[Issue], dict[str, Any]]:
+    issues: list[Issue] = []
+    params_out: dict[str, Any] = {}
+    for name in sorted(overrides):
+        if name not in declaration.parameters:
+            issues.append(
+                error(
+                    f"scenario.parameter_overrides.{declaration.id}.{name}",
+                    "unknown parameter override",
+                )
+            )
+
+    for name, param in sorted(declaration.parameters.items()):
+        overridden, value = contract.effective_parameter_value(param, overrides, declaration.id)
+        path = f"scenario.parameter_overrides.{declaration.id}.{name}"
+        if param.get("required") is True and value is None:
+            issues.append(error(path, "required parameter has no default or override"))
+        if overridden:
+            issues.extend(contract.validate_override_allowed(param, value, authority, path))
+        params_out[name] = {
+            "value": value,
+            "source": "override" if overridden else "default",
+            "sensitivity": param.get("sensitivity"),
+            "tuning_authority": param.get("tuning_authority"),
+        }
+    return issues, params_out
+
+
+def sequence_declarations(
+    selected: dict[str, Declaration],
+    initial_trust: set[str],
+) -> tuple[list[Issue], list[Declaration], set[str]]:
+    issues: list[Issue] = []
+    established = set(initial_trust)
+    remaining = dict(selected)
+    ordered: list[Declaration] = []
+
+    while remaining:
+        eligible = sorted(
+            (
+                declaration
+                for declaration in remaining.values()
+                if trust_states(declaration, "requires") <= established
+            ),
+            key=lambda declaration: declaration.id,
+        )
+        if not eligible:
+            for declaration_id, declaration in sorted(remaining.items()):
+                missing = sorted(trust_states(declaration, "requires") - established)
+                issues.append(
+                    error(
+                        f"composition.execution_steps.{declaration_id}",
+                        f"cannot sequence declaration; unresolved trust states: {missing}",
+                    )
+                )
+            break
+        chosen = eligible[0]
+        ordered.append(chosen)
+        established.update(trust_states(chosen, "satisfies"))
+        del remaining[chosen.id]
+
+    return issues, ordered, established
+
+
+def compose_scenario(
+    declarations: list[Declaration],
+    scenario: dict[str, Any],
+) -> tuple[list[Issue], dict[str, Any] | None]:
+    issues = validate_scenario(scenario)
+    issues.extend(validate_declarations(declarations))
+    if has_errors(issues):
+        return issues, None
+
+    selection_issues, selected, capability_providers = select_declarations(declarations, scenario)
+    issues.extend(selection_issues)
+    if has_errors(issues):
+        return issues, None
+
+    overrides = scenario.get("parameter_overrides", {})
+    for declaration_id in sorted(overrides):
+        if declaration_id not in selected:
+            issues.append(
+                error(
+                    f"scenario.parameter_overrides.{declaration_id}",
+                    "override targets an unselected declaration",
+                )
+            )
+
+    parameters: dict[str, dict[str, Any]] = {}
+    authority = str(scenario["authority"])
+    for declaration_id, declaration in sorted(selected.items()):
+        param_issues, values = effective_parameters(
+            declaration,
+            overrides.get(declaration_id, {}),
+            authority,
+        )
+        issues.extend(param_issues)
+        parameters[declaration_id] = values
+
+    sequence_issues, ordered, planned_trust = sequence_declarations(
+        selected,
+        set(scenario.get("initial_trust", [])),
+    )
+    issues.extend(sequence_issues)
+    if has_errors(issues):
+        return issues, None
+
+    execution_steps: list[dict[str, Any]] = []
+    responsibilities: list[dict[str, Any]] = []
+    for index, declaration in enumerate(ordered, start=1):
+        metadata = declaration.data["metadata"]
+        spec = declaration.data["spec"]
+        required = declaration_trust(declaration, "requires")
+        satisfied = declaration_trust(declaration, "satisfies")
+        readiness = [
+            {
+                "state": item["state"],
+                "checks": item.get("readiness_checks", []),
+            }
+            for item in satisfied
+        ]
+        execution_steps.append(
+            {
+                "order": index,
+                "declaration_id": declaration.id,
+                "executor_owner": metadata["owner"],
+                "repo": metadata["repo"],
+                "capabilities": sorted(
+                    cap_id
+                    for cap_id, provider in capability_providers.items()
+                    if provider == declaration.id
+                ),
+                "entry_point": spec["playbook"],
+                "parameters": parameters[declaration.id],
+                "requires_trust": required,
+                "satisfies_trust": satisfied,
+                "readiness_obligations": readiness,
+            }
+        )
+        for responsibility in spec["responsibilities"]:
+            responsibilities.append(
+                {
+                    "declaration_id": declaration.id,
+                    **responsibility,
+                }
+            )
+
+    composition = {
+        "apiVersion": PLAN_API_VERSION,
+        "kind": PLAN_KIND,
+        "scenario": scenario["id"],
+        "authority": authority,
+        "composition_digest": stable_digest(scenario, selected),
+        "requested_capabilities": list(scenario["requires"]["capabilities"]),
+        "capability_providers": capability_providers,
+        "initial_trust": sorted(scenario.get("initial_trust", [])),
+        "planned_final_trust": sorted(planned_trust),
+        "execution": {
+            "mode": "plan-only",
+            "permitted": False,
+            "reason": "Composition does not authorize or perform provider execution.",
+        },
+        "execution_steps": execution_steps,
+        "responsibility_map": responsibilities,
+    }
+    return issues, composition
+
+
+def load_declarations(paths: list[str]) -> tuple[list[Declaration], list[Issue]]:
+    declarations: list[Declaration] = []
+    issues: list[Issue] = []
+    for raw_path in paths:
+        path = Path(raw_path)
+        try:
+            data = contract.load_yaml(path)
+        except Exception as exc:
+            issues.append(error(f"declarations[{path}]", f"failed to load: {exc}"))
+            continue
+        declarations.append(Declaration(path=path, data=data))
+    return declarations, issues
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Compose a deterministic, non-executing NetKingdom security scenario."
+    )
+    parser.add_argument("declarations", nargs="+", help="Playbook capability declaration YAML files")
+    parser.add_argument("--scenario", required=True, help="Security scenario YAML file")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    declarations, issues = load_declarations(args.declarations)
+    try:
+        scenario = contract.load_yaml(Path(args.scenario))
+    except Exception as exc:
+        issues.append(error("scenario", f"failed to load {args.scenario}: {exc}"))
+        scenario = {}
+
+    composition: dict[str, Any] | None = None
+    if not has_errors(issues):
+        composition_issues, composition = compose_scenario(declarations, scenario)
+        issues.extend(composition_issues)
+
+    payload = {
+        "ok": not has_errors(issues),
+        "issues": [item.__dict__ for item in issues],
+        "composition": composition,
+    }
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 1 if has_errors(issues) else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tools/security-scenario-composer/tests/test_security_scenario_composer.py

@@ -0,0 +1,266 @@
+import copy
+import importlib.util
+import sys
+from pathlib import Path
+
+import yaml
+
+
+TOOL_DIR = Path(__file__).resolve().parents[1]
+TOOL_PATH = TOOL_DIR / "security_scenario_composer.py"
+sys.path.insert(0, str(TOOL_DIR))
+SPEC = importlib.util.spec_from_file_location("security_scenario_composer", TOOL_PATH)
+composer = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+sys.modules[SPEC.name] = composer
+SPEC.loader.exec_module(composer)
+
+
+def declaration_data(
+    declaration_id="owner.c0",
+    capability="c0.bootstrap-identity",
+    tier="C0",
+    requires=None,
+    satisfies=None,
+):
+    requires = ["bare_host_trust"] if requires is None else requires
+    satisfies = ["bootstrap_identity_trust"] if satisfies is None else satisfies
+    return {
+        "apiVersion": composer.contract.API_VERSION,
+        "kind": composer.contract.KIND,
+        "metadata": {
+            "id": declaration_id,
+            "name": declaration_id,
+            "owner": "owner",
+            "repo": "owner-repo",
+            "domain": "infotech",
+            "contract_version": "0.1",
+        },
+        "spec": {
+            "playbook": {
+                "path": "playbooks/reference.yaml",
+                "type": "reference",
+                "invocation": "make reference",
+                "description": "Reference entry point.",
+            },
+            "capabilities": [
+                {
+                    "id": capability,
+                    "tier": tier,
+                    "resource_kinds": ["identities"],
+                    "description": "Reference capability.",
+                }
+            ],
+            "parameters": [
+                {
+                    "name": "target",
+                    "type": "string",
+                    "required": True,
+                    "default": "reference",
+                    "sensitivity": "operational",
+                    "tuning_authority": "netkingdom_tunable",
+                    "description": "Reference target.",
+                },
+                {
+                    "name": "secure_mode",
+                    "type": "boolean",
+                    "required": False,
+                    "default": True,
+                    "sensitivity": "security_sensitive",
+                    "tuning_authority": "platform_only",
+                    "description": "Security-sensitive reference switch.",
+                },
+            ],
+            "responsibilities": [
+                {
+                    "resource_kind": "identities",
+                    "owner": "owner",
+                    "resources": ["identity:reference"],
+                    "repo_owns": "Reference execution.",
+                    "netkingdom_orchestrates": "Reference selection.",
+                }
+            ],
+            "trust": {
+                "requires": [
+                    {"state": state, "readiness_checks": []}
+                    for state in requires
+                ],
+                "satisfies": [
+                    {
+                        "state": state,
+                        "readiness_checks": [
+                            {
+                                "id": f"{state}-ready",
+                                "description": f"{state} is ready.",
+                                "evidence": "reference evidence",
+                            }
+                        ],
+                    }
+                    for state in satisfies
+                ],
+            },
+            "catalog": {
+                "publish": f"capabilities/playbooks/{declaration_id}.yaml",
+                "maturity": "reference",
+                "consumers": ["netkingdom-security-scenario-composer"],
+            },
+        },
+    }
+
+
+def declaration(tmp_path, data, filename="declaration.yaml"):
+    path = tmp_path / filename
+    path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
+    return composer.Declaration(path=path, data=data)
+
+
+def scenario(capabilities=None):
+    capabilities = ["c0.bootstrap-identity"] if capabilities is None else capabilities
+    return {
+        "id": "scenario:test",
+        "authority": "netkingdom",
+        "initial_trust": ["bare_host_trust"],
+        "requires": {"capabilities": capabilities},
+        "providers": {},
+        "parameter_overrides": {},
+    }
+
+
+def messages(issues):
+    return [item.message for item in issues if item.level == "ERROR"]
+
+
+def test_single_provider_emits_plan_only_handoff(tmp_path):
+    item = declaration(tmp_path, declaration_data())
+
+    issues, plan = composer.compose_scenario([item], scenario())
+
+    assert messages(issues) == []
+    assert plan["execution"] == {
+        "mode": "plan-only",
+        "permitted": False,
+        "reason": "Composition does not authorize or perform provider execution.",
+    }
+    assert plan["execution_steps"][0]["executor_owner"] == "owner"
+    assert plan["execution_steps"][0]["readiness_obligations"][0]["checks"]
+    assert plan["responsibility_map"][0]["declaration_id"] == "owner.c0"
+    assert plan["composition_digest"].startswith("sha256:")
+
+
+def test_ambiguous_provider_requires_pin(tmp_path):
+    first = declaration(tmp_path, declaration_data("owner.c0-a"), "a.yaml")
+    second = declaration(tmp_path, declaration_data("owner.c0-b"), "b.yaml")
+
+    issues, plan = composer.compose_scenario([second, first], scenario())
+
+    assert plan is None
+    assert any("ambiguous providers" in message for message in messages(issues))
+
+
+def test_explicit_provider_pin_resolves_ambiguity(tmp_path):
+    first = declaration(tmp_path, declaration_data("owner.c0-a"), "a.yaml")
+    second = declaration(tmp_path, declaration_data("owner.c0-b"), "b.yaml")
+    request = scenario()
+    request["providers"] = {"c0.bootstrap-identity": "owner.c0-b"}
+
+    issues, plan = composer.compose_scenario([second, first], request)
+
+    assert messages(issues) == []
+    assert plan["capability_providers"] == {"c0.bootstrap-identity": "owner.c0-b"}
+
+
+def test_bad_provider_pin_fails_closed(tmp_path):
+    item = declaration(tmp_path, declaration_data())
+    request = scenario()
+    request["providers"] = {"c0.bootstrap-identity": "owner.missing"}
+
+    issues, plan = composer.compose_scenario([item], request)
+
+    assert plan is None
+    assert any("does not provide" in message for message in messages(issues))
+
+
+def test_override_for_unselected_declaration_fails(tmp_path):
+    item = declaration(tmp_path, declaration_data())
+    request = scenario()
+    request["parameter_overrides"] = {"owner.other": {"target": "wrong"}}
+
+    issues, plan = composer.compose_scenario([item], request)
+
+    assert plan is None
+    assert "override targets an unselected declaration" in messages(issues)
+
+
+def test_tenant_cannot_override_platform_only_parameter(tmp_path):
+    item = declaration(tmp_path, declaration_data())
+    request = scenario()
+    request["authority"] = "tenant"
+    request["parameter_overrides"] = {"owner.c0": {"secure_mode": False}}
+
+    issues, plan = composer.compose_scenario([item], request)
+
+    assert plan is None
+    assert any("tenant authority cannot override" in message for message in messages(issues))
+
+
+def test_trust_dependencies_override_request_order(tmp_path):
+    c0 = declaration(tmp_path, declaration_data(), "c0.yaml")
+    c1_data = declaration_data(
+        "owner.c1",
+        "c1.lightweight-sso",
+        "C1",
+        requires=["bootstrap_identity_trust"],
+        satisfies=["runtime_identity_trust"],
+    )
+    c1 = declaration(tmp_path, c1_data, "c1.yaml")
+    request = scenario(["c1.lightweight-sso", "c0.bootstrap-identity"])
+
+    issues, plan = composer.compose_scenario([c1, c0], request)
+
+    assert messages(issues) == []
+    assert [step["declaration_id"] for step in plan["execution_steps"]] == [
+        "owner.c0",
+        "owner.c1",
+    ]
+    assert "runtime_identity_trust" in plan["planned_final_trust"]
+
+
+def test_unresolved_trust_fails_closed(tmp_path):
+    item = declaration(tmp_path, declaration_data())
+    request = scenario()
+    request["initial_trust"] = []
+
+    issues, plan = composer.compose_scenario([item], request)
+
+    assert plan is None
+    assert any("unresolved trust states" in message for message in messages(issues))
+
+
+def test_duplicate_declaration_ids_fail_closed(tmp_path):
+    data = declaration_data()
+    first = declaration(tmp_path, data, "a.yaml")
+    second = declaration(tmp_path, copy.deepcopy(data), "b.yaml")
+
+    issues, plan = composer.compose_scenario([first, second], scenario())
+
+    assert plan is None
+    assert any("duplicate declaration id" in message for message in messages(issues))
+
+
+def test_checked_in_c0_scenario_composes():
+    repo_root = Path(__file__).resolve().parents[3]
+    declaration_path = repo_root / "capabilities/playbooks/net-kingdom.local-identity.yaml"
+    scenario_path = repo_root / "examples/security-scenarios/c0-local-identity.yaml"
+    item = composer.Declaration(
+        path=declaration_path,
+        data=composer.contract.load_yaml(declaration_path),
+    )
+    request = composer.contract.load_yaml(scenario_path)
+
+    issues, plan = composer.compose_scenario([item], request)
+
+    assert messages(issues) == []
+    assert plan["capability_providers"] == {
+        "c0.bootstrap-identity": "net-kingdom.local-identity"
+    }
+    assert plan["execution_steps"][0]["repo"] == "net-kingdom"