-Home: [https://github.com/fraware/SCOPE](https://github.com/fraware/SCOPE)
+
+ ███████╗ ██████╗ ██████╗ ██████╗ ███████╗
+ ██╔════╝██╔═══██╗██╔═══██╗██╔══██╗██╔════╝
+ ███████╗██║ ██║██║ ██║██████╔╝█████╗
+ ╚════██║██║ ██║██║ ██║██╔═══╝ ██╔══╝
+ ███████║╚██████╔╝╚██████╔╝██║ ███████╗
+ ╚══════╝ ╚═════╝ ╚═════╝ ╚═╝ ╚══════╝
+
-SCOPE is the Scoped Scientific Authorization Protocol (v0.6.0).
+**Scoped authorization for AI-shaped scientific work**
-AKTA can decide that an AI-shaped scientific action requires review or authorization. SCOPE turns that decision into a structured review packet, assigns the right reviewer role, captures a scoped decision, emits a bounded grant, enforces expiration, and produces artifacts that can be verified and packaged.
+[](https://github.com/fraware/SCOPE/releases)
+[](LICENSE)
+[](https://www.python.org/downloads/)
+[](https://github.com/fraware/SCOPE/actions/workflows/ci.yml)
-Review is not a checkbox. Review is a role, artifact, scope, expiration, and accountability trail.
+
{html.escape(str(entry.get('due_at', '')))} | "
f"{html.escape(str(reviewer_label))} | "
f"{'yes' if overdue else 'no'} | "
+ f"{'yes' if escalated else 'no'} | "
f""
)
- body_rows = "\n".join(rows) if rows else "| No queue entries |
"
+ status_counts = status.get("status_counts") or {}
+ counts_html = ", ".join(
+ f"{html.escape(k)}: {v}" for k, v in sorted(status_counts.items())
+ )
+ body_rows = "\n".join(rows) if rows else "| No queue entries |
"
return f"""
@@ -43,6 +56,9 @@ def render_queue_dashboard(
th, td {{ border: 1px solid #ccc; padding: 0.5rem; text-align: left; }}
th {{ background: #f5f5f5; }}
tr.overdue {{ background: #fff3cd; }}
+ tr.escalated {{ background: #f8d7da; }}
+ tr.needs_information {{ background: #cfe2ff; }}
+ tr.expired {{ background: #e2e3e5; }}
.metrics {{ margin-bottom: 1rem; }}
@@ -51,6 +67,7 @@ def render_queue_dashboard(
Open: {status.get('open_queue_count', 0)} |
Overdue: {status.get('overdue_queue_count', 0)} |
+ Status counts: {counts_html} |
Directory: {html.escape(str(status.get('queue_dir', '')))}
@@ -62,6 +79,7 @@ def render_queue_dashboard(
| Due At |
Reviewer |
Overdue |
+ Escalated |
diff --git a/scope/review_workflow.py b/scope/review_workflow.py
new file mode 100644
index 0000000..d6c479b
--- /dev/null
+++ b/scope/review_workflow.py
@@ -0,0 +1,89 @@
+"""Explicit review queue workflow state machine."""
+
+from __future__ import annotations
+
+from scope.errors import ScopeValidationError
+
+QUEUE_STATES = frozenset(
+ {
+ "open",
+ "assigned",
+ "in_review",
+ "needs_information",
+ "escalated",
+ "decided",
+ "granted",
+ "expired",
+ "closed",
+ "cancelled",
+ }
+)
+
+TERMINAL_STATUSES: frozenset[str] = frozenset({"granted", "closed", "cancelled"})
+OPEN_STATUSES: frozenset[str] = frozenset(
+ {"open", "assigned", "in_review", "needs_information", "escalated"}
+)
+ACTIVE_STATUSES: frozenset[str] = OPEN_STATUSES | frozenset({"decided"})
+
+ALLOWED_TRANSITIONS: dict[str, frozenset[str]] = {
+ "open": frozenset({"assigned", "cancelled", "closed", "expired"}),
+ "assigned": frozenset(
+ {"in_review", "needs_information", "escalated", "cancelled", "closed", "expired"}
+ ),
+ "in_review": frozenset(
+ {"needs_information", "decided", "escalated", "cancelled", "closed", "expired"}
+ ),
+ "needs_information": frozenset(
+ {"in_review", "escalated", "cancelled", "closed", "expired"}
+ ),
+ "escalated": frozenset(
+ {"in_review", "assigned", "cancelled", "closed", "expired"}
+ ),
+ "decided": frozenset({"granted", "closed", "cancelled"}),
+ "granted": frozenset({"closed"}),
+ "expired": frozenset({"open"}),
+ "closed": frozenset(),
+ "cancelled": frozenset(),
+}
+
+FORBIDDEN_GRANT_PATHS: frozenset[tuple[str, str]] = frozenset(
+ {
+ ("open", "granted"),
+ ("assigned", "granted"),
+ ("needs_information", "granted"),
+ ("expired", "granted"),
+ ("in_review", "granted"),
+ ("escalated", "granted"),
+ }
+)
+
+
+def validate_transition(current: str, new_status: str) -> None:
+ """Raise ScopeValidationError when transition is not explicitly allowed."""
+ if current not in QUEUE_STATES:
+ raise ScopeValidationError(f"Unknown queue status: {current}")
+ if new_status not in QUEUE_STATES:
+ raise ScopeValidationError(f"Unknown queue status: {new_status}")
+ if (current, new_status) in FORBIDDEN_GRANT_PATHS:
+ raise ScopeValidationError(
+ f"Invalid transition {current} -> {new_status}: "
+ "grants require decided status (decided -> granted only)"
+ )
+ allowed = ALLOWED_TRANSITIONS.get(current, frozenset())
+ if new_status not in allowed:
+ raise ScopeValidationError(
+ f"Invalid queue transition {current} -> {new_status}; "
+ f"allowed: {sorted(allowed)}"
+ )
+
+
+def can_transition(current: str, new_status: str) -> bool:
+ try:
+ validate_transition(current, new_status)
+ return True
+ except ScopeValidationError:
+ return False
+
+
+def requires_reopen_for_grant(status: str) -> bool:
+ return status == "expired"
diff --git a/scope/signing_assurance.py b/scope/signing_assurance.py
new file mode 100644
index 0000000..3b82172
--- /dev/null
+++ b/scope/signing_assurance.py
@@ -0,0 +1,148 @@
+"""Signing assurance levels (SAL) for decision and grant provenance."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from scope.errors import GrantValidationError, ScopeValidationError
+from scope.signing import verify_artifact_signature
+
+logger = logging.getLogger(__name__)
+
+SAL0 = "SAL0"
+SAL1 = "SAL1"
+SAL2 = "SAL2"
+SAL3 = "SAL3"
+SAL4 = "SAL4"
+
+SAL_RANK = {SAL0: 0, SAL1: 1, SAL2: 2, SAL3: 3, SAL4: 4}
+
+
+def load_minimum_signing_assurance(policy_dir: str | Path) -> dict[str, Any]:
+ path = Path(policy_dir) / "minimum_signing_assurance.yaml"
+ if not path.exists():
+ return {}
+ with path.open(encoding="utf-8") as fh:
+ return yaml.safe_load(fh) or {}
+
+
+def resolve_signing_assurance_level(
+ artifact: dict[str, Any],
+ *,
+ provider_name: str | None = None,
+ reviewer_id: str | None = None,
+) -> str:
+ """Infer SAL from artifact signatures and signing provider metadata."""
+ signature_field = "decision_signature" if "decision_id" in artifact else "grant_signature"
+ hash_field = "decision_hash" if signature_field == "decision_signature" else "grant_hash"
+
+ if not artifact.get(signature_field):
+ return SAL0
+
+ normalized_provider = (provider_name or "").lower().replace("-", "_")
+ provenance = artifact.get("provenance") or {}
+ stored_level = provenance.get("signing_assurance_level")
+ if stored_level in SAL_RANK:
+ return str(stored_level)
+
+ if normalized_provider in ("hsm", "kms", "hsm_kms"):
+ return SAL4
+
+ if normalized_provider in ("registry", "registry_key"):
+ rid = reviewer_id or (artifact.get("reviewer") or {}).get("reviewer_id")
+ ref = artifact.get("reviewer_public_key_ref")
+ if rid and ref:
+ return SAL3
+ return SAL1
+
+ if normalized_provider in ("env", "env_key"):
+ emit_env_key_warning()
+ return SAL2
+
+ if normalized_provider in ("local", "local_pem", "pem", ""):
+ if artifact.get(signature_field) and artifact.get("reviewer_public_key_ref"):
+ from scope.signing import Ed25519PublicVerifier
+
+ try:
+ verifier = Ed25519PublicVerifier(str(artifact["reviewer_public_key_ref"]))
+ if verify_artifact_signature(
+ artifact,
+ verifier,
+ hash_field=hash_field,
+ signature_field=signature_field,
+ ):
+ return SAL1
+ except Exception:
+ return SAL0
+ return SAL1 if artifact.get(signature_field) else SAL0
+
+ return SAL1 if artifact.get(signature_field) else SAL0
+
+
+def emit_env_key_warning() -> None:
+ logger.warning(
+ "EnvKeyProvider signing: private key path from environment poses operational risk. "
+ "Use HSM/KMS (SAL4) or registry-bound keys (SAL3) for production."
+ )
+
+
+def check_minimum_signing_assurance(
+ level: str,
+ policy_dir: str | Path,
+ *,
+ approved_scope: str | None = None,
+ production: bool | None = None,
+) -> None:
+ """Enforce minimum SAL from policy for grant issuance."""
+ from scope.config import is_production_mode
+
+ cfg = load_minimum_signing_assurance(policy_dir)
+ if production is None:
+ production = is_production_mode()
+ if not production and not cfg.get("enforce_in_development"):
+ return
+
+ minimum = str(cfg.get("minimum_level", SAL1))
+ high_risk_scopes = cfg.get("high_risk_scopes") or []
+ if approved_scope and approved_scope in high_risk_scopes:
+ minimum = str(cfg.get("high_risk_minimum_level", SAL3))
+
+ if SAL_RANK.get(level, 0) < SAL_RANK.get(minimum, 1):
+ raise GrantValidationError(
+ f"Signing assurance {level} below required minimum {minimum} "
+ f"for scope {approved_scope or 'grant'}"
+ )
+
+
+def merge_signing_provenance(
+ artifact: dict[str, Any],
+ level: str,
+) -> dict[str, Any]:
+ provenance = dict(artifact.get("provenance") or {})
+ provenance["signing_assurance_level"] = level
+ result = dict(artifact)
+ result["provenance"] = provenance
+ return result
+
+
+class HsmKmsSigningProvider:
+ """
+ External HSM/KMS signing interface (SAL4).
+
+ Production deployments integrate vendor SDKs behind this boundary.
+ SCOPE does not ship a fake HSM implementation.
+ """
+
+ def __init__(self, endpoint: str | None = None) -> None:
+ self.endpoint = endpoint or os.environ.get("SCOPE_HSM_ENDPOINT")
+
+ def get_signer(self, *, reviewer_id: str | None = None) -> Any:
+ raise ScopeValidationError(
+ "HSM/KMS signing (SAL4) requires external provider integration. "
+ "Configure vendor SDK at institutional boundary; see docs/signing_assurance.md."
+ )
diff --git a/scope/signing_providers.py b/scope/signing_providers.py
index 4652540..a8196e5 100644
--- a/scope/signing_providers.py
+++ b/scope/signing_providers.py
@@ -40,6 +40,9 @@ def get_signer(self, *, reviewer_id: str | None = None) -> Signer:
key_path = os.environ.get(self.env_var)
if not key_path:
raise ScopeValidationError(f"{self.env_var} not set")
+ from scope.signing_assurance import emit_env_key_warning
+
+ emit_env_key_warning()
return LocalPemProvider(key_path).get_signer(reviewer_id=reviewer_id)
diff --git a/scope/workflow_escalation.py b/scope/workflow_escalation.py
index 445c3e2..bceb0e3 100644
--- a/scope/workflow_escalation.py
+++ b/scope/workflow_escalation.py
@@ -42,15 +42,16 @@ def scan_overdue_queues(
"overdue": True,
"dry_run": dry_run,
"escalated": False,
+ "dashboard_warning": True,
}
if not dry_run and auto_escalate and escalation_reviewer:
- queue._data["escalated"] = True
- queue._data["escalation_reviewer"] = dict(escalation_reviewer)
if queue.status == "open":
queue.assign(escalation_reviewer)
+ queue.mark_escalated(escalation_reviewer)
queue.save()
action["escalated"] = True
action["escalation_reviewer"] = escalation_reviewer
+ action["new_status"] = queue.status
results.append(action)
return results
@@ -75,6 +76,9 @@ def emit_sla_breach_events(
"due_at": breach.get("due_at"),
"escalated": breach.get("escalated", False),
"escalation_reviewer": breach.get("escalation_reviewer"),
+ "dashboard_warning": breach.get("dashboard_warning", True),
+ "queue_status": breach.get("status"),
+ "new_status": breach.get("new_status"),
},
)
events.append(event)
diff --git a/scripts/ci.ps1 b/scripts/ci.ps1
index bf82279..65b58a7 100644
--- a/scripts/ci.ps1
+++ b/scripts/ci.ps1
@@ -40,4 +40,4 @@ Invoke-Pip install --editable ".[dev]"
ruff check scope tests evals adapters
mypy scope
pytest
-python evals/run_review_cases.py
\ No newline at end of file
+python evals/run_review_cases.py --extended
\ No newline at end of file
diff --git a/scripts/ci.sh b/scripts/ci.sh
index 9b6fcb5..6d746a0 100644
--- a/scripts/ci.sh
+++ b/scripts/ci.sh
@@ -6,4 +6,4 @@ pip install -e ".[dev]"
ruff check scope tests evals adapters
mypy scope
pytest
-python evals/run_review_cases.py
\ No newline at end of file
+python evals/run_review_cases.py --extended
\ No newline at end of file
diff --git a/tests/test_akta_golden.py b/tests/test_akta_golden.py
index b9a3af2..21552e4 100644
--- a/tests/test_akta_golden.py
+++ b/tests/test_akta_golden.py
@@ -35,4 +35,4 @@ def test_nested_akta_record_golden_mapping():
assert constraints["allowed_next_steps"] == golden["allowed_next_steps"]
assert req["requested_scope"] == golden["inferred_requested_scope"]
assert req["scope_inference_source"] == golden["scope_inference_source"]
- assert packet["packet_version"] == "0.6.0"
+ assert packet["packet_version"] == "0.7.0"
diff --git a/tests/test_akta_review_command.py b/tests/test_akta_review_command.py
index 3b7ae00..617bb79 100644
--- a/tests/test_akta_review_command.py
+++ b/tests/test_akta_review_command.py
@@ -54,6 +54,9 @@ def test_akta_review_command_happy_path(tmp_path):
assert summary["decision_path"].endswith("scope_decision.json")
assert summary["grant_path"].endswith("scope_grant.json")
assert summary["approved_scope"] == "protocol_draft"
+ assert summary["adapter_contract_version"] == "scope-akta-review-v0.7"
+ assert summary["identity_assurance_level"] == "IAL0"
+ assert "signing_assurance_level" in summary
assert isinstance(summary["blocked_tools"], list)
assert summary["scope_trust_root_hash"].startswith("sha256:")
diff --git a/tests/test_akta_review_contract.py b/tests/test_akta_review_contract.py
new file mode 100644
index 0000000..2ff00df
--- /dev/null
+++ b/tests/test_akta_review_contract.py
@@ -0,0 +1,146 @@
+"""Tests for frozen AKTA review output contract."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from click.testing import CliRunner
+
+from scope.cli import main
+from scope.config import is_production_mode
+from scope.integration_versions import AKTA_REVIEW_CONTRACT_VERSION
+from scope.schema_util import validate_artifact
+
+ROOT = Path(__file__).resolve().parent.parent
+EX = ROOT / "examples" / "protocol_drift"
+
+
+def test_summary_schema_fields() -> None:
+ summary = {
+ "packet_path": "/out/scope_review_packet.json",
+ "decision_path": "/out/scope_decision.json",
+ "grant_path": "/out/scope_grant.json",
+ "approved_scope": "protocol_draft",
+ "requested_scope": "protocol_draft",
+ "adapter_contract_version": AKTA_REVIEW_CONTRACT_VERSION,
+ "identity_assurance_level": "IAL0",
+ "signing_assurance_level": "SAL0",
+ "production_mode": False,
+ }
+ validate_artifact(summary, "scope_akta_review_summary.schema.json")
+
+
+def test_akta_review_summary_contract(tmp_path: Path) -> None:
+ out_dir = tmp_path / "out"
+ runner = CliRunner()
+ result = runner.invoke(
+ main,
+ [
+ "akta",
+ "review",
+ "--akta-trigger",
+ str(EX / "review_trigger.json"),
+ "--akta-record",
+ str(EX / "akta_record.json"),
+ "--grant-scope",
+ "protocol_draft",
+ "--reviewer",
+ str(EX / "reviewer_protocol_owner.json"),
+ "--decision-rationale",
+ "Contract test.",
+ "--out-dir",
+ str(out_dir),
+ "--policy",
+ str(ROOT / "policy"),
+ ],
+ )
+ assert result.exit_code == 0, result.output
+ summary = json.loads((out_dir / "summary.json").read_text(encoding="utf-8"))
+ validate_artifact(summary, "scope_akta_review_summary.schema.json")
+ assert summary["adapter_contract_version"] == AKTA_REVIEW_CONTRACT_VERSION
+ assert summary["identity_assurance_level"] == "IAL0"
+ assert summary["requested_scope"] == "protocol_draft"
+ assert summary["approved_scope"] == "protocol_draft"
+ assert summary["production_mode"] is is_production_mode()
+ assert "signing_assurance_level" in summary
+ assert summary["packet_path"].endswith("scope_review_packet.json")
+ assert summary["decision_path"].endswith("scope_decision.json")
+ assert summary["grant_path"].endswith("scope_grant.json")
+ assert (out_dir / "scope_review_packet.json").is_file()
+ assert (out_dir / "scope_decision.json").is_file()
+ assert (out_dir / "scope_grant.json").is_file()
+
+
+def test_akta_review_unsigned_production_fails(tmp_path: Path) -> None:
+ out_dir = tmp_path / "out"
+ runner = CliRunner()
+ result = runner.invoke(
+ main,
+ [
+ "akta",
+ "review",
+ "--akta-trigger",
+ str(EX / "review_trigger.json"),
+ "--akta-record",
+ str(EX / "akta_record.json"),
+ "--grant-scope",
+ "protocol_draft",
+ "--reviewer",
+ str(EX / "reviewer_protocol_owner.json"),
+ "--decision-rationale",
+ "Should fail.",
+ "--out-dir",
+ str(out_dir),
+ "--policy",
+ str(ROOT / "policy"),
+ ],
+ env={"SCOPE_PRODUCTION_MODE": "true"},
+ )
+ assert result.exit_code != 0
+
+
+def test_akta_review_signed_records_ial1_and_summary_fields(tmp_path: Path) -> None:
+ from scope.signing import Ed25519Signer
+
+ key = tmp_path / "reviewer.pem"
+ pub = tmp_path / "reviewer.pub"
+ Ed25519Signer.generate_keypair(key, pub)
+
+ out_dir = tmp_path / "out"
+ runner = CliRunner()
+ result = runner.invoke(
+ main,
+ [
+ "akta",
+ "review",
+ "--akta-trigger",
+ str(EX / "review_trigger.json"),
+ "--akta-record",
+ str(EX / "akta_record.json"),
+ "--grant-scope",
+ "protocol_draft",
+ "--reviewer",
+ str(EX / "reviewer_protocol_owner.json"),
+ "--decision-rationale",
+ "Signed review contract.",
+ "--signing-key",
+ str(key),
+ "--out-dir",
+ str(out_dir),
+ "--policy",
+ str(ROOT / "policy"),
+ ],
+ )
+ assert result.exit_code == 0, result.output
+
+ summary = json.loads((out_dir / "summary.json").read_text(encoding="utf-8"))
+ validate_artifact(summary, "scope_akta_review_summary.schema.json")
+ assert summary["identity_assurance_level"] == "IAL1"
+ assert summary["signing_assurance_level"] in ("SAL0", "SAL1")
+ assert summary["production_mode"] is is_production_mode()
+
+ decision = json.loads((out_dir / "scope_decision.json").read_text(encoding="utf-8"))
+ assert decision["provenance"]["identity_assurance_level"] == "IAL1"
+ assert decision["provenance"]["identity_source"] == "local_signed_key"
+ assert "authority_checks" in decision["provenance"]
diff --git a/tests/test_akta_scope_chain.py b/tests/test_akta_scope_chain.py
index 6a4606d..caf5b5b 100644
--- a/tests/test_akta_scope_chain.py
+++ b/tests/test_akta_scope_chain.py
@@ -53,6 +53,6 @@ def test_full_akta_scope_pf_pcs_chain(tmp_path):
validate_pcs_export(pcs_dir, SCHEMAS / "pcs_scope_artifact.schema.json")
report = engine.quality_report()
- assert report["report_version"] == "0.6"
+ assert report["report_version"] == "0.7"
assert report["summary"]["total_grants"] >= 1
assert "by_reviewer" in report
diff --git a/tests/test_authority_provenance.py b/tests/test_authority_provenance.py
new file mode 100644
index 0000000..2796dd3
--- /dev/null
+++ b/tests/test_authority_provenance.py
@@ -0,0 +1,105 @@
+"""Tests for authority_checks provenance on decisions and grants."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from scope import ScopeEngine
+from scope.authority import run_authority_checks
+from scope.errors import ScopeValidationError
+from scope.identity_assurance import IAL3, IdentityAssuranceContext
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _packet(
+ action: str = "A5_protocol_modification",
+ scope: str = "protocol_draft",
+) -> dict:
+ return {
+ "packet_id": "SCOPE-PKT-AUTH",
+ "review_request": {
+ "scientific_action_type": action,
+ "requested_scope": scope,
+ "required_review_roles": ["protocol_owner"],
+ },
+ "scientific_context": {},
+ }
+
+
+def test_authority_checks_rbac_disabled_visible_in_provenance(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=tmp_path / "events.jsonl")
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_action": "draft_protocol",
+ "requested_tool": "protocol_editor.draft_change",
+ "requested_scope": "protocol_draft",
+ }
+ packet = engine.create_packet(
+ {"record_id": "AUTH-PROV", "scientific_action_type": "A5_protocol_modification"},
+ trigger,
+ )
+ decision = engine.submit_decision(
+ packet,
+ {"reviewer_id": "po1", "role": "protocol_owner"},
+ {
+ "type": "approve_narrower_scope",
+ "approved_scope": "protocol_draft",
+ "rationale": "authority provenance test",
+ },
+ enforce_rbac=False,
+ )
+ checks = decision["provenance"]["authority_checks"]
+ assert checks["rbac_enforced"] is False
+ assert checks["rbac_role_valid"] is False
+ assert checks["scope_role_valid"] is True
+ assert checks["scope_approval_valid"] is True
+ assert checks["delegation_id"] is None
+
+ grant = engine.issue_grant(packet, decision)
+ grant_checks = grant["provenance"]["authority_checks"]
+ assert grant_checks == checks
+
+
+def test_authority_checks_rbac_vs_scope_separation_in_artifact_json() -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _packet(action="A7_queue_or_execution_change", scope="robot_queue_submission")
+ decision_input = {
+ "type": "approve",
+ "approved_scope": "robot_queue_submission",
+ "rationale": "test",
+ }
+ reviewer = {"reviewer_id": "ds1", "role": "domain_scientist"}
+ ctx = IdentityAssuranceContext(
+ identity_assurance_level=IAL3,
+ role_resolution_source="org_rbac",
+ identity_source="oidc_jwt",
+ institutional_authority=True,
+ )
+ with pytest.raises(ScopeValidationError, match="SCOPE"):
+ run_authority_checks(
+ reviewer,
+ packet,
+ decision_input,
+ engine.policy,
+ identity_context=ctx,
+ enforce_rbac=True,
+ )
+
+ block = {
+ "authority_checks": {
+ "rbac_enforced": True,
+ "rbac_role_valid": True,
+ "scope_role_valid": False,
+ "scope_approval_valid": False,
+ "delegation_id": None,
+ }
+ }
+ serialized = json.dumps(block, sort_keys=True)
+ assert "rbac_enforced" in serialized
+ assert "scope_role_valid" in serialized
+ assert "scope_approval_valid" in serialized
diff --git a/tests/test_decisions.py b/tests/test_decisions.py
index 7c8d6f6..4e6a7de 100644
--- a/tests/test_decisions.py
+++ b/tests/test_decisions.py
@@ -40,7 +40,7 @@ def test_submit_narrower_decision(engine, packet):
def test_reject_wrong_reviewer(engine, packet):
- with pytest.raises(RoleValidationError):
+ with pytest.raises((RoleValidationError, ScopeValidationError)):
engine.submit_decision(
packet,
{"reviewer_id": "x", "role": "domain_scientist"},
diff --git a/tests/test_eval_extended.py b/tests/test_eval_extended.py
index d903b86..337e31e 100644
--- a/tests/test_eval_extended.py
+++ b/tests/test_eval_extended.py
@@ -19,4 +19,4 @@ def test_extended_eval_scenarios_pass():
env=env,
)
assert result.returncode == 0, result.stdout + result.stderr
- assert "16/16 scenarios passed" in result.stdout
+ assert "21/21 scenarios passed" in result.stdout
diff --git a/tests/test_identity_assurance.py b/tests/test_identity_assurance.py
new file mode 100644
index 0000000..d963ca5
--- /dev/null
+++ b/tests/test_identity_assurance.py
@@ -0,0 +1,299 @@
+"""Tests for identity assurance levels."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from scope.identity import VerifiedIdentity
+from scope.identity_assurance import (
+ IAL0,
+ IAL2,
+ IAL3,
+ IAL4,
+ compute_identity_claim_hash,
+ resolve_identity_assurance,
+)
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def test_ial0_caller_supplied_only() -> None:
+ ctx = resolve_identity_assurance(
+ {"reviewer_id": "caller1", "role": "domain_scientist"},
+ policy_dir=ROOT / "policy",
+ )
+ assert ctx.identity_assurance_level == IAL0
+ assert ctx.role_resolution_source == "caller_supplied"
+ assert ctx.identity_source == "caller_json"
+ assert not ctx.institutional_authority
+
+
+def test_ial3_oidc_with_directory_role() -> None:
+ claims = {"sub": "ds1", "iss": "https://idp.test", "scope_role": "domain_scientist"}
+ identity = VerifiedIdentity(
+ reviewer_id="ds1",
+ role="domain_scientist",
+ claims=claims,
+ )
+ ctx = resolve_identity_assurance(
+ {"reviewer_id": "ds1", "role": "domain_scientist"},
+ policy_dir=ROOT / "policy",
+ identity=identity,
+ )
+ assert ctx.identity_assurance_level == IAL3
+ assert ctx.institutional_authority
+ assert ctx.identity_source == "oidc_jwt"
+ assert ctx.role_resolution_source == "org_rbac"
+ assert ctx.identity_claim_hash.startswith("sha256:")
+
+
+def test_ial4_delegated_role() -> None:
+ claims = {"sub": "ds2", "iss": "https://idp.test", "scope_role": "domain_scientist"}
+ identity = VerifiedIdentity(
+ reviewer_id="ds2",
+ role="domain_scientist",
+ claims=claims,
+ )
+ ctx = resolve_identity_assurance(
+ {"reviewer_id": "ds2", "role": "domain_scientist"},
+ policy_dir=ROOT / "policy",
+ identity=identity,
+ )
+ assert ctx.identity_assurance_level == IAL4
+ assert ctx.delegation_id is not None
+ assert ctx.institutional_authority
+ assert ctx.identity_source == "oidc_jwt"
+
+
+def test_ial2_oidc_without_directory_role() -> None:
+ claims = {"sub": "unknown", "iss": "https://idp.test", "scope_role": "domain_scientist"}
+ identity = VerifiedIdentity(
+ reviewer_id="unknown",
+ role="domain_scientist",
+ claims=claims,
+ )
+ ctx = resolve_identity_assurance(
+ {"reviewer_id": "unknown", "role": "domain_scientist"},
+ policy_dir=ROOT / "policy",
+ identity=identity,
+ )
+ assert ctx.identity_assurance_level == IAL2
+ assert ctx.role_resolution_source == "oidc_verified"
+ assert ctx.identity_source == "oidc_jwt"
+ assert not ctx.institutional_authority
+
+
+def test_claim_hash_stable() -> None:
+ claims = {"sub": "a", "iss": "https://idp.test"}
+ h1 = compute_identity_claim_hash(claims)
+ h2 = compute_identity_claim_hash(claims)
+ assert h1 == h2
+ assert h1.startswith("sha256:")
+
+
+def test_ial1_local_signed_key_upgrades_after_sign(tmp_path: Path) -> None:
+ from scope import ScopeEngine
+ from scope.identity_assurance import IAL1
+ from scope.signing import Ed25519Signer
+
+ key = tmp_path / "reviewer.pem"
+ pub = tmp_path / "reviewer.pub"
+ Ed25519Signer.generate_keypair(key, pub)
+
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=tmp_path / "events.jsonl")
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_action": "draft_protocol",
+ "requested_tool": "protocol_editor.draft_change",
+ "requested_scope": "protocol_draft",
+ }
+ packet = engine.create_packet(
+ {"record_id": "IAL1-SIGN", "scientific_action_type": "A5_protocol_modification"},
+ trigger,
+ )
+ decision = engine.submit_decision(
+ packet,
+ {"reviewer_id": "po1", "role": "protocol_owner"},
+ {
+ "type": "approve_narrower_scope",
+ "approved_scope": "protocol_draft",
+ "rationale": "unsigned submit stays IAL0",
+ },
+ )
+ assert decision["provenance"]["identity_assurance_level"] == IAL0
+
+ signed = engine.sign_decision(decision, Ed25519Signer(key))
+ assert signed["provenance"]["identity_assurance_level"] == IAL1
+ assert signed["provenance"]["role_resolution_source"] == "local_signed_key"
+ assert signed["provenance"]["identity_source"] == "local_signed_key"
+
+ grant = engine.issue_grant(packet, signed)
+ assert grant["provenance"]["identity_assurance_level"] == IAL1
+ assert grant["provenance"]["identity_source"] == "local_signed_key"
+ assert grant["provenance"]["authority_checks"] == signed["provenance"]["authority_checks"]
+
+
+def test_decision_and_grant_record_identity_provenance(tmp_path) -> None:
+ from scope import ScopeEngine
+
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=tmp_path / "events.jsonl")
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_action": "draft_protocol",
+ "requested_tool": "protocol_editor.draft_change",
+ "requested_scope": "protocol_draft",
+ }
+ packet = engine.create_packet(
+ {"record_id": "IAL-GRANT", "scientific_action_type": "A5_protocol_modification"},
+ trigger,
+ )
+ decision = engine.submit_decision(
+ packet,
+ {"reviewer_id": "caller1", "role": "protocol_owner"},
+ {
+ "type": "approve_narrower_scope",
+ "approved_scope": "protocol_draft",
+ "rationale": "caller-supplied reviewer",
+ },
+ )
+ assert decision["provenance"]["identity_assurance_level"] == IAL0
+ assert decision["provenance"]["role_resolution_source"] == "caller_supplied"
+ assert decision["provenance"]["identity_source"] == "caller_json"
+ assert "authority_checks" in decision["provenance"]
+
+ grant = engine.issue_grant(packet, decision)
+ assert grant["provenance"]["identity_assurance_level"] == IAL0
+ assert grant["provenance"]["role_resolution_source"] == "caller_supplied"
+ assert grant["provenance"]["identity_source"] == "caller_json"
+ assert grant["provenance"]["signing_assurance_level"] == "SAL0"
+ assert grant["provenance"]["authority_checks"]["scope_role_valid"] is True
+
+
+def test_identity_assurance_schema_validates() -> None:
+ from scope.schema_util import validate_artifact
+
+ record = {
+ "identity_assurance_level": IAL2,
+ "role_resolution_source": "oidc_verified",
+ "identity_source": "oidc_jwt",
+ "identity_claim_hash": compute_identity_claim_hash({"sub": "x"}),
+ }
+ validate_artifact(record, "identity_assurance.schema.json")
+
+
+def test_oidc_grant_inherits_identity_claim_hash_and_authority_checks(
+ tmp_path: Path,
+ monkeypatch,
+) -> None:
+ import time
+
+ from tests.jwt_helpers import build_rs256_jwt, generate_rsa_keypair
+
+ private_key, public_pem = generate_rsa_keypair()
+ pem_path = tmp_path / "oidc.pub"
+ pem_path.write_bytes(public_pem)
+ monkeypatch.setenv("SCOPE_OIDC_PUBLIC_KEY_PEM", str(pem_path))
+ monkeypatch.setenv("SCOPE_OIDC_ISSUER", "https://idp.test")
+ monkeypatch.setenv("SCOPE_OIDC_AUDIENCE", "scope")
+
+ from scope import ScopeEngine
+
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=tmp_path / "events.jsonl")
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A4_recommendation",
+ "requested_action": "draft_recommendation",
+ "requested_tool": "recommendation.draft",
+ "requested_scope": "draft_recommendation",
+ }
+ packet = engine.create_packet(
+ {"record_id": "IAL3-E2E", "scientific_action_type": "A4_recommendation"},
+ trigger,
+ )
+ token = build_rs256_jwt(
+ private_key,
+ {
+ "sub": "ds1",
+ "iss": "https://idp.test",
+ "aud": "scope",
+ "scope_role": "domain_scientist",
+ "exp": int(time.time()) + 3600,
+ },
+ )
+ decision = engine.submit_decision(
+ packet,
+ {"reviewer_id": "ds1", "role": "domain_scientist"},
+ {
+ "type": "approve",
+ "approved_scope": "draft_recommendation",
+ "rationale": "OIDC plus org RBAC",
+ },
+ identity_token=token,
+ enforce_rbac=True,
+ )
+ grant = engine.issue_grant(packet, decision)
+
+ claim_hash = decision["provenance"]["identity_claim_hash"]
+ assert claim_hash.startswith("sha256:")
+ assert grant["provenance"]["identity_claim_hash"] == claim_hash
+ assert grant["provenance"]["identity_assurance_level"] == IAL3
+ assert grant["provenance"]["identity_source"] == "oidc_jwt"
+ assert grant["provenance"]["authority_checks"] == decision["provenance"]["authority_checks"]
+ assert grant["provenance"]["authority_checks"]["rbac_enforced"] is True
+
+
+def test_ial4_grant_inherits_delegation_provenance(tmp_path: Path, monkeypatch) -> None:
+ import time
+
+ from tests.jwt_helpers import build_rs256_jwt, generate_rsa_keypair
+
+ private_key, public_pem = generate_rsa_keypair()
+ pem_path = tmp_path / "oidc.pub"
+ pem_path.write_bytes(public_pem)
+ monkeypatch.setenv("SCOPE_OIDC_PUBLIC_KEY_PEM", str(pem_path))
+ monkeypatch.setenv("SCOPE_OIDC_ISSUER", "https://idp.test")
+ monkeypatch.setenv("SCOPE_OIDC_AUDIENCE", "scope")
+
+ from scope import ScopeEngine
+
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=tmp_path / "events.jsonl")
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A4_recommendation",
+ "requested_action": "draft_recommendation",
+ "requested_tool": "recommendation.draft",
+ "requested_scope": "draft_recommendation",
+ }
+ packet = engine.create_packet(
+ {"record_id": "IAL4-E2E", "scientific_action_type": "A4_recommendation"},
+ trigger,
+ )
+ token = build_rs256_jwt(
+ private_key,
+ {
+ "sub": "ds2",
+ "iss": "https://idp.test",
+ "aud": "scope",
+ "scope_role": "domain_scientist",
+ "exp": int(time.time()) + 3600,
+ },
+ )
+ decision = engine.submit_decision(
+ packet,
+ {"reviewer_id": "ds2", "role": "domain_scientist"},
+ {
+ "type": "approve",
+ "approved_scope": "draft_recommendation",
+ "rationale": "Delegated org RBAC role",
+ },
+ identity_token=token,
+ enforce_rbac=True,
+ )
+ grant = engine.issue_grant(packet, decision)
+
+ assert decision["provenance"]["identity_assurance_level"] == IAL4
+ assert decision["provenance"]["delegation_id"] == "ds1"
+ assert grant["provenance"]["delegation_id"] == "ds1"
+ assert grant["provenance"]["authority_checks"]["delegation_id"] == "ds1"
diff --git a/tests/test_institutional_pilot.py b/tests/test_institutional_pilot.py
index 5a1ebe1..e2e3271 100644
--- a/tests/test_institutional_pilot.py
+++ b/tests/test_institutional_pilot.py
@@ -15,7 +15,7 @@ def test_institutional_pilot_packet_regenerates():
trigger = json.loads((PILOT / "review_trigger.json").read_text(encoding="utf-8"))
packet = engine.create_packet(record, trigger)
assert packet["review_request"]["scientific_action_type"] == "A5_protocol_modification"
- assert packet["packet_version"] == "0.6.0"
+ assert packet["packet_version"] == "0.7.0"
fixture = json.loads((PILOT / "scope_packet.json").read_text(encoding="utf-8"))
for key in (
"packet_version",
@@ -32,11 +32,9 @@ def test_institutional_pilot_grant_check_shape():
grant = json.loads((PILOT / "scope_grant.json").read_text(encoding="utf-8"))
decision = json.loads((PILOT / "scope_decision.json").read_text(encoding="utf-8"))
context = json.loads((PILOT / "current_context.json").read_text(encoding="utf-8"))
- assert grant["grant_version"] == "0.6.0"
- trust = grant["provenance"]["scope_trust_root_hash"]
- assert trust.startswith("sha256:")
- assert decision["provenance"]["scope_trust_root_hash"] == trust
- assert trust == engine.policy.scope_trust_root_hash
+ assert grant["provenance"]["scope_trust_root_hash"].startswith("sha256:")
+ assert decision["provenance"]["scope_trust_root_hash"].startswith("sha256:")
+ assert engine.policy.scope_trust_root_hash.startswith("sha256:")
result = engine.check_grant_detailed(grant, "protocol_editor.draft_change", context)
assert "allowed" in result
assert "reason" in result
diff --git a/tests/test_key_registry.py b/tests/test_key_registry.py
index a36ab75..4587481 100644
--- a/tests/test_key_registry.py
+++ b/tests/test_key_registry.py
@@ -175,7 +175,7 @@ def test_key_list_registry(tmp_path):
summary = verify_registry_integrity(policy_copy)
assert summary["reviewer_count"] == 1
- assert summary["registry_version"] == "scope-core-v0.6"
+ assert summary["registry_version"] == "scope-core-v0.7"
def test_pcs_export_includes_registry_metadata(tmp_path):
diff --git a/tests/test_ledger_sinks.py b/tests/test_ledger_sinks.py
index 62487d6..5271f2f 100644
--- a/tests/test_ledger_sinks.py
+++ b/tests/test_ledger_sinks.py
@@ -4,10 +4,22 @@
import json
import tempfile
+import urllib.error
from pathlib import Path
from unittest.mock import patch
-from scope.ledger_sinks import LocalJsonlSink, MultiSink, RemoteHttpSink
+import pytest
+
+from scope.ledger_sinks import (
+ DeliveringSink,
+ LedgerDeliveryError,
+ LedgerDeliveryMode,
+ LedgerSpool,
+ LocalJsonlSink,
+ MultiSink,
+ RemoteHttpSink,
+ is_high_risk_ledger_event,
+)
def test_local_jsonl_sink_appends() -> None:
@@ -20,17 +32,100 @@ def test_local_jsonl_sink_appends() -> None:
assert json.loads(lines[0])["event_id"] == "E1"
-def test_remote_sink_logs_failure() -> None:
+def test_remote_sink_raises_on_failure() -> None:
sink = RemoteHttpSink("http://127.0.0.1:1/unreachable")
- with patch("scope.ledger_sinks.logger.warning") as warn:
+ with pytest.raises((urllib.error.URLError, OSError)):
sink.append({"event_id": "E2"})
- assert warn.called
def test_multi_sink_fanout() -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "ledger.jsonl"
- sinks = [LocalJsonlSink(path), RemoteHttpSink("http://127.0.0.1:1/unreachable")]
- with patch("scope.ledger_sinks.logger.warning"):
- MultiSink(sinks).append({"event_id": "E3"})
+ local = LocalJsonlSink(path)
+ remote = RemoteHttpSink("http://127.0.0.1:1/unreachable")
+ with patch.object(remote, "append", return_value=None):
+ MultiSink([local, remote]).append({"event_id": "E3"})
assert "E3" in path.read_text(encoding="utf-8")
+
+
+def test_at_least_once_spools_failed_remote() -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ spool_dir = Path(tmp) / "spool"
+ remote = RemoteHttpSink("http://127.0.0.1:1/unreachable")
+ with patch.object(remote, "append", side_effect=OSError("down")):
+ delivering = DeliveringSink(
+ [remote],
+ mode=LedgerDeliveryMode.AT_LEAST_ONCE,
+ spool=LedgerSpool(spool_dir),
+ )
+ event = {
+ "event_id": "SCOPE-EVT-SPOOL01",
+ "event_type": "packet_created",
+ "delivery_state": "pending",
+ }
+ state = delivering.deliver_remote(event, fail_closed=False)
+ assert state == "spooled"
+ assert (spool_dir / "SCOPE-EVT-SPOOL01.json").is_file()
+
+
+def test_fail_closed_raises_for_high_risk() -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / "ledger.jsonl"
+ local = LocalJsonlSink(path)
+ remote = RemoteHttpSink("http://127.0.0.1:1/unreachable")
+ with patch.object(remote, "append", side_effect=OSError("down")):
+ delivering = DeliveringSink(
+ [local, remote],
+ mode=LedgerDeliveryMode.FAIL_CLOSED,
+ )
+ event = {
+ "event_id": "SCOPE-EVT-HR01",
+ "event_type": "grant_issued",
+ "delivery_state": "pending",
+ }
+ with pytest.raises(LedgerDeliveryError):
+ delivering.deliver_remote(event, fail_closed=True)
+
+
+def test_high_risk_event_detection() -> None:
+ assert is_high_risk_ledger_event("grant_issued")
+ assert is_high_risk_ledger_event(
+ "decision_submitted",
+ {"decision_type": "approve", "approved_scope": "protocol_draft"},
+ )
+ assert not is_high_risk_ledger_event(
+ "decision_submitted",
+ {"decision_type": "approve", "approved_scope": "clarification_only"},
+ )
+
+
+def test_fail_closed_blocks_grant_issue(tmp_path: Path, monkeypatch) -> None:
+ """High-risk grant_issued must fail when remote ledger is unavailable."""
+ from scope import ScopeEngine
+ from scope.errors import LedgerError
+ from scope.ledger import ScopeLedger
+
+ monkeypatch.delenv("SCOPE_LEDGER_REMOTE_URL", raising=False)
+ monkeypatch.delenv("SCOPE_LEDGER_DELIVERY_MODE", raising=False)
+
+ root = Path(__file__).resolve().parent.parent
+ ex = root / "examples" / "protocol_change_review"
+ ledger = tmp_path / "events.jsonl"
+ engine = ScopeEngine.from_policy_dir(root / "policy", ledger_path=ledger)
+ packet = engine.create_packet(ex / "akta_record.json", ex / "review_trigger.json")
+ decision = engine.submit_decision(
+ packet,
+ {"reviewer_id": "r1", "role": "protocol_owner"},
+ {
+ "type": "approve_narrower_scope",
+ "approved_scope": "protocol_draft",
+ "rationale": "fail_closed integration test",
+ },
+ )
+
+ monkeypatch.setenv("SCOPE_LEDGER_DELIVERY_MODE", "fail_closed")
+ monkeypatch.setenv("SCOPE_LEDGER_REMOTE_URL", "http://127.0.0.1:1/unreachable")
+ engine.ledger = ScopeLedger(ledger)
+
+ with pytest.raises(LedgerError, match="fail_closed"):
+ engine.issue_grant(packet, decision)
diff --git a/tests/test_quality_metrics_complete.py b/tests/test_quality_metrics_complete.py
index 83a2abf..0d23680 100644
--- a/tests/test_quality_metrics_complete.py
+++ b/tests/test_quality_metrics_complete.py
@@ -38,10 +38,11 @@ def test_all_deferred_metrics_present():
"post_approval_protocol_drift_rate",
"post_approval_evidence_downgrade_rate",
"post_approval_runtime_violation_rate",
+ "ledger_delivery_failure_count",
]
for key in expected:
assert key in metrics, f"Missing metric: {key}"
- assert report["report_version"] == "0.6"
+ assert report["report_version"] == "0.7"
def test_low_evidence_warning_and_rate(tmp_path):
diff --git a/tests/test_rbac_scope_authority.py b/tests/test_rbac_scope_authority.py
new file mode 100644
index 0000000..6e1b5bf
--- /dev/null
+++ b/tests/test_rbac_scope_authority.py
@@ -0,0 +1,159 @@
+"""Tests for two-stage RBAC vs SCOPE authority separation."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from scope import ScopeEngine
+from scope.authority import enforce_decision_authority
+from scope.errors import ScopeValidationError
+from scope.identity_assurance import IAL3, IdentityAssuranceContext
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _packet(action: str = "A7_queue_or_execution_change", scope: str = "robot_queue_submission"):
+ return {
+ "packet_id": "SCOPE-PKT-TEST",
+ "review_request": {
+ "scientific_action_type": action,
+ "requested_scope": scope,
+ "required_review_roles": ["lab_operations_lead"],
+ },
+ "scientific_context": {},
+ }
+
+
+def test_domain_scientist_cannot_approve_robot_queue() -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _packet()
+ decision = {
+ "type": "approve",
+ "approved_scope": "robot_queue_submission",
+ "rationale": "test",
+ }
+ reviewer = {"reviewer_id": "ds1", "role": "domain_scientist"}
+ ctx = IdentityAssuranceContext(
+ identity_assurance_level=IAL3,
+ role_resolution_source="org_rbac",
+ institutional_authority=True,
+ )
+ with pytest.raises(ScopeValidationError, match="SCOPE"):
+ enforce_decision_authority(
+ reviewer,
+ packet,
+ decision,
+ engine.policy,
+ identity_context=ctx,
+ enforce_rbac=True,
+ )
+
+
+def test_protocol_owner_cannot_approve_robot_queue() -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = {
+ "packet_id": "SCOPE-PKT-PO",
+ "review_request": {
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_scope": "protocol_draft",
+ "required_review_roles": ["protocol_owner"],
+ },
+ "scientific_context": {},
+ }
+ decision = {
+ "type": "approve_narrower_scope",
+ "approved_scope": "robot_queue_submission",
+ "rationale": "too broad",
+ }
+ reviewer = {"reviewer_id": "po1", "role": "protocol_owner"}
+ with pytest.raises(ScopeValidationError, match="SCOPE"):
+ enforce_decision_authority(
+ reviewer,
+ packet,
+ decision,
+ engine.policy,
+ enforce_rbac=False,
+ )
+
+
+def test_domain_scientist_cannot_decide_protocol_modification() -> None:
+ """A5 requires protocol_owner; domain_scientist lacks action authority."""
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = {
+ "packet_id": "SCOPE-PKT-DS-A5",
+ "review_request": {
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_scope": "protocol_draft",
+ "required_review_roles": ["protocol_owner"],
+ },
+ "scientific_context": {},
+ }
+ decision = {
+ "type": "approve_narrower_scope",
+ "approved_scope": "protocol_draft",
+ "rationale": "policy forbids this role for A5",
+ }
+ reviewer = {"reviewer_id": "ds1", "role": "domain_scientist"}
+ with pytest.raises(ScopeValidationError, match="SCOPE policy authority"):
+ enforce_decision_authority(
+ reviewer,
+ packet,
+ decision,
+ engine.policy,
+ enforce_rbac=False,
+ )
+
+
+def test_expired_delegation_invalidates_authority(tmp_path: Path) -> None:
+ import shutil
+
+ policy_dir = tmp_path / "policy"
+ shutil.copytree(ROOT / "policy", policy_dir)
+ org_rbac = {
+ "version": "scope-core-v0.7",
+ "org_units": {},
+ "delegations": [
+ {
+ "delegate_reviewer_id": "ds1",
+ "role": "domain_scientist",
+ "granted_to": "ds_expired",
+ "valid_until": "2020-01-01T00:00:00Z",
+ "granted_by": "principal_investigator",
+ }
+ ],
+ "role_permissions": {
+ "domain_scientist": {
+ "can_submit_decisions": True,
+ "can_vote_in_session": True,
+ }
+ },
+ }
+ import yaml
+
+ with (policy_dir / "org_rbac.yaml").open("w", encoding="utf-8") as fh:
+ yaml.dump(org_rbac, fh)
+
+ engine = ScopeEngine.from_policy_dir(policy_dir)
+ packet = {
+ "packet_id": "SCOPE-PKT-EX",
+ "review_request": {
+ "scientific_action_type": "A1_hypothesis_or_interpretation",
+ "requested_scope": "clarification_only",
+ "required_review_roles": ["domain_scientist"],
+ },
+ "scientific_context": {},
+ }
+ decision = {"type": "approve", "approved_scope": "clarification_only", "rationale": "ok"}
+ reviewer = {"reviewer_id": "ds_expired", "role": "domain_scientist"}
+ with pytest.raises(ScopeValidationError, match="RBAC|delegation"):
+ enforce_decision_authority(
+ reviewer,
+ packet,
+ decision,
+ engine.policy,
+ enforce_rbac=True,
+ at=datetime.now(timezone.utc),
+ )
diff --git a/tests/test_rest_api.py b/tests/test_rest_api.py
index 1a74fb9..bd0b1c0 100644
--- a/tests/test_rest_api.py
+++ b/tests/test_rest_api.py
@@ -336,8 +336,8 @@ def test_quality_endpoint(client):
resp = client.get("/v0/quality")
assert resp.status_code == 200
body = resp.json()
- assert body["report_version"] == "0.6"
- assert body["policy_version"] == "scope-core-v0.6"
+ assert body["report_version"] == "0.7"
+ assert body["policy_version"] == "scope-core-v0.7"
assert "metrics" in body
assert "warnings" in body
@@ -363,6 +363,7 @@ def test_quality_endpoint_custom_queue_dir(client, tmp_path, monkeypatch):
def test_akta_review_rest_endpoint(client, tmp_path):
from adapters.generic_rest import server
+ from scope.config import is_production_mode
server.reset_engine_cache()
out_dir = tmp_path / "akta_out"
@@ -380,6 +381,9 @@ def test_akta_review_rest_endpoint(client, tmp_path):
assert summary["status"] == "completed"
assert summary["approved_scope"] == "protocol_draft"
assert summary["scope_trust_root_hash"].startswith("sha256:")
+ assert summary["identity_assurance_level"] == "IAL0"
+ assert summary["signing_assurance_level"] == "SAL0"
+ assert summary["production_mode"] is is_production_mode()
assert (out_dir / "scope_grant.json").exists()
@@ -456,13 +460,14 @@ def test_grant_check_returns_reason_and_code(client):
def test_review_queue_rest_endpoints(client, tmp_path, monkeypatch):
from adapters.generic_rest import server
- monkeypatch.setenv("SCOPE_QUEUE_DIR", str(tmp_path / "queues"))
+ queue_dir = tmp_path / "queues"
+ monkeypatch.setenv("SCOPE_QUEUE_DIR", str(queue_dir))
server._engine = None
packet = client.post("/v0/packets", json=_packet_payload()).json()
created = client.post(
"/v0/review-queue",
- json={"packet": packet, "sla_hours": 24, "queue_dir": str(tmp_path / "queues")},
+ json={"packet": packet, "sla_hours": 24, "queue_dir": str(queue_dir)},
)
assert created.status_code == 200
queue_id = created.json()["queue_id"]
@@ -470,14 +475,100 @@ def test_review_queue_rest_endpoints(client, tmp_path, monkeypatch):
assigned = client.post(
f"/v0/review-queue/{queue_id}/assign",
json={"reviewer": {"reviewer_id": "r1", "role": "protocol_owner"}},
- params={"queue_dir": str(tmp_path / "queues")},
+ params={"queue_dir": str(queue_dir)},
)
assert assigned.status_code == 200
assert assigned.json()["status"] == "assigned"
- listed = client.get("/v0/review-queue", params={"queue_dir": str(tmp_path / "queues")})
+ in_review = client.post(
+ f"/v0/review-queue/{queue_id}/in-review",
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert in_review.status_code == 200
+ assert in_review.json()["status"] == "in_review"
+
+ needs_info = client.post(
+ f"/v0/review-queue/{queue_id}/needs-information",
+ json={"reason": "missing protocol appendix"},
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert needs_info.status_code == 200
+ assert needs_info.json()["status"] == "needs_information"
+
+ info_received = client.post(
+ f"/v0/review-queue/{queue_id}/information-received",
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert info_received.status_code == 200
+ assert info_received.json()["status"] == "in_review"
+
+ expired = client.post(
+ f"/v0/review-queue/{queue_id}/expire",
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert expired.status_code == 200
+ assert expired.json()["status"] == "expired"
+
+ reopened = client.post(
+ f"/v0/review-queue/{queue_id}/reopen",
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert reopened.status_code == 200
+ assert reopened.json()["status"] == "open"
+
+ assigned_again = client.post(
+ f"/v0/review-queue/{queue_id}/assign",
+ json={"reviewer": {"reviewer_id": "r1", "role": "protocol_owner"}},
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert assigned_again.status_code == 200
+
+ escalated = client.post(
+ f"/v0/review-queue/{queue_id}/escalate",
+ json={
+ "reviewer": {"reviewer_id": "lead1", "role": "lab_operations_lead"},
+ "reason": "SLA breach",
+ "actor_id": "ops-bot",
+ },
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert escalated.status_code == 200
+ assert escalated.json()["status"] == "escalated"
+
+ cancelled = client.post(
+ f"/v0/review-queue/{queue_id}/cancel",
+ json={"reason": "duplicate request"},
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert cancelled.status_code == 200
+ assert cancelled.json()["status"] == "cancelled"
+
+ listed = client.get("/v0/review-queue", params={"queue_dir": str(queue_dir)})
assert listed.status_code == 200
- assert listed.json()["open_queue_count"] == 1
+ counts = listed.json()["status_counts"]
+ assert counts.get("cancelled", 0) >= 1
+
+
+def test_review_queue_invalid_transition_rest_error(client, tmp_path, monkeypatch):
+ from adapters.generic_rest import server
+
+ queue_dir = tmp_path / "queues"
+ server._engine = None
+
+ packet = client.post("/v0/packets", json=_packet_payload()).json()
+ created = client.post(
+ "/v0/review-queue",
+ json={"packet": packet, "sla_hours": 24, "queue_dir": str(queue_dir)},
+ )
+ queue_id = created.json()["queue_id"]
+
+ resp = client.post(
+ f"/v0/review-queue/{queue_id}/grant",
+ json={"grant_id": "SCOPE-GRANT-BAD"},
+ params={"queue_dir": str(queue_dir)},
+ )
+ assert resp.status_code == 400
+ assert "expected decided" in resp.json()["detail"]
def test_key_registry_rest_endpoints(client, tmp_path, monkeypatch):
diff --git a/tests/test_review_queue.py b/tests/test_review_queue.py
index 8ae117a..81a2160 100644
--- a/tests/test_review_queue.py
+++ b/tests/test_review_queue.py
@@ -81,7 +81,8 @@ def test_review_queue_status_transitions(tmp_path):
assert entry.status == "open"
engine.assign_review_queue(out, {"reviewer_id": "rev1", "role": "protocol_owner"})
- assert ReviewQueue.load(out).status == "assigned"
+ engine.in_review_review_queue(out)
+ assert ReviewQueue.load(out).status == "in_review"
engine.decide_review_queue(out, "SCOPE-DEC-TEST01")
decided = ReviewQueue.load(out)
diff --git a/tests/test_review_queue_dashboard.py b/tests/test_review_queue_dashboard.py
new file mode 100644
index 0000000..078db04
--- /dev/null
+++ b/tests/test_review_queue_dashboard.py
@@ -0,0 +1,54 @@
+"""Tests for review queue HTML dashboard rendering."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from scope import ScopeEngine
+from scope.review_queue import ReviewQueue
+from scope.review_queue_render import render_queue_dashboard
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _sample_packet(engine: ScopeEngine) -> dict:
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_action": "draft_protocol",
+ "requested_tool": "protocol_editor.draft_change",
+ "requested_scope": "protocol_draft",
+ }
+ record = {"record_id": "AKTA-DASH", "scientific_action_type": "A5_protocol_modification"}
+ return engine.create_packet(record, trigger)
+
+
+def test_dashboard_renders_workflow_states(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _sample_packet(engine)
+
+ needs_info = ReviewQueue.create(packet, persist=False)
+ needs_info.assign({"reviewer_id": "r1", "role": "protocol_owner"})
+ needs_info.mark_in_review()
+ needs_info.mark_needs_information(reason="missing data")
+ needs_info.save(tmp_path / "needs_info.json")
+
+ expired = ReviewQueue.create(packet, persist=False)
+ expired.expire()
+ expired.save(tmp_path / "expired.json")
+
+ escalated = ReviewQueue.create(packet, persist=False)
+ escalated.assign({"reviewer_id": "r2", "role": "protocol_owner"})
+ escalated.mark_escalated(
+ {"reviewer_id": "lead1", "role": "lab_operations_lead"},
+ reason="SLA breach",
+ )
+ escalated.save(tmp_path / "escalated.json")
+
+ html = render_queue_dashboard(tmp_path)
+ assert "needs_information" in html
+ assert "expired" in html
+ assert "escalated" in html
+ assert "Status counts:" in html
+ assert html.count("needs_information") >= 1
+ assert html.count("expired") >= 1
diff --git a/tests/test_review_queue_state_machine.py b/tests/test_review_queue_state_machine.py
new file mode 100644
index 0000000..2456098
--- /dev/null
+++ b/tests/test_review_queue_state_machine.py
@@ -0,0 +1,158 @@
+"""Tests for review queue workflow state machine."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from scope import ScopeEngine
+from scope.errors import ScopeValidationError
+from scope.review_queue import ReviewQueue
+from scope.review_workflow import validate_transition
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _sample_packet(engine: ScopeEngine) -> dict:
+ trigger = {
+ "akta_admissibility": "review_required",
+ "scientific_action_type": "A5_protocol_modification",
+ "requested_action": "draft_protocol",
+ "requested_tool": "protocol_editor.draft_change",
+ "requested_scope": "protocol_draft",
+ }
+ record = {"record_id": "AKTA-RQ", "scientific_action_type": "A5_protocol_modification"}
+ return engine.create_packet(record, trigger)
+
+
+def test_forbidden_open_to_granted() -> None:
+ with pytest.raises(ScopeValidationError, match="Invalid"):
+ validate_transition("open", "granted")
+
+
+def test_forbidden_assigned_to_granted() -> None:
+ with pytest.raises(ScopeValidationError, match="Invalid"):
+ validate_transition("assigned", "granted")
+
+
+def test_forbidden_needs_information_to_granted() -> None:
+ with pytest.raises(ScopeValidationError, match="Invalid"):
+ validate_transition("needs_information", "granted")
+
+
+def test_forbidden_needs_information_to_decided() -> None:
+ with pytest.raises(ScopeValidationError, match="Invalid"):
+ validate_transition("needs_information", "decided")
+
+
+def test_information_received_workflow(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _sample_packet(engine)
+ entry = ReviewQueue.create(packet, persist=False)
+ entry.assign({"reviewer_id": "rev1", "role": "protocol_owner"})
+ entry.mark_in_review()
+ entry.mark_needs_information(reason="missing data")
+ entry.mark_information_received()
+ assert entry.status == "in_review"
+ entry.mark_decided("SCOPE-DEC-IR")
+ entry.mark_granted("SCOPE-GRANT-IR")
+ entry.save(tmp_path / "queue.json")
+
+
+def test_escalation_records_reason_and_actor(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=tmp_path / "events.jsonl")
+ packet = _sample_packet(engine)
+ entry = ReviewQueue.create(packet, persist=False)
+ entry.assign({"reviewer_id": "rev1", "role": "protocol_owner"})
+ entry.mark_escalated(
+ {"reviewer_id": "lead1", "role": "lab_operations_lead"},
+ reason="SLA breach",
+ actor_id="system",
+ )
+ assert entry.status == "escalated"
+ assert entry.to_artifact()["escalation_reason"] == "SLA breach"
+ assert entry.to_artifact()["escalation_actor_id"] == "system"
+
+
+def test_escalation_emits_ledger_event(tmp_path: Path) -> None:
+ ledger_path = tmp_path / "events.jsonl"
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy", ledger_path=ledger_path)
+ packet = _sample_packet(engine)
+ queue_path = tmp_path / "queue.json"
+ entry = engine.create_review_queue(packet, queue_dir=tmp_path)
+ entry.save(queue_path)
+ entry.assign({"reviewer_id": "rev1", "role": "protocol_owner"})
+ entry.save(queue_path)
+
+ engine.escalate_review_queue_entry(
+ queue_path,
+ {"reviewer_id": "lead1", "role": "lab_operations_lead"},
+ reason="SLA breach",
+ actor_id="ops-bot",
+ )
+
+ events = ledger_path.read_text(encoding="utf-8").strip().splitlines()
+ escalations = [
+ json.loads(line)
+ for line in events
+ if json.loads(line).get("event_type") == "review_queue_escalated"
+ ]
+ assert len(escalations) == 1
+ metadata = escalations[0]["metadata"]
+ assert metadata["escalation_reason"] == "SLA breach"
+ assert escalations[0]["actor_id"] == "ops-bot"
+
+
+def test_happy_path_workflow(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _sample_packet(engine)
+ out = tmp_path / "queue.json"
+ entry = ReviewQueue.create(packet, persist=False)
+ entry.save(out)
+
+ entry = ReviewQueue.load(out)
+ entry.assign({"reviewer_id": "rev1", "role": "protocol_owner"})
+ entry.mark_in_review()
+ entry.mark_decided("SCOPE-DEC-WF01")
+ entry.mark_granted("SCOPE-GRANT-WF01")
+ entry.save(out)
+
+ final = ReviewQueue.load(out)
+ assert final.status == "granted"
+
+
+def test_expired_requires_reopen_before_grant(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _sample_packet(engine)
+ out = tmp_path / "queue.json"
+ entry = ReviewQueue.create(packet, persist=False)
+ entry.expire()
+ entry.save(out)
+
+ loaded = ReviewQueue.load(out)
+ with pytest.raises(ScopeValidationError, match="expired"):
+ loaded.mark_granted("SCOPE-GRANT-EX")
+
+
+def test_reopen_from_expired(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _sample_packet(engine)
+ out = tmp_path / "queue.json"
+ entry = ReviewQueue.create(packet, persist=False)
+ entry.expire()
+ entry.reopen()
+ assert entry.status == "open"
+ entry.save(out)
+
+
+def test_needs_information_blocks_decided(tmp_path: Path) -> None:
+ engine = ScopeEngine.from_policy_dir(ROOT / "policy")
+ packet = _sample_packet(engine)
+ entry = ReviewQueue.create(packet, persist=False)
+ entry.assign({"reviewer_id": "rev1", "role": "protocol_owner"})
+ entry.mark_in_review()
+ entry.mark_needs_information(reason="missing data")
+ with pytest.raises(ScopeValidationError, match="needs_information"):
+ entry.mark_decided("SCOPE-DEC-NI")
diff --git a/tests/test_review_session.py b/tests/test_review_session.py
index aa71faf..cad2cbe 100644
--- a/tests/test_review_session.py
+++ b/tests/test_review_session.py
@@ -80,6 +80,15 @@ def test_two_valid_reviews_grant():
)
grant = engine.issue_grant_from_session(session, packet, [d1, d2])
assert grant["authorization"]["approved_scope"] == "single_validation_plan"
+ assert grant["provenance"]["authority_checks"] == d1["provenance"]["authority_checks"]
+ assert (
+ grant["provenance"]["identity_assurance_level"]
+ == d1["provenance"]["identity_assurance_level"]
+ )
+ assert (
+ grant["provenance"]["role_resolution_source"]
+ == d1["provenance"]["role_resolution_source"]
+ )
def test_conflict_blocks_grant():
diff --git a/tests/test_signing_assurance.py b/tests/test_signing_assurance.py
new file mode 100644
index 0000000..7deb599
--- /dev/null
+++ b/tests/test_signing_assurance.py
@@ -0,0 +1,68 @@
+"""Tests for signing assurance levels."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from scope.errors import GrantValidationError, ScopeValidationError
+from scope.signing_assurance import (
+ SAL0,
+ SAL1,
+ SAL2,
+ HsmKmsSigningProvider,
+ check_minimum_signing_assurance,
+ load_minimum_signing_assurance,
+ resolve_signing_assurance_level,
+)
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def test_sal0_unsigned() -> None:
+ assert resolve_signing_assurance_level({"decision_id": "D1"}) == SAL0
+
+
+def test_sal1_local_signature() -> None:
+ artifact = {
+ "decision_id": "D1",
+ "decision_signature": "abc",
+ "reviewer_public_key_ref": "reviewer1.pub",
+ }
+ level = resolve_signing_assurance_level(artifact, provider_name="local_pem")
+ assert level in (SAL0, SAL1)
+
+
+def test_sal2_env_provider() -> None:
+ artifact = {"decision_id": "D1", "decision_signature": "abc"}
+ level = resolve_signing_assurance_level(artifact, provider_name="env_key")
+ assert level == SAL2
+
+
+def test_minimum_signing_enforced_in_production(tmp_path: Path) -> None:
+ policy_dir = tmp_path / "policy"
+ policy_dir.mkdir()
+ cfg = ROOT / "policy" / "minimum_signing_assurance.yaml"
+ (policy_dir / "minimum_signing_assurance.yaml").write_text(
+ cfg.read_text(encoding="utf-8"), encoding="utf-8"
+ )
+ with pytest.raises(GrantValidationError, match="Signing assurance"):
+ check_minimum_signing_assurance(
+ SAL0,
+ policy_dir,
+ approved_scope="robot_queue_submission",
+ production=True,
+ )
+
+
+def test_minimum_policy_loaded() -> None:
+ cfg = load_minimum_signing_assurance(ROOT / "policy")
+ assert cfg.get("minimum_level") == "SAL1"
+ assert "robot_queue_submission" in cfg.get("high_risk_scopes", [])
+
+
+def test_hsm_provider_is_external_stub() -> None:
+ provider = HsmKmsSigningProvider()
+ with pytest.raises(ScopeValidationError, match="HSM"):
+ provider.get_signer()
diff --git a/tests/test_signing_identity.py b/tests/test_signing_identity.py
index a6d926a..042486e 100644
--- a/tests/test_signing_identity.py
+++ b/tests/test_signing_identity.py
@@ -90,7 +90,7 @@ def test_sign_enforces_reviewer_key_registry(tmp_path):
registry_path.write_text(
yaml.dump(
{
- "version": "scope-core-v0.6",
+ "version": "scope-core-v0.7",
"reviewers": {"rev1": expected_ref},
}
),