Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions apps/trustops-art-runner/Makefile
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
.PHONY: validate test smoke clean

validate: test smoke

test:
test -d .venv || python3 -m venv .venv
. .venv/bin/activate && python -m pip install --upgrade pip pytest && PYTHONPATH=src pytest -q tests

smoke:
mkdir -p ../../build/trustops-art-runner
PYTHONPATH=src python3 -m trustops_art_runner.cli run \
--profile art-smoke \
--manifest examples/functional-service.demo.json \
--output ../../build/trustops-art-runner/trustops-art-smoke.receipt.json \
--created-at 2026-05-05T00:00:00Z \
--artifact-ref artifact://trustops/art-smoke/demo-classifier/receipt.json
test -s ../../build/trustops-art-runner/trustops-art-smoke.receipt.json

clean:
rm -rf .venv ../../build/trustops-art-runner
27 changes: 27 additions & 0 deletions apps/trustops-art-runner/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
# TrustOps ART Runner

`trustops-art-runner` is the first TrustOps Fabric runtime slice for Prophet Platform.

It is intentionally a thin runner boundary. ART is treated as an isolated provider backend, not a core platform dependency. The first slice emits a deterministic synthetic `trustops-receipt.v1` robustness receipt so the platform can wire manifest input, receipt output, policy action, ledger ingestion, and guardrail consumption before adding heavyweight adversarial dependencies.

## Goals

- Consume a functional service manifest.
- Run the `art-smoke` TrustOps profile.
- Emit a normalized `trustops-receipt.v1` robustness receipt.
- Preserve data-boundary guarantees: raw data is not exported by default.
- Keep provider details behind the runner interface.

## Example

```bash
PYTHONPATH=apps/trustops-art-runner/src \
python3 -m trustops_art_runner.cli run \
--profile art-smoke \
--manifest apps/trustops-art-runner/examples/functional-service.demo.json \
--output build/trustops-art-runner/receipt.json
```

## Next implementation step

Replace the synthetic metric backend with isolated ART-backed attack probes while preserving the same receipt contract and CLI/API surface.
46 changes: 46 additions & 0 deletions apps/trustops-art-runner/examples/functional-service.demo.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
{
"schemaVersion": "functional-service.v1",
"service": {
"id": "demo-classifier",
"name": "Demo Classifier",
"ownerRepository": "SocioProphet/prophet-platform",
"status": "experimental",
"description": "Synthetic functional service fixture for the TrustOps ART smoke runner."
},
"function": "nlp",
"model": {
"modelRef": "model.demo-classifier@0.1.0",
"adapterRefs": [],
"runtime": "python-synthetic",
"mutableStatePolicy": "governed-runtime-only"
},
"inputs": [
"text/plain"
],
"outputs": [
"classification/demo"
],
"evals": {
"required": true,
"references": [
"trustops.profile.art-smoke"
],
"minimumPromotionGate": "trustgate.platform-default.robustness"
},
"governance": {
"ledgerRequired": true,
"guardrailRequired": true,
"routingRequired": true,
"policyRefs": [
"trustgate.platform-default"
]
},
"sourceosCarry": {
"allowed": true,
"carriesMutableModelState": false,
"clientRefRequired": true,
"launchProfileRefs": [
"sourceos.launch.trustops-art-smoke"
]
}
}
5 changes: 5 additions & 0 deletions apps/trustops-art-runner/src/trustops_art_runner/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
"""TrustOps ART runner package."""

__all__ = ["__version__"]

__version__ = "0.1.0"
52 changes: 52 additions & 0 deletions apps/trustops-art-runner/src/trustops_art_runner/cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
"""Command line interface for the TrustOps ART runner."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from trustops_art_runner.receipt import TrustOpsRunnerError, build_art_smoke_receipt, write_receipt


def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="trustops-art-runner")
subparsers = parser.add_subparsers(dest="command", required=True)

run_parser = subparsers.add_parser("run", help="Run a TrustOps ART profile")
run_parser.add_argument("--profile", default="art-smoke", help="TrustOps profile to run")
run_parser.add_argument("--manifest", required=True, type=Path, help="Functional service manifest JSON path")
run_parser.add_argument("--output", required=True, type=Path, help="Output receipt path")
run_parser.add_argument("--source-commit", default="0000000000000000000000000000000000000000")
run_parser.add_argument("--created-at", default=None, help="Optional RFC3339 timestamp for deterministic tests")
run_parser.add_argument("--artifact-ref", default=None, help="Optional durable artifact reference for receipt evidence")
return parser


def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)

try:
if args.command == "run":
receipt = build_art_smoke_receipt(
manifest_path=args.manifest,
profile=args.profile,
output_ref=args.artifact_ref,
source_commit=args.source_commit,
created_at=args.created_at,
)
write_receipt(receipt, args.output)
print(json.dumps({"status": "ok", "receipt": str(args.output), "receiptId": receipt["receiptId"]}, sort_keys=True))
return 0
except TrustOpsRunnerError as exc:
print(f"trustops-art-runner: {exc}", file=sys.stderr)
return 2

parser.error(f"unsupported command: {args.command}")
return 2


if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
199 changes: 199 additions & 0 deletions apps/trustops-art-runner/src/trustops_art_runner/receipt.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,199 @@
"""Receipt builder for the TrustOps ART runner.

