Personal and enterprise agent teams

python -m pip install https://docs.soterias.io/downloads/api-preview/soterias_local-0.1.0rc5%2Bapi2-cp311-abi3-win_amd64.whl

npm install https://docs.soterias.io/downloads/api-preview/soterias-core-0.1.0-rc.5-api.2.tgz
npm install --save-dev typescript@5.8.3 @types/node@24

python team.py travel travel-review

npx tsc team.mts --target ES2022 --module NodeNext --moduleResolution NodeNext --skipLibCheck
node team.mjs travel travel-review-ts

"""Runnable synthetic teams. No model, account or external effect required.

python team_workflows.py aml NEW_DIRECTORY
Also: travel, inbox, procurement, incident. These role functions are application
agents, not separate security principals. Ordinary observation enforces no policy.
"""

import argparse
import asyncio
import json
from pathlib import Path
from soterias import Soterias, canonicalize, canonical_digest

/** Deterministic application roles, with native observation. No external effects. */
import assert from "node:assert/strict";
import { mkdtempSync, mkdirSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { Soterias } from "@soterias/core";

type Workflow = "aml" | "travel" | "inbox" | "procurement" | "incident";
type Evidence = {
  workflow: Workflow;
  items: { id: string; value: number; eligible: boolean }[];
};
type Proposal = { selected: string[] };

# %% Collect versioned inputs
# Each role receives explicit data, rather than hidden shared state.
def collect(kind):
    """Collector agent: return synthetic source data with an explicit version."""
    datasets = {
        "aml": {
            "transactions": [
                {"account": "A01", "cents": 125000},
                {"account": "A01", "cents": 87500},
                {"account": "A02", "cents": 12000},
                {"account": "A03", "cents": 255000},
            ],
            "threshold": 200000,
        },
        "travel": {
            "options": [{"name": "Rail", "cost": 90}, {"name": "Flight", "cost": 240}],
            "budget": 150,
        },
        "inbox": {
            "messages": [{"id": "M1", "urgent": True}, {"id": "M2", "urgent": False}]
        },
        "procurement": {
            "quotes": [
                {"vendor": "V1", "cost": 700, "approved": True},
                {"vendor": "V2", "cost": 500, "approved": False},
            ],
            "budget": 800,
        },
        "incident": {
            "signals": [
                {"service": "api", "errors": 12},
                {"service": "worker", "errors": 0},
            ]
        },
    }
    return {"kind": kind, "version": "synthetic-v1", "data": datasets[kind]}

// %% Collect versioned inputs
// Pass explicit data between roles and preserve the independent check.
function collect(workflow: Workflow): Evidence {
  const datasets: Record<Workflow, Evidence["items"]> = {
    aml: [
      { id: "A01", value: 125000, eligible: true },
      { id: "A01", value: 87500, eligible: true },
      { id: "A02", value: 12000, eligible: true },
      { id: "A03", value: 255000, eligible: true },
    ],
    travel: [
      { id: "Rail", value: 90, eligible: true },
      { id: "Flight", value: 240, eligible: true },
    ],
    inbox: [
      { id: "M1", value: 1, eligible: true },
      { id: "M2", value: 0, eligible: true },
    ],
    procurement: [
      { id: "V1", value: 700, eligible: true },
      { id: "V2", value: 500, eligible: false },
    ],
    incident: [
      { id: "api", value: 12, eligible: true },
      { id: "worker", value: 0, eligible: true },
    ],
  };
  return { workflow, items: datasets[workflow] };
}

# %% Analyse the data
# This deterministic analyst can be replaced with your existing agent.
def analyse(source):
    """Analyst agent: calculate a proposal. It cannot approve or execute it."""
    data, kind = source["data"], source["kind"]
    if kind == "aml":
        totals = {}
        for row in data["transactions"]:
            totals[row["account"]] = totals.get(row["account"], 0) + row["cents"]
        return {
            "review_accounts": sorted(
                k for k, v in totals.items() if v >= data["threshold"]
            ),
            "review_total_cents": sum(
                v for v in totals.values() if v >= data["threshold"]
            ),
        }
    if kind == "travel":
        return {
            "shortlist": [
                x["name"] for x in data["options"] if x["cost"] <= data["budget"]
            ]
        }
    if kind == "inbox":
        return {"review_messages": [x["id"] for x in data["messages"] if x["urgent"]]}
    if kind == "procurement":
        return {
            "shortlist": [
                x["vendor"]
                for x in data["quotes"]
                if x["approved"] and x["cost"] <= data["budget"]
            ]
        }
    return {"investigate": [x["service"] for x in data["signals"] if x["errors"] > 0]}

// %% Analyse the data
// Pass explicit data between roles and preserve the independent check.
function analyse(evidence: Evidence): Proposal {
  const { workflow, items } = evidence;
  if (workflow === "aml") {
    const totals = new Map<string, number>();
    for (const item of items)
      totals.set(item.id, (totals.get(item.id) ?? 0) + item.value);
    return {
      selected: [...totals]
        .filter(([, total]) => total >= 200000)
        .map(([id]) => id)
        .sort(),
    };
  }
  return {
    selected: items
      .filter(
        (item) =>
          item.eligible &&
          (workflow === "travel"
            ? item.value <= 150
            : workflow === "procurement"
              ? item.value <= 800
              : item.value > 0),
      )
      .map((item) => item.id),
  };
}

# %% Verify the proposal independently
# Recompute the business criterion; returning a value does not prove task quality.
def verify(source, proposal):
    """Independent reviewer agent. It returns evidence, never signed authority."""
    if source["kind"] == "aml":
        rows, threshold = source["data"]["transactions"], source["data"]["threshold"]
        accounts = sorted({r["account"] for r in rows})
        flagged = [
            a
            for a in accounts
            if sum(r["cents"] for r in rows if r["account"] == a) >= threshold
        ]
        return proposal == {
            "review_accounts": flagged,
            "review_total_cents": sum(
                r["cents"] for r in rows if r["account"] in flagged
            ),
        }
    expected = {
        "travel": {"shortlist": ["Rail"]},
        "inbox": {"review_messages": ["M1"]},
        "procurement": {"shortlist": ["V1"]},
        "incident": {"investigate": ["api"]},
    }
    return proposal == expected[source["kind"]]

// %% Verify the proposal independently
// Pass explicit data between roles and preserve the independent check.
function verify(evidence: Evidence, proposal: Proposal): boolean {
  // Independent fixture expectations, not the analyst's own success flag.
  const expected: Record<Workflow, string[]> = {
    aml: ["A01", "A03"],
    travel: ["Rail"],
    inbox: ["M1"],
    procurement: ["V1"],
    incident: ["api"],
  };
  return (
    JSON.stringify(proposal.selected) ===
    JSON.stringify(expected[evidence.workflow])
  );
}

const workflow = process.argv[2] ?? "aml";
assert(
  ["aml", "travel", "inbox", "procurement", "incident"].includes(workflow),
);
const selectedDirectory = process.argv[3];
const directory = selectedDirectory ? resolve(selectedDirectory)
  : mkdtempSync(join(tmpdir(), `soterias-${workflow}-`));
if (selectedDirectory) mkdirSync(directory, {mode: 0o700});
const sdk = new Soterias({
  workflow: `team-${workflow}`,
  directory: join(directory, "metrics"),
});
const analyst = sdk.cache(analyse, {
  version: "team-analysis-v1",
  verify: verify,
});
try {
  const evidence = sdk.invoke(collect, workflow as Workflow);
  const proposal = sdk.invoke(analyst, evidence);
  assert.deepEqual(sdk.invoke(analyst, evidence), proposal);
  assert(sdk.invoke(verify, evidence, proposal));
  writeFileSync(
    join(directory, "proposal.json"),
    JSON.stringify(proposal, null, 2),
  );
  // Export the same four real observations in each local format.
  for (const format of ['json', 'csv', 'otlp.json']) {
    let offset = 0, number = 0;
    while (true) {
      const page = format === 'json' ? sdk.metrics.exportJson(offset)
        : format === 'csv' ? sdk.metrics.exportCsv(offset) : sdk.metrics.exportOtlp();
      writeFileSync(join(directory, `metrics-${number}.${format}`), page.data);
      if (page.nextOffset === null) break;
      offset = page.nextOffset; number++;
    }
  }
  console.log({ workflow, proposal, directory, securityEnforcement: false });
  console.log(sdk.metrics.snapshot().data());
} finally {
  analyst.close();
  await sdk.closeAsync();
}

# %% Coordinate and observe the team
# Use the same SDK client around the existing role calls and export local observations.
async def run(kind, directory):
    directory.mkdir(mode=0o700)  # Refuse an existing output directory.
    sdk = Soterias(workflow="team-" + kind, directory=directory / "metrics")
    cached = sdk.cache(analyse, version="team-analysis-v1", verify=verify)
    try:
        # Coordinator orchestrates role boundaries; no provider credentials are needed.
        source = sdk.invoke(collect, kind)
        proposal = sdk.invoke(cached, source)
        repeated = sdk.invoke(cached, source)
        assert repeated == proposal and sdk.invoke(verify, source, proposal)
        assert cached.cache_stats()["caller_accepted_hits"] == "1"
        record = canonicalize(
            json.dumps({"source": source, "proposal": proposal}).encode()
        )
        (directory / "proposal.json").write_bytes(record)
        # Local application output only: this is not a broker-approved case closure.
        print(
            json.dumps(
                {
                    "workflow": kind,
                    "proposal": proposal,
                    "digest": canonical_digest(record),
                    "security_enforcement": False,
                    "external_effects": False,
                }
            )
        )
        await sdk.flush()
        for suffix, export in (
            ("json", sdk.metrics.export_json),
            ("csv", sdk.metrics.export_csv),
            ("otlp.json", sdk.metrics.export_otlp),
        ):
            page, offset, number = None, 0, 0
            while True:
                page = export(offset) if suffix != "otlp.json" else export()
                (directory / f"metrics-{number}.{suffix}").write_bytes(page.data)
                if page.next_offset is None:
                    break
                offset, number = page.next_offset, number + 1
        print(json.dumps(sdk.metrics.snapshot().data()))
    finally:
        cached.close()
        await sdk.close_async()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "workflow", choices=("aml", "travel", "inbox", "procurement", "incident")
    )
    parser.add_argument("directory", type=Path)
    args = parser.parse_args()
    asyncio.run(run(args.workflow, args.directory.resolve()))