Action schemas in CI
An action schema says what an entry for one action type must carry: inputs, a rule, an approval,
a target pattern, and any member by its dotted path. You will write a tenbri.toml for
issue_refund, check a stream against it with tenbri schema check, test that the entries your
Python or TypeScript code writes pass it, have tenbrid refuse entries that break it, and run the
checks on every pull request.
Prerequisites
Section titled “Prerequisites”-
The
tenbriCLI, which runs every check:Terminal window cargo install tenbri-cli --version 0.1.0-beta.1 --locked -
A stream with refund decisions, such as the one from Record a refund, and
TENBRI_KEYSset as there (export TENBRI_KEYS=./keys). -
To test your own code, the binding for your language:
Nothing more.
Python 3.10 or later and pytest.
pip install --pre tenbri pytestuv add --prerelease allow tenbriuv add --dev pytestpoetry add --allow-prereleases tenbripoetry add --group dev pytestNode 22.18 or later, which runs .ts files directly, in a project with "type": "module"
(npm pkg set type=module).
npm install tenbri@nextpnpm add tenbri@nextyarn add tenbri@nextbun add tenbri@nextcurl and tenbrid:
cargo install tenbri-server --version 0.1.0-beta.1 --locked-
Write tenbri.toml.
tenbri.toml # Every action type must be declared here.unknown_actions = "reject"[actions.issue_refund]description = "Refund a card charge through Stripe."inputs = ["policy:refund"]rule = trueapproval = truetarget = "stripe:charge/*"[actions.issue_refund.fields."rule.result"]allowed = ["allow"]The requirements apply to intents. Outcomes, approvals and other entries that name an action are only checked for a declared action type.
-
Check a stream.
schema checkreadstenbri.tomlfrom the current directory (or--schema FILE), then every entry of the streams you name: directories, ors3://andr2://URLs. It exits 0 when every entry passes, 1 when one breaks its schema, and 2 when the file itself is invalid.Terminal window tenbri schema check audit/support; echo "exit $?"tenbri.toml: 1 action schemas, unknown actions rejectedaudit/support: 5 entries, 0 break their schemaexit 0Append an intent with no rule and no approval, and one for an action the file does not declare, then check again:
Terminal window tenbri append audit/support > /dev/null <<'EOF'{"kind":"decision","phase":"intent","source":"sdk","principal":{"kind":"agent","id":"support-assistant"},"action":{"type":"issue_refund","target":"stripe:charge/ch_9Zk2"},"inputs":[{"ref":"policy:refund","version":"15","digest":"hmac-sha256:9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa"}]}EOFtenbri append audit/support > /dev/null <<'EOF'{"kind":"decision","phase":"intent","source":"sdk","principal":{"kind":"agent","id":"support-assistant"},"action":{"type":"close_ticket","target":"zendesk:ticket/4821"}}EOFtenbri schema check audit/support; echo "exit $?"tenbri.toml: 1 action schemas, unknown actions rejectedaudit/support seq 6: rule: requiredaudit/support seq 6: approval: requiredaudit/support seq 7: action.type: `close_ticket` is not declared in tenbri.tomlaudit/support: 7 entries, 2 break their schemaexit 1A typo in
tenbri.tomlfails before any stream is read. Hereruleis misspelledrul:tenbri: tenbri.toml: TOML parse error at line 7, column 1|7 | rul = true| ^^^unknown field `rul`, expected one of `description`, `inputs`, `rule`, `approval`, `durability`, `target`, `fields`With no stream given,
schema checkonly validates the file:Terminal window tenbri schema check; echo "exit $?"tenbri.toml: 1 action schemas, unknown actions rejectedexit 0 -
Test the entries your code writes.
A production stream only shows a broken entry after it was written. A test catches it first: run the code that records a refund against a throwaway stream, then check that stream.
If your agent writes with
tenbri appendortenbrid, point it at a fixture stream in your test suite and runtenbri schema check <fixture>afterwards, as in step 2.Replace
record_refundwith a call into your own code.test_audit_schema.py import hashlibimport jsonimport subprocessfrom tenbri import Tenbridef entry_hash(entry: dict) -> str:line = json.dumps(entry, sort_keys=True, separators=(",", ":"), ensure_ascii=False)return "sha256:" + hashlib.sha256(b"\0" + line.encode()).hexdigest()def record_refund(log: Tenbri, charge: str) -> None:"""What the agent writes for a refund: replace with a call into your own code."""target = f"stripe:charge/{charge}"approval = log.append({"kind": "approval", "principal": "person:alice", "action": {"type": "issue_refund", "target": target}})with log.run("agent:support-assistant") as run:run.read("policy:refund", "14", digest="hmac-sha256:" + "9f2c41aa" * 8)with run.decision("issue_refund",target,rule={"ref": "rule:refund-under-100", "version": "3", "result": "allow"},approval={"by": approval["principal"], "entry": entry_hash(approval)},):passdef test_refund_entries_match_tenbri_toml(tmp_path, monkeypatch):monkeypatch.setenv("TENBRI_KEYS", str(tmp_path / "keys"))stream = tmp_path / "audit" / "support"with Tenbri.open(stream, create=True) as log:record_refund(log, "ch_3PqX")check = subprocess.run(["tenbri", "schema", "check", "--schema", "tenbri.toml", str(stream)],capture_output=True,text=True,)print(check.stdout)assert check.returncode == 0, check.stdout + check.stderrTerminal window python -m pytest -q test_audit_schema.py. [100%]1 passed in 0.17sDrop the
approval=argument and the test fails with the violation:E /private/var/folders/…/test_refund_entries_match_tenb0/audit/support seq 4: approval: required1 failed in 0.14sReplace
recordRefundwith a call into your own code.audit-schema.test.ts import assert from 'node:assert/strict'import { execFileSync } from 'node:child_process'import { createHash } from 'node:crypto'import { mkdtempSync } from 'node:fs'import { tmpdir } from 'node:os'import { join } from 'node:path'import { test } from 'node:test'import { Tenbri, type Entry } from 'tenbri'function canonical(v: unknown): string {if (Array.isArray(v)) return `[${v.map(canonical).join(',')}]`if (v !== null && typeof v === 'object') {const keys = Object.keys(v).sort()return `{${keys.map((k) => `${JSON.stringify(k)}:${canonical((v as Record<string, unknown>)[k])}`).join(',')}}`}return JSON.stringify(v)}const entryHash = (e: Entry) => 'sha256:' + createHash('sha256').update('\0' + canonical(e)).digest('hex')// What the agent writes for a refund: replace with a call into your own code.async function recordRefund(log: Tenbri, charge: string): Promise<void> {const target = `stripe:charge/${charge}`const approval = await log.append({ kind: 'approval', principal: 'person:alice', action: { type: 'issue_refund', target } })await log.run({ principal: 'agent:support-assistant' }, async (run) => {run.read('policy:refund', { version: '14', digest: 'hmac-sha256:' + '9f2c41aa'.repeat(8) })await run.decision({action: 'issue_refund',target,rule: { ref: 'rule:refund-under-100', version: '3', result: 'allow' },approval: { by: approval.principal, entry: entryHash(approval) },},async () => {},)})}test('refund entries match tenbri.toml', async () => {const dir = mkdtempSync(join(tmpdir(), 'tenbri-'))process.env.TENBRI_KEYS = join(dir, 'keys')const stream = join(dir, 'audit', 'support')const log = await Tenbri.open(stream, { create: true })await recordRefund(log, 'ch_3PqX')await log.close()// execFileSync throws, failing the test, when tenbri exits 1 (an entry breaks its schema) or 2.const out = execFileSync('tenbri', ['schema', 'check', '--schema', 'tenbri.toml', stream], { encoding: 'utf8' })console.log(out)assert.match(out, /0 break their schema/)})Terminal window node --test audit-schema.test.tstenbri.toml: 1 action schemas, unknown actions rejected/var/folders/…/tenbri-5hRPij/audit/support: 5 entries, 0 break their schema✔ refund entries match tenbri.toml (147.448833ms)ℹ tests 1ℹ pass 1ℹ fail 0Run
tenbridwith the schema in your test environment (step 4): a write that breaks it fails there with status 422, so your client’s tests fail too. -
Enforce the schema at write time.
tenbri appenddoes not applytenbri.toml; check the stream withtenbri schema check, or write throughtenbrid(the REST tab) to refuse bad entries before they are stored.The binding writes to storage directly and does not apply
tenbri.toml. Keep the test from step 3, or write throughtenbrid(the REST tab) to refuse bad entries before they are stored.The binding writes to storage directly and does not apply
tenbri.toml. Keep the test from step 3, or write throughtenbrid(the REST tab) to refuse bad entries before they are stored.Set
schemaintenbrid.toml(orTENBRI_SCHEMA). Make the token withtenbrid token.tenbrid.toml keys = "keys"schema = "tenbri.toml"[streams.support]path = "audit/support"[[tokens]]sha256 = "sha256:…" # from tenbrid tokenprincipal = { kind = "agent", id = "support-assistant" }Terminal window tenbrid --config tenbrid.tomlIn another terminal, read the schemas back, then send an intent with no rule and no approval:
Terminal window export TENBRI_URL=http://127.0.0.1:7070export TOKEN=tnbt_… # the token tenbrid printedcurl -s $TENBRI_URL/v1/schemas -H "Authorization: Bearer $TOKEN"curl -s -w '\n%{http_code}\n' -X POST $TENBRI_URL/v1/streams/support/intents \-H "Authorization: Bearer $TOKEN" \-H "Idempotency-Key: 01K80B2A3B4C5D6E7F8G9H0J1K" \-H "Content-Type: application/json" \-d '{"action":{"type":"issue_refund","target":"stripe:charge/ch_9Zk2"},"inputs":[{"ref":"policy:refund","version":"15","digest":"hmac-sha256:9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa9f2c41aa"}]}'{"actions":{"issue_refund":{"approval":true,"description":"Refund a card charge through Stripe.","fields":{"rule.result":{"allowed":["allow"]}},"inputs":["policy:refund"],"rule":true,"target":"stripe:charge/*"}},"unknown_actions":"reject"}{"error":{"code":"TENBRI_SCHEMA_VIOLATION","message":"entry 01K80B2A3B4C5D6E7F8G9H0J1K: rule: required; approval: required","violations":[{"path":"rule","message":"required"},{"path":"approval","message":"required"}]}}422An undeclared action type is refused the same way:
{"error":{"code":"TENBRI_SCHEMA_VIOLATION","message":"entry 01K80B2B3B4C5D6E7F8G9H0J1K: action.type: `close_ticket` is not declared in tenbri.toml","violations":[{"path":"action.type","message":"`close_ticket` is not declared in tenbri.toml"}]}}422Nothing is written: the entry never reaches the stream.
-
Run the checks on every pull request.
This workflow validates
tenbri.toml, then runs the test for your language. Add stream paths afterschema check(a test fixture, or ans3://stream with credentials from secrets) to check stored entries too..github/workflows/tenbri-schema.yml name: tenbri schemaon:pull_request:jobs:schema:runs-on: ubuntu-lateststeps:- uses: actions/checkout@v4- uses: dtolnay/rust-toolchain@stable- name: Install tenbrirun: cargo install tenbri-cli --version 0.1.0-beta.1 --locked- name: Check tenbri.tomlrun: tenbri schema check --schema tenbri.toml.github/workflows/tenbri-schema.yml name: tenbri schemaon:pull_request:jobs:schema:runs-on: ubuntu-lateststeps:- uses: actions/checkout@v4- uses: dtolnay/rust-toolchain@stable- uses: actions/setup-python@v5with:python-version: "3.12"- name: Install tenbrirun: |cargo install tenbri-cli --version 0.1.0-beta.1 --lockedpip install --pre tenbri pytest- name: Check tenbri.tomlrun: tenbri schema check --schema tenbri.toml- name: Check the entries the code writesrun: python -m pytest -q test_audit_schema.py.github/workflows/tenbri-schema.yml name: tenbri schemaon:pull_request:jobs:schema:runs-on: ubuntu-lateststeps:- uses: actions/checkout@v4- uses: dtolnay/rust-toolchain@stable- uses: actions/setup-node@v4with:node-version: 24- name: Install tenbrirun: |cargo install tenbri-cli --version 0.1.0-beta.1 --lockednpm ci- name: Check tenbri.tomlrun: tenbri schema check --schema tenbri.toml- name: Check the entries the code writesrun: node --test audit-schema.test.ts
What happened
Section titled “What happened”schema check loaded the schema, then checked every entry and printed each violation with its
sequence number and member path. It exits 1 when an entry breaks its schema and 2 when the file
itself is invalid, so CI fails either way, and the test in your language catches a missing rule or
approval before any real entry is written. tenbrid applies the same file at write time and
refuses a breaking append with TENBRI_SCHEMA_VIOLATION.