The first implementation is deliberately deterministic and synthetic. It proves the
Prophet Platform TrustOps control-plane seam before importing ART or other heavy
provider dependencies into an isolated runner image.
"""

from __future__ import annotations

import datetime as dt
import hashlib
import json
from pathlib import Path
from typing import Any


TRUSTOPS_SCHEMA_VERSION = "trustops-receipt.v1"
DEFAULT_RUNNER_ID = "trustops-art-runner"
DEFAULT_PROVIDER_VERSION = "synthetic-art-smoke-0.1.0"
ZERO_COMMIT = "0000000000000000000000000000000000000000"


class TrustOpsRunnerError(ValueError):
"""Raised when runner input cannot produce a valid receipt."""


def _load_json(path: Path) -> dict[str, Any]:
try:
with path.open("r", encoding="utf-8") as handle:
data = json.load(handle)
except json.JSONDecodeError as exc: # pragma: no cover - stdlib message is enough
raise TrustOpsRunnerError(f"invalid JSON in {path}: {exc}") from exc
if not isinstance(data, dict):
raise TrustOpsRunnerError(f"expected object JSON in {path}")
return data


def _service_subject(manifest: dict[str, Any]) -> dict[str, str]:
service = manifest.get("service")
if not isinstance(service, dict):
raise TrustOpsRunnerError("manifest.service is required")

service_id = service.get("id")
owner_repository = service.get("ownerRepository")
if not isinstance(service_id, str) or not service_id:
raise TrustOpsRunnerError("manifest.service.id is required")
if not isinstance(owner_repository, str) or "/" not in owner_repository:
raise TrustOpsRunnerError("manifest.service.ownerRepository must be owner/name")

model = manifest.get("model", {})
model_ref = model.get("modelRef") if isinstance(model, dict) else None
version_ref = model_ref if isinstance(model_ref, str) and model_ref else "manifest-local"

return {
"kind": "functional-service",
"id": service_id,
"ownerRepository": owner_repository,
"versionRef": version_ref,
"functionalServiceRef": f"functional-service.{service_id}",
}


def _digest_json(value: dict[str, Any]) -> str:
payload = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
return "sha256:" + hashlib.sha256(payload).hexdigest()


def _receipt_digest(receipt: dict[str, Any]) -> str:
digestable = json.loads(json.dumps(receipt))
digestable.setdefault("provenance", {})["receiptDigest"] = "sha256:pending"
return _digest_json(digestable)


def build_art_smoke_receipt(
*,
manifest_path: Path,
profile: str = "art-smoke",
output_ref: str | None = None,
source_commit: str = ZERO_COMMIT,
created_at: str | None = None,
) -> dict[str, Any]:
"""Build a deterministic TrustOps robustness receipt.

The synthetic backend intentionally emits stable metrics below blocking
thresholds. Later ART-backed probes must preserve this external contract.
"""

if profile != "art-smoke":
raise TrustOpsRunnerError(f"unsupported profile: {profile}")

manifest = _load_json(manifest_path)
subject = _service_subject(manifest)
timestamp = created_at or dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
model_ref = subject["versionRef"]
manifest_hash = hashlib.sha1(str(manifest_path).encode("utf-8")).hexdigest()[:8]
receipt_id = f"trustops.art-smoke.{subject['id']}.{manifest_hash}"

receipt: dict[str, Any] = {
"schemaVersion": TRUSTOPS_SCHEMA_VERSION,
"receiptId": receipt_id,
"receiptType": "robustness",
"subject": subject,
"runner": {
"id": DEFAULT_RUNNER_ID,
"provider": "art",
"version": DEFAULT_PROVIDER_VERSION,
"executionMode": "ci",
"containerRef": "ghcr.io/socioprophet/trustops-art-runner:0.1.0",
"codeDigest": "sha256:synthetic-art-smoke",
},
"inputs": {
"manifests": [str(manifest_path)],
"datasetRefs": ["dataset-risk.synthetic-art-smoke"],
"modelRefs": [model_ref],
"policyRefs": ["trustgate.platform-default"],
"protectedAttributeRefs": [],
"dataBoundary": "synthetic",
"rawDataExported": False,
},
"evaluation": {
"profile": profile,
"threatClasses": ["evasion", "inference", "extraction"],
"metricFamilies": ["robustness", "privacy-leakage"],
"metrics": [
{
"name": "evasion_attack_success_rate",
"value": 0.08,
"threshold": 0.10,
"direction": "at-or-below",
"status": "pass",
"sliceRef": "all",
},
{
"name": "membership_inference_risk",
"value": "low",
"threshold": "medium",
"direction": "at-or-below",
"status": "pass",
"sliceRef": "all",
},
{
"name": "model_extraction_probe",
"value": "not-detected",
"threshold": "not-detected",
"direction": "equals",
"status": "pass",
"sliceRef": "all",
},
],
},
"policy": {
"gateRef": "trustgate.platform-default.robustness",
"decision": "allow",
"reviewer": "automation",
},
"result": {
"status": "pass",
"summary": "Synthetic ART smoke profile passed. This receipt validates the TrustOps runner seam before full ART-backed adversarial probes are enabled.",
"residualRisk": "low",
"recommendedMitigations": [
"Run full ART-backed robustness profile before production promotion.",
"Record the receipt in model-governance-ledger before enabling runtime promotion.",
],
},
"evidence": {
"artifactRefs": [output_ref or "artifact://trustops/art-smoke/receipt.json"],
"redactionPolicy": "metrics-only",
"factsheetRefs": [f"factsheet.{subject['id']}"],
},
"actions": [
{
"target": "model-governance-ledger",
"action": "record",
"details": "Store robustness receipt and promotion evidence.",
},
{
"target": "guardrail-fabric",
"action": "allow",
"details": "Synthetic smoke profile produced no blocking robustness signal.",
},
],
"provenance": {
"createdAt": timestamp,
"createdBy": DEFAULT_RUNNER_ID,
"sourceCommit": source_commit,
"receiptDigest": "sha256:pending",
"signatureRef": "unsigned://local-smoke",
},
}
receipt["provenance"]["receiptDigest"] = _receipt_digest(receipt)
return receipt


def write_receipt(receipt: dict[str, Any], output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")


__all__ = ["TrustOpsRunnerError", "build_art_smoke_receipt", "write_receipt"]
44 changes: 44 additions & 0 deletions apps/trustops-art-runner/tests/test_receipt.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
from pathlib import Path

import pytest

from trustops_art_runner.receipt import TrustOpsRunnerError, build_art_smoke_receipt


FIXTURE = Path(__file__).resolve().parents[1] / "examples" / "functional-service.demo.json"


def test_build_art_smoke_receipt_shape():
receipt = build_art_smoke_receipt(
manifest_path=FIXTURE,
created_at="2026-05-05T00:00:00Z",
source_commit="abc123",
output_ref="artifact://trustops/art-smoke/test.json",
)

assert receipt["schemaVersion"] == "trustops-receipt.v1"
assert receipt["receiptType"] == "robustness"
assert receipt["subject"]["kind"] == "functional-service"
assert receipt["subject"]["id"] == "demo-classifier"
assert receipt["runner"]["provider"] == "art"
assert receipt["inputs"]["rawDataExported"] is False
assert receipt["evaluation"]["profile"] == "art-smoke"
assert receipt["policy"]["decision"] == "allow"
assert receipt["result"]["status"] == "pass"
assert receipt["provenance"]["receiptDigest"].startswith("sha256:")


def test_art_smoke_receipt_has_required_actions():
receipt = build_art_smoke_receipt(
manifest_path=FIXTURE,
created_at="2026-05-05T00:00:00Z",
)
actions = {(item["target"], item["action"]) for item in receipt["actions"]}

assert ("model-governance-ledger", "record") in actions
assert ("guardrail-fabric", "allow") in actions


def test_unsupported_profile_fails_fast():
with pytest.raises(TrustOpsRunnerError, match="unsupported profile"):
build_art_smoke_receipt(manifest_path=FIXTURE, profile="full-art")
Loading