diff --git a/.gitattributes b/.gitattributes index 372df16..99c50b8 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1 +1,3 @@ +# Preserve output-only terminal recording bytes for scriptreplay. +docs/demos/stable-vocabulary/output.log -text whitespace=cr-at-eol docs/demos/single-occurrence.log whitespace=cr-at-eol diff --git a/docs/demos/stable-vocabulary/README.md b/docs/demos/stable-vocabulary/README.md new file mode 100644 index 0000000..bee2774 --- /dev/null +++ b/docs/demos/stable-vocabulary/README.md @@ -0,0 +1,15 @@ +# Stable vocabulary builder replay + +Run `python3 docs/demos/stable-vocabulary/demo.py`. Identical private definition, +start, transition and registration inputs use old vocabulary on base +`0b6b8492497c6f9e83027efb624f7e3a6dc13188` and published vocabulary on the candidate +(the containing revision). The four builders with old vocabulary become zero; +the run identity and signature bytes stay unchanged. Unknown term ownership +refuses. This invokes pure builders with a stub public key; it proves no installed +or live-store behavior and performs no network, graph, key or scheduler operation. + +[Output](output.log) and [timing](timing.log) are an output-only terminal recording. +Download both and replay with +`scriptreplay --log-out output.log --log-timing timing.log`. +No inline playback is claimed. These small sanitized files stay in this existing +demo tree with source history. diff --git a/docs/demos/stable-vocabulary/demo.py b/docs/demos/stable-vocabulary/demo.py new file mode 100644 index 0000000..f068544 --- /dev/null +++ b/docs/demos/stable-vocabulary/demo.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Private before/after RDF builder replay; no network, keys or graph writes.""" +import importlib.util +import json +from pathlib import Path +import subprocess +import sys +import tempfile +from unittest import mock + +ROOT = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(ROOT)) +from shuttle import export, signing + +BASE = '0b6b8492497c6f9e83027efb624f7e3a6dc13188' +OLD = 'http://aegis.gastown.local/ontology/' + + +def load(directory, name): + path = directory / (name + '.py') + path.write_bytes(subprocess.check_output( + ['git', '-C', str(ROOT), 'show', BASE + ':shuttle/' + name + '.py'])) + spec = importlib.util.spec_from_file_location('shuttle._baseline_' + name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def measure(label, writer, signer): + record = {'run': 'example', 'definition': 'example', 'state': 'open', + 'at': '2026-11-13T00:00:00Z', 'agent': 'fixture', 'step': 'claim', + 'from': 'open', 'to': 'claimed', 'signature': 'ab' * 64} + with mock.patch.object(signer, 'public_key_hex', return_value='cd' * 32): + bodies = [writer.definition_turtle({'name': 'example', 'transitions': [{'step': 'claim'}]}), + writer.start_turtle(record), writer.transition_turtle(record, 3), + signer.registration_turtle('fixture', 'urn:shuttle:workflow:example')] + old = sum(OLD in body for body in bodies) + result = {'source': label, 'builders': len(bodies), 'builders_with_legacy_vocabulary': old, + 'run_identity': writer.run_iri('example'), + 'signature_hex_unchanged': record['signature'] in bodies[2]} + print(json.dumps(result, sort_keys=True), flush=True) + assert old == (4 if label == 'base' else 0) + assert result['run_identity'] == 'urn:shuttle:run:example' + assert result['signature_hex_unchanged'] + + +with tempfile.TemporaryDirectory() as directory, \ + mock.patch.dict('os.environ', {'SHUTTLE_ENTITY_NS': 'urn:shuttle:'}): + print('Same private records: definition, start, transition and registration builders.') + measure('base', load(Path(directory), 'export'), load(Path(directory), 'signing')) + measure('candidate', export, signing) + from shuttle import vocabulary + try: + vocabulary.term('UnknownTerm') + except KeyError: + print('PASS: unknown vocabulary owner refuses; no legacy fallback.') + else: + raise AssertionError('unknown term accepted') diff --git a/docs/demos/stable-vocabulary/output.log b/docs/demos/stable-vocabulary/output.log new file mode 100644 index 0000000..926a508 --- /dev/null +++ b/docs/demos/stable-vocabulary/output.log @@ -0,0 +1,7 @@ +Script started on 2026-11-13 11:56:44-05:00 [COMMAND="python3 docs/demos/stable-vocabulary/demo.py" ] +Same private records: definition, start, transition and registration builders. +{"builders": 4, "builders_with_legacy_vocabulary": 4, "run_identity": "urn:shuttle:run:example", "signature_hex_unchanged": true, "source": "base"} +{"builders": 4, "builders_with_legacy_vocabulary": 0, "run_identity": "urn:shuttle:run:example", "signature_hex_unchanged": true, "source": "candidate"} +PASS: unknown vocabulary owner refuses; no legacy fallback. + +Script done on 2026-11-13 11:56:44-05:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/stable-vocabulary/timing.log b/docs/demos/stable-vocabulary/timing.log new file mode 100644 index 0000000..43c97ae --- /dev/null +++ b/docs/demos/stable-vocabulary/timing.log @@ -0,0 +1,4 @@ +0.083690 80 +0.006088 149 +0.000153 154 +0.000024 61 diff --git a/docs/knowledge/repository-knowledge.json b/docs/knowledge/repository-knowledge.json index 4de0304..6186689 100644 --- a/docs/knowledge/repository-knowledge.json +++ b/docs/knowledge/repository-knowledge.json @@ -16,7 +16,7 @@ "source": "shuttle/cli.py", "subject": "http://aegis.gastown.local/ontology/code/shuttle/shuttle%2Fcli.py", "summary": "The CLI defines and advances workflows, exports history, verifies signatures and freezes windows. It has no HTTP or MCP server.", - "source_sha256": "0460cc7819309e7a87fcc4d4dfbca79f8307bb1a5fcad946e934f4f3fb3d1d4a", + "source_sha256": "33d945358af48e0aceea367e4bc969bdb70757e93ca7d3e59ca02ddb92631d83", "properties": { "interfaceKind": "CLI", "publicName": "shuttle" @@ -38,7 +38,7 @@ "source": "shuttle/signing.py", "subject": "http://aegis.gastown.local/ontology/code/shuttle/shuttle%2Fsigning.py", "summary": "SHUTTLE_KEY_DIR selects agent key storage; require-existing-key mode refuses missing or unsafe key files and does not create replacements.", - "source_sha256": "4e9a268d068030989e5131bb8d4d9e83e671b3c9e497aaa8fbd10449acc246e3", + "source_sha256": "28846cac3874b68524c3d60f041cce1091517b7e247dbdd7641f5d0ebda171fe", "properties": { "interfaceKind": "config", "publicName": "SHUTTLE_KEY_DIR / existing_keys_only" diff --git a/docs/stable-vocabulary.md b/docs/stable-vocabulary.md new file mode 100644 index 0000000..1e279ad --- /dev/null +++ b/docs/stable-vocabulary.md @@ -0,0 +1,36 @@ +# Stable vocabulary and historical run identity + +Fresh workflow, start, transition and verifier-registration Turtle uses the +published Intigraph vocabulary. The 19-term crosswalk comes from +[ownership.json](https://intigraph.ai/ids/ownership.json), SHA-256 +`a389be53b74259d0c515ff6db16d602474dc5fe38dc6cc47f0f54a2d1347d34e`. +Five classes belong to `https://w3id.org/intigraph/quipu/ns#`; fourteen properties +belong to `https://w3id.org/intigraph/camayoc/core/ns#`. +Unknown term names refuse instead of falling back to another vocabulary. + +Vocabulary ownership is separate from instance identity. The existing neutral +`urn:shuttle:` instance default and explicit `SHUTTLE_ENTITY_NS` contract remain. +Run, workflow, step, event and registration identities, signature messages, local +logs and watermarks retain their existing bytes. This change allocates no new +identity scheme, invokes no scoped mint, enrolls no verifier and rewrites no +historical graph. An explicitly configured historical instance base remains an +operator-controlled identity parameter; it must not be changed to migrate terms. + +Verification and signed fixed/calendar history readers query both the published +and historical literal predicates using explicit UNION branches, retaining the +established native-reader compatibility form rather than alternate predicate +paths for literal objects. Known vocabulary terms are normalized only +for comparison; subject and reference identities and signatures remain exact. +Duplicate aliases remain visible to existing cardinality checks and refuse. +No inference or `sameAs` activation is required for these readers. + +The legacy constant in `vocabulary.py` is a read-compatibility identifier. +Repository-share builders and queries retain their separate former-name contract +and private intermediate-index behavior; their old IRIs are not rewritten by +the runtime workflow writer change. + +Source tests prove the fresh default builders emit no old vocabulary IRIs and +historical signed facts remain readable. They do not establish installed, +released or live-store acceptance. Deployment still needs compatible loaded +shapes, normal release/adoption, one smoke check and an explicit rollback. +Source rollback is a normal revert; preserve existing outboxes and identities. diff --git a/shuttle/calendar_history.py b/shuttle/calendar_history.py index 097b1e7..e226bbd 100644 --- a/shuttle/calendar_history.py +++ b/shuttle/calendar_history.py @@ -9,7 +9,7 @@ import re from zoneinfo import ZoneInfo -from . import export, windows +from . import export, windows, vocabulary from .scheduled import ScheduledRefused, _rows from .scheduled_history import ScheduledHistory @@ -98,12 +98,13 @@ def measure(self, *, now, occurrence, positive): return {'verdict': 'NOT_DUE', 'slot': None} if any(v.eligible(dict(v.binding), s) is not True for s in (control, slot)): raise ScheduledRefused('current declaration eligibility unavailable') - a = export.AEGIS registration = _rows(v.query( f'SELECT ?verifier ?pk ?workflow (isIRI(?workflow) AS ?scopeIRI) WHERE {{ ' - f'<{v.registration}> a ?type ; <{a}verifier> ?verifier ; ' - f'<{a}publicKey> ?pk ; <{a}attests> ?workflow . ' - f'FILTER(?type = <{a}VerifierRegistration>) }} LIMIT 8', + f'<{v.registration}> a ?type . ' + f'{vocabulary.pattern("<"+v.registration+">", "verifier", "?verifier")} ' + f'{vocabulary.pattern("<"+v.registration+">", "publicKey", "?pk")} ' + f'{vocabulary.pattern("<"+v.registration+">", "attests", "?workflow")} ' + f'{vocabulary.type_filter("?type", "VerifierRegistration")} }} LIMIT 8', graph=v.identity_graph), 8) if registration != [{'verifier': v.binding['agent'], 'pk': v.public_key, 'workflow': v.binding['workflow'], 'scopeIRI': True}] or \ @@ -115,7 +116,7 @@ def measure(self, *, now, occurrence, positive): '(STR(?raw) AS ?value) (isIRI(?raw) AS ?iri) ' '(STR(DATATYPE(?raw)) AS ?datatype) WHERE { VALUES ?run { ' + ' '.join('<'+export.run_iri(run)+'>' for run in runs) + ' } ' - + f'?event <{a}inRun> ?run ; ?predicate ?raw }} LIMIT 64', + + f'{vocabulary.pattern("?event", "inRun", "?run")} ?event ?predicate ?raw }} LIMIT 64', graph=windows.window_iri(month)), 64) grouped = [[r for r in rows if r.get('ev') in (export.event_iri(run, 2), export.event_iri(run, 3))] for run in runs] diff --git a/shuttle/cli.py b/shuttle/cli.py index 37095a8..1ed4be4 100644 --- a/shuttle/cli.py +++ b/shuttle/cli.py @@ -245,12 +245,13 @@ def cmd_verify(args) -> int: defs, runs = _index() _, run, _ = _run_state(defs, runs, args.run) r = export_mod.run_iri(args.run) - a = export_mod.AEGIS + v = export_mod.vocabulary rows = qc.query( f"SELECT ?ev ?step ?from ?to ?at ?agent ?sig WHERE {{ " - f"?ev <{a}inRun> <{r}> ; <{a}atStep> ?step ; <{a}fromState> ?from ; " - f"<{a}toState> ?to ; <{export_mod.PROV}endedAtTime> ?at ; " - f"<{a}performedBy> ?agent ; <{a}signature> ?sig . }}", + f"{v.pattern('?ev', 'inRun', '<'+r+'>')} {v.pattern('?ev', 'atStep', '?step')} " + f"{v.pattern('?ev', 'fromState', '?from')} {v.pattern('?ev', 'toState', '?to')} " + f"?ev <{export_mod.PROV}endedAtTime> ?at . " + f"{v.pattern('?ev', 'performedBy', '?agent')} {v.pattern('?ev', 'signature', '?sig')} }}", graph=windows.window_iri(run["window"]), ).get("rows", []) if not rows: @@ -259,8 +260,9 @@ def cmd_verify(args) -> int: f"{run['window']} — export first, or the window moved" ) regs = qc.query( - f"SELECT ?verifier ?pk WHERE {{ ?reg a <{a}VerifierRegistration> ; " - f"<{a}verifier> ?verifier ; <{a}publicKey> ?pk . }}", + f"SELECT ?verifier ?pk WHERE {{ ?reg a ?registrationType . " + f"{v.pattern('?reg', 'verifier', '?verifier')} {v.pattern('?reg', 'publicKey', '?pk')} " + f"{v.type_filter('?registrationType', 'VerifierRegistration')} }}", graph=args.identity_graph, ).get("rows", []) keys = {row["verifier"]: row["pk"] for row in regs} diff --git a/shuttle/export.py b/shuttle/export.py index 561919d..2a332c3 100644 --- a/shuttle/export.py +++ b/shuttle/export.py @@ -11,10 +11,10 @@ What a transition exports (the camayoc workflow slice, `competency/workflow-and-archive.md` Q1–Q8): -- an `aegis:TransitionEvent` with step, from/to states, time, - `aegis:performedBy` and `aegis:signature` (hex ed25519 over the canonical +- an `qp:TransitionEvent` with step, from/to states, time, + `core:performedBy` and `core:signature` (hex ed25519 over the canonical `shuttle-transition-v1` message — re-derivable from these facts alone); -- a re-asserted `aegis:currentState` on the run — derived convenience; the +- a re-asserted `core:currentState` on the run — derived convenience; the events are the truth, nothing is mutated in place. Pre-export, shuttle verifies its OWN signature (fail fast on key drift); @@ -26,7 +26,7 @@ import os from . import quipu_client as qc -from . import signing, state, windows +from . import signing, state, windows, vocabulary class ExportError(RuntimeError): @@ -49,23 +49,23 @@ def event_iri(run_id: str, seq: int) -> str: return f"{entity_ns()}event:{run_id}:{seq}" -AEGIS = "http://aegis.gastown.local/ontology/" +AEGIS = vocabulary.LEGACY # compatibility only; new facts use exact term owners PROV = "http://www.w3.org/ns/prov#" def _prefixes() -> str: - return f"@prefix aegis: <{AEGIS}> .\n@prefix prov: <{PROV}> .\n" + return vocabulary.prefixes() + f"@prefix prov: <{PROV}> .\n" def definition_turtle(rec: dict) -> str: """A WorkflowDefinition and its declared steps — durable knowledge shape, but exported beside the runs so the window is self-describing.""" wf = workflow_iri(rec["name"]) - lines = [_prefixes(), f"<{wf}> a aegis:WorkflowDefinition ."] + lines = [_prefixes(), f"<{wf}> a qp:WorkflowDefinition ."] for t in rec["transitions"]: step = f"{wf}/step/{t['step']}" - lines.append(f"<{wf}> aegis:hasStep <{step}> .") - lines.append(f"<{step}> a aegis:WorkflowStep .") + lines.append(f"<{wf}> core:hasStep <{step}> .") + lines.append(f"<{step}> a qp:WorkflowStep .") return "\n".join(lines) + "\n" @@ -73,11 +73,11 @@ def start_turtle(rec: dict) -> str: r = run_iri(rec["run"]) return ( _prefixes() - + f"<{r}> a aegis:WorkflowRun ;\n" - + f' aegis:runOf <{workflow_iri(rec["definition"])}> ;\n' - + f' aegis:currentState "{rec["state"]}" ;\n' - + f' aegis:createdAt "{rec["at"]}" ;\n' - + f' aegis:sourceKind "observed" .\n' + + f"<{r}> a qp:WorkflowRun ;\n" + + f' core:runOf <{workflow_iri(rec["definition"])}> ;\n' + + f' core:currentState "{rec["state"]}" ;\n' + + f' core:createdAt "{rec["at"]}" ;\n' + + f' core:sourceKind "observed" .\n' ) @@ -87,16 +87,16 @@ def transition_turtle(rec: dict, seq: int) -> str: wf_step = f'{workflow_iri(rec["definition"])}/step/{rec["step"]}' return ( _prefixes() - + f"<{ev}> a aegis:TransitionEvent ;\n" - + f" aegis:inRun <{r}> ;\n" - + f" aegis:atStep <{wf_step}> ;\n" - + f' aegis:fromState "{rec["from"]}" ;\n' - + f' aegis:toState "{rec["to"]}" ;\n' + + f"<{ev}> a qp:TransitionEvent ;\n" + + f" core:inRun <{r}> ;\n" + + f" core:atStep <{wf_step}> ;\n" + + f' core:fromState "{rec["from"]}" ;\n' + + f' core:toState "{rec["to"]}" ;\n' + f' prov:endedAtTime "{rec["at"]}" ;\n' - + f' aegis:performedBy <{entity_ns()}agent:{rec["agent"]}> ;\n' - + f' aegis:signature "{rec["signature"]}" ;\n' - + f' aegis:sourceKind "observed" .\n' - + f'<{r}> aegis:currentState "{rec["to"]}" .\n' + + f' core:performedBy <{entity_ns()}agent:{rec["agent"]}> ;\n' + + f' core:signature "{rec["signature"]}" ;\n' + + f' core:sourceKind "observed" .\n' + + f'<{r}> core:currentState "{rec["to"]}" .\n' ) @@ -115,7 +115,7 @@ def _definitions_by_name(recs: list[dict]) -> dict[str, dict]: def _workflow_of(rec: dict) -> str | None: - """The workflow a record asserts `aegis:runOf` against, if any.""" + """The workflow a record asserts `core:runOf` against, if any.""" kind = rec.get("type") if kind in ("start", "transition"): return rec.get("definition") @@ -182,7 +182,7 @@ def export(root=None, strict: bool = False) -> dict: A `define` record exports into the window of the month it was RECORDED. Runs export into the window of the month they RAN. Those are the same month exactly once — the month a workflow is introduced. Every month - afterwards, a run asserts `aegis:runOf` against a workflow that the new + afterwards, a run asserts `core:runOf` against a workflow that the new window has never heard of, SHACL's class constraint refuses it, and one refusal stops the whole drain. diff --git a/shuttle/scheduled.py b/shuttle/scheduled.py index 6701704..2d1d660 100644 --- a/shuttle/scheduled.py +++ b/shuttle/scheduled.py @@ -17,7 +17,7 @@ import time from typing import Callable -from . import export, quipu_client as qc, signing, state, windows +from . import export, quipu_client as qc, signing, state, windows, vocabulary _FIELDS = {'job', 'revision', 'host', 'manifest_sha256', 'agent', 'workflow'} _IRI = re.compile(r'^[A-Za-z][A-Za-z0-9+.-]*:[^\s<>"{}|^`\\]+$') @@ -104,12 +104,13 @@ def admit(self, scope, scheduled_at): raise ScheduledRefused('nonnegative UTC slot required') if self.eligible(dict(self.binding), scheduled_at) is not True: raise ScheduledRefused('current declaration eligibility is not proven') - a = export.AEGIS rows = _rows(self.query( f'SELECT ?verifier ?pk ?workflow (isIRI(?workflow) AS ?scopeIRI) WHERE {{ ' - f'<{self.registration}> a ?type ; <{a}verifier> ?verifier ; ' - f'<{a}publicKey> ?pk ; <{a}attests> ?workflow . ' - f'FILTER(?type = <{a}VerifierRegistration>) }} LIMIT 8', + f'<{self.registration}> a ?type . ' + f'{vocabulary.pattern("<"+self.registration+">", "verifier", "?verifier")} ' + f'{vocabulary.pattern("<"+self.registration+">", "publicKey", "?pk")} ' + f'{vocabulary.pattern("<"+self.registration+">", "attests", "?workflow")} ' + f'{vocabulary.type_filter("?type", "VerifierRegistration")} }} LIMIT 8', graph=self.identity_graph), 8) if len(rows) != 1 or rows[0].get('scopeIRI') is not True or rows[0] != { 'verifier': self.binding['agent'], 'pk': self.public_key, @@ -242,24 +243,25 @@ def publish(self, run, scope, outcome): 'SELECT (STR(?event) AS ?ev) (STR(?predicate) AS ?p) ' '(STR(?raw) AS ?value) (isIRI(?raw) AS ?iri) ' '(STR(DATATYPE(?raw)) AS ?datatype) WHERE { ' - f'?event <{export.AEGIS}inRun> <{export.run_iri(run)}> ; ?predicate ?raw ' + f'{vocabulary.pattern("?event", "inRun", "<"+export.run_iri(run)+">")} ?event ?predicate ?raw ' '} LIMIT 64', graph=graph), 64) expected = set() for seq, rec in enumerate(recs[2:], 2): ev = export.event_iri(run, seq) - facts = [(_RDF+'type', export.AEGIS+'TransitionEvent', True), - (export.AEGIS+'inRun', export.run_iri(run), True), - (export.AEGIS+'atStep', self.binding['workflow']+'/step/'+rec['step'], True), - (export.AEGIS+'fromState', rec['from'], False), - (export.AEGIS+'toState', rec['to'], False), + facts = [(_RDF+'type', vocabulary.term('TransitionEvent'), True), + (vocabulary.term('inRun'), export.run_iri(run), True), + (vocabulary.term('atStep'), self.binding['workflow']+'/step/'+rec['step'], True), + (vocabulary.term('fromState'), rec['from'], False), + (vocabulary.term('toState'), rec['to'], False), (export.PROV+'endedAtTime', rec['at'], False), - (export.AEGIS+'performedBy', self.namespace+'agent:'+rec['agent'], True), - (export.AEGIS+'signature', rec['signature'], False), - (export.AEGIS+'sourceKind', 'observed', False)] + (vocabulary.term('performedBy'), self.namespace+'agent:'+rec['agent'], True), + (vocabulary.term('signature'), rec['signature'], False), + (vocabulary.term('sourceKind'), 'observed', False)] expected.update((ev, p, value, iri, None if iri else _XSD+'string') for p, value, iri in facts) observed = set() for row in rows: + row = vocabulary.normalize_fact(row) if type(row.get('iri')) is not bool: raise ScheduledRefused('typed read-back capability unavailable') observed.add((row.get('ev'), row.get('p'), row.get('value'), row['iri'], diff --git a/shuttle/scheduled_history.py b/shuttle/scheduled_history.py index 88ee7be..1bce5c7 100644 --- a/shuttle/scheduled_history.py +++ b/shuttle/scheduled_history.py @@ -12,7 +12,7 @@ import json import re -from . import export, quipu_client, signing, windows +from . import export, quipu_client, signing, windows, vocabulary from .scheduled import ScheduledRefused, _FIELDS, _RDF, _XSD, _iri, _rows @@ -76,16 +76,16 @@ def _month(self, slot): def _event(self, rows, run, seq, step, old, new): event = export.event_iri(run, seq) + rows = [vocabulary.normalize_fact(row) for row in rows] values = {row['p']: row['value'] for row in rows} - a = export.AEGIS - at, signature = values.get(export.PROV+'endedAtTime'), values.get(a+'signature') - expected = [(_RDF+'type', a+'TransitionEvent', True), - (a+'inRun', export.run_iri(run), True), - (a+'atStep', self.binding['workflow']+'/step/'+step, True), - (a+'fromState', old, False), (a+'toState', new, False), + at, signature = values.get(export.PROV+'endedAtTime'), values.get(vocabulary.term('signature')) + expected = [(_RDF+'type', vocabulary.term('TransitionEvent'), True), + (vocabulary.term('inRun'), export.run_iri(run), True), + (vocabulary.term('atStep'), self.binding['workflow']+'/step/'+step, True), + (vocabulary.term('fromState'), old, False), (vocabulary.term('toState'), new, False), (export.PROV+'endedAtTime', at, False), - (a+'performedBy', self.namespace+'agent:'+self.binding['agent'], True), - (a+'signature', signature, False), (a+'sourceKind', 'observed', False)] + (vocabulary.term('performedBy'), self.namespace+'agent:'+self.binding['agent'], True), + (vocabulary.term('signature'), signature, False), (vocabulary.term('sourceKind'), 'observed', False)] actual = set() for row in rows: if (set(row) - {'ev', 'p', 'value', 'iri', 'datatype'} or @@ -106,6 +106,7 @@ def _event(self, rows, run, seq, step, old, new): def _outcome(self, rows, slot, now): if not rows: return 'MISSED' + rows = [vocabulary.normalize_fact(row) for row in rows] run = self._run(slot) start = [r for r in rows if r.get('ev') == export.event_iri(run, 2)] end = [r for r in rows if r.get('ev') == export.event_iri(run, 3)] @@ -117,7 +118,7 @@ def _outcome(self, rows, slot, now): if not end: return 'MISSED' terminal = next((r.get('value') for r in end - if r.get('p') == export.AEGIS+'toState'), None) + if r.get('p') == vocabulary.term('toState')), None) if terminal not in ('succeeded', 'failed', 'skipped-by-hold'): raise ScheduledRefused('unknown signed terminal') ended = self._event(end, run, 3, terminal, 'running', terminal) @@ -149,12 +150,13 @@ def measure(self, *, now, positive_slot): raise ScheduledRefused('positive control must cover the same unambiguous month') if any(self.eligible(dict(self.binding), s) is not True for s in (positive_slot, slot)): raise ScheduledRefused('current declaration eligibility unavailable') - a = export.AEGIS registration = _rows(self.query( f'SELECT ?verifier ?pk ?workflow (isIRI(?workflow) AS ?scopeIRI) WHERE {{ ' - f'<{self.registration}> a ?type ; <{a}verifier> ?verifier ; ' - f'<{a}publicKey> ?pk ; <{a}attests> ?workflow . ' - f'FILTER(?type = <{a}VerifierRegistration>) }} LIMIT 8', + f'<{self.registration}> a ?type . ' + f'{vocabulary.pattern("<"+self.registration+">", "verifier", "?verifier")} ' + f'{vocabulary.pattern("<"+self.registration+">", "publicKey", "?pk")} ' + f'{vocabulary.pattern("<"+self.registration+">", "attests", "?workflow")} ' + f'{vocabulary.type_filter("?type", "VerifierRegistration")} }} LIMIT 8', graph=self.identity_graph), 8) if (len(registration) != 1 or registration[0].get('scopeIRI') is not True or registration != [{'verifier': self.binding['agent'], 'pk': self.public_key, @@ -167,7 +169,7 @@ def measure(self, *, now, positive_slot): '(STR(?raw) AS ?value) (isIRI(?raw) AS ?iri) ' '(STR(DATATYPE(?raw)) AS ?datatype) WHERE { VALUES ?run { ' + ' '.join('<'+iri+'>' for iri in iris) + ' } ' - + f'?event <{a}inRun> ?run ; ?predicate ?raw }} LIMIT 64', + + f'{vocabulary.pattern("?event", "inRun", "?run")} ?event ?predicate ?raw }} LIMIT 64', graph=windows.window_iri(month)), 64) grouped = [[r for r in rows if r.get('ev') in (export.event_iri(run, 2), export.event_iri(run, 3))] for run in runs] diff --git a/shuttle/signing.py b/shuttle/signing.py index 44823e0..5035acc 100644 --- a/shuttle/signing.py +++ b/shuttle/signing.py @@ -3,7 +3,7 @@ Mirrors quipu `src/signing.rs`'s custody exactly: one keypair per agent in a host file created 0600, auto-generated on first use, never printed. The private key never leaves the host; the PUBLIC key is registered by a HUMAN -as an `aegis:VerifierRegistration` fact in a `dataKind=identity` graph — +as an `qp:VerifierRegistration` fact in a `dataKind=identity` graph — shuttle never registers its own keys, which is what keeps the trust root human-owned (the quipu governance rule: quipu never self-registers either). @@ -26,6 +26,8 @@ import os import stat + +from . import vocabulary from contextlib import contextmanager from contextvars import ContextVar from pathlib import Path @@ -200,17 +202,17 @@ def verify_transition( def registration_turtle(agent: str, workflow_iri: str, root: Path | None = None) -> str: - """The `aegis:VerifierRegistration` a HUMAN loads into the identity graph. + """The `qp:VerifierRegistration` a HUMAN loads into the identity graph. Printed for a person to review and load — shuttle never posts this itself. The vocabulary is quipu-owned aegis terms, the same shape quipu's governance router verifies decisions against. """ pk = public_key_hex(agent, root) - return f"""@prefix aegis: . + return f"""{vocabulary.prefixes()} - a aegis:VerifierRegistration ; - aegis:verifier "{agent}" ; - aegis:attests <{workflow_iri}> ; - aegis:publicKey "{pk}" . + a qp:VerifierRegistration ; + core:verifier "{agent}" ; + core:attests <{workflow_iri}> ; + core:publicKey "{pk}" . """ diff --git a/shuttle/vocabulary.py b/shuttle/vocabulary.py new file mode 100644 index 0000000..dc18766 --- /dev/null +++ b/shuttle/vocabulary.py @@ -0,0 +1,50 @@ +"""Published term ownership; independent of deployment/run instance identities. + +Crosswalk: https://intigraph.ai/ids/ownership.json +SHA256 a389be53b74259d0c515ff6db16d602474dc5fe38dc6cc47f0f54a2d1347d34e. +Legacy IRIs are read aliases, never a fallback owner for an unknown term. +""" +LEGACY = 'http://aegis.gastown.local/ontology/' +CORE = 'https://w3id.org/intigraph/camayoc/core/ns#' +QUIPU = 'https://w3id.org/intigraph/quipu/ns#' +_OWNERS = {**{name: QUIPU for name in ( + 'WorkflowDefinition', 'WorkflowStep', 'WorkflowRun', 'TransitionEvent', + 'VerifierRegistration')}, **{name: CORE for name in ( + 'hasStep', 'runOf', 'currentState', 'createdAt', 'sourceKind', 'inRun', + 'atStep', 'fromState', 'toState', 'performedBy', 'signature', 'verifier', + 'attests', 'publicKey')}} +_ALIASES = {LEGACY + name: owner + name for name, owner in _OWNERS.items()} + + +def term(name: str) -> str: + return _OWNERS[name] + name + + +def pattern(subject: str, name: str, obj: str) -> str: + """Explicit UNION for native readers that drop literals on alternate paths. + + Subject/object are internal SPARQL fragments, never caller text. Keep each + alternative grouped before joining it to the next required field. + """ + return (f'{{ {{ {subject} <{term(name)}> {obj} }} UNION ' + f'{{ {subject} <{LEGACY + name}> {obj} }} }}') + + +def type_filter(variable: str, name: str) -> str: + return f'FILTER({variable} IN (<{term(name)}>, <{LEGACY + name}>))' + + +def normalize_fact(row: dict) -> dict: + """Keep row cardinality and all identity/signature bytes; map known terms only.""" + result = dict(row) + p = row.get('p') + if isinstance(p, str): + result['p'] = _ALIASES.get(p, p) + if (p == 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type' + and row.get('iri') is True and isinstance(row.get('value'), str)): + result['value'] = _ALIASES.get(row['value'], row['value']) + return result + + +def prefixes() -> str: + return f'@prefix core: <{CORE}> .\n@prefix qp: <{QUIPU}> .\n' diff --git a/tests/fixtures/shuttle-ownership.json b/tests/fixtures/shuttle-ownership.json new file mode 100644 index 0000000..769d99a --- /dev/null +++ b/tests/fixtures/shuttle-ownership.json @@ -0,0 +1,101 @@ +{ + "source": "https://intigraph.ai/ids/ownership.json", + "source_sha256": "a389be53b74259d0c515ff6db16d602474dc5fe38dc6cc47f0f54a2d1347d34e", + "terms": [ + { + "old": "http://aegis.gastown.local/ontology/TransitionEvent", + "new": "https://w3id.org/intigraph/quipu/ns#TransitionEvent", + "owner": "quipu/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/VerifierRegistration", + "new": "https://w3id.org/intigraph/quipu/ns#VerifierRegistration", + "owner": "quipu/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/WorkflowDefinition", + "new": "https://w3id.org/intigraph/quipu/ns#WorkflowDefinition", + "owner": "quipu/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/WorkflowRun", + "new": "https://w3id.org/intigraph/quipu/ns#WorkflowRun", + "owner": "quipu/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/WorkflowStep", + "new": "https://w3id.org/intigraph/quipu/ns#WorkflowStep", + "owner": "quipu/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/atStep", + "new": "https://w3id.org/intigraph/camayoc/core/ns#atStep", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/attests", + "new": "https://w3id.org/intigraph/camayoc/core/ns#attests", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/createdAt", + "new": "https://w3id.org/intigraph/camayoc/core/ns#createdAt", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/currentState", + "new": "https://w3id.org/intigraph/camayoc/core/ns#currentState", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/fromState", + "new": "https://w3id.org/intigraph/camayoc/core/ns#fromState", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/hasStep", + "new": "https://w3id.org/intigraph/camayoc/core/ns#hasStep", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/inRun", + "new": "https://w3id.org/intigraph/camayoc/core/ns#inRun", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/performedBy", + "new": "https://w3id.org/intigraph/camayoc/core/ns#performedBy", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/publicKey", + "new": "https://w3id.org/intigraph/camayoc/core/ns#publicKey", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/runOf", + "new": "https://w3id.org/intigraph/camayoc/core/ns#runOf", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/signature", + "new": "https://w3id.org/intigraph/camayoc/core/ns#signature", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/sourceKind", + "new": "https://w3id.org/intigraph/camayoc/core/ns#sourceKind", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/toState", + "new": "https://w3id.org/intigraph/camayoc/core/ns#toState", + "owner": "camayoc/core/ns" + }, + { + "old": "http://aegis.gastown.local/ontology/verifier", + "new": "https://w3id.org/intigraph/camayoc/core/ns#verifier", + "owner": "camayoc/core/ns" + } + ] +} diff --git a/tests/test_calendar_history.py b/tests/test_calendar_history.py index 2d5ff86..0ecd78c 100644 --- a/tests/test_calendar_history.py +++ b/tests/test_calendar_history.py @@ -55,7 +55,7 @@ def query(self, text, graph): self.assertEqual(graph, windows.window_iri('2026-10')) requested = set(re.findall(r'<(urn:shuttle:run:[^>]+)>', text)) events = {r['ev'] for r in f.facts.values() - if r['p'] == export.AEGIS+'inRun' and r['value'] in requested} + if r['p'] == export.vocabulary.term('inRun') and r['value'] in requested} rows = [dict(r) for r in f.facts.values() if r['ev'] in events] + self.extra return {'rows': rows, 'count': len(rows), 'truncated': self.truncated} @@ -95,7 +95,7 @@ def test_empty_partial_tampered_duplicate_truncated_and_transport(self): del self.f.facts[key] self.assertEqual(self.measure()['verdict'], 'UNKNOWN') self.f.facts = dict(original) - key = next(k for k in original if k[1] == export.AEGIS+'signature') + key = next(k for k in original if k[1] == export.vocabulary.term('signature')) self.f.facts[key] = dict(original[key], value='0'*128) self.assertEqual(self.measure()['verdict'], 'UNKNOWN') self.f.facts = original diff --git a/tests/test_export_and_cli.py b/tests/test_export_and_cli.py index 3f168d4..f55363a 100644 --- a/tests/test_export_and_cli.py +++ b/tests/test_export_and_cli.py @@ -20,19 +20,19 @@ def test_a_transition_exports_event_signature_and_reasserted_state(self): "agent": "agent-a", "signature": "ab" * 64, } t = export.transition_turtle(rec, 3) - self.assertIn("aegis:TransitionEvent", t) - self.assertIn('aegis:signature "' + "ab" * 64, t) - self.assertIn('aegis:fromState "open"', t) - self.assertIn('aegis:toState "claimed"', t) + self.assertIn("qp:TransitionEvent", t) + self.assertIn('core:signature "' + "ab" * 64, t) + self.assertIn('core:fromState "open"', t) + self.assertIn('core:toState "claimed"', t) # The re-asserted derived state rides along; the event is the truth. - self.assertIn('aegis:currentState "claimed"', t) - self.assertIn('aegis:sourceKind "observed"', t) + self.assertIn('core:currentState "claimed"', t) + self.assertIn('core:sourceKind "observed"', t) def test_a_start_exports_the_run_shape(self): rec = {"type": "start", "run": "r1", "definition": "triage", "state": "open", "at": "2026-08-01T00:00:00Z", "agent": "a"} t = export.start_turtle(rec) - self.assertIn("aegis:WorkflowRun", t) + self.assertIn("qp:WorkflowRun", t) self.assertIn("urn:shuttle:workflow:triage", t) @@ -144,7 +144,7 @@ def test_the_full_flow_folds_signs_and_exports(self): run_window = [kn for kn in knots if "2026-08" in kn["graph"]] self.assertTrue( any( - "a aegis:WorkflowDefinition" in kn["turtle"] + "a qp:WorkflowDefinition" in kn["turtle"] and "urn:shuttle:workflow:triage" in kn["turtle"] for kn in run_window ), @@ -184,7 +184,7 @@ class constraint refused it, and one refusal stopped the whole drain. for month in ("2026-08", "2026-09"): declared = [ kn for kn in knots - if month in kn["graph"] and "a aegis:WorkflowDefinition" in kn["turtle"] + if month in kn["graph"] and "a qp:WorkflowDefinition" in kn["turtle"] ] self.assertTrue( declared, diff --git a/tests/test_scheduled.py b/tests/test_scheduled.py index 1b1fbbf..1542a7d 100644 --- a/tests/test_scheduled.py +++ b/tests/test_scheduled.py @@ -67,12 +67,13 @@ def post(self, path, body): return {'members': []} if path == '/knot' and 'TransitionEvent' in body['turtle']: text = body['turtle'] - ev = re.search(r'<([^>]+)> a aegis:TransitionEvent', text).group(1) - facts = [(export.AEGIS+'TransitionEvent', 'type', True, _RDF())] - for predicate, value in re.findall(r' ((?:aegis|prov):\w+) (<[^>]+>|"[^"]*") [;.]', text): + ev = re.search(r'<([^>]+)> a qp:TransitionEvent', text).group(1) + prefixes = dict(re.findall(r'@prefix (\w+): <([^>]+)>', text)) + facts = [(export.vocabulary.term('TransitionEvent'), 'type', True, _RDF())] + for predicate, value in re.findall(r' ((?:core|prov):\w+) (<[^>]+>|"[^"]*") [;.]', text): prefix, name = predicate.split(':') facts.append((value[1:-1], name, value.startswith('<'), - export.AEGIS if prefix == 'aegis' else export.PROV)) + prefixes[prefix])) for value, predicate, iri, prefix in facts: self.facts[(ev, prefix+predicate)] = { 'ev': ev, 'p': prefix+predicate, 'value': value, 'iri': iri, diff --git a/tests/test_scheduled_history.py b/tests/test_scheduled_history.py index 202aff5..9bf73fc 100644 --- a/tests/test_scheduled_history.py +++ b/tests/test_scheduled_history.py @@ -40,7 +40,8 @@ def query(self, text, graph): self.assertEqual(graph, windows.window_iri('2026-10')) requested = set(re.findall(r'<(urn:shuttle:run:[^>]+)>', text)) events = {row['ev'] for row in f.facts.values() - if row['p'] == export.AEGIS+'inRun' and row['value'] in requested} + if export.vocabulary.normalize_fact(row)['p'] == export.vocabulary.term('inRun') + and row['value'] in requested} rows = [dict(row) for row in f.facts.values() if row['ev'] in events] rows += self.extra return {'rows': rows, 'count': len(rows), 'truncated': self.truncated} @@ -94,6 +95,31 @@ def test_empty_history_is_unknown_not_a_healthy_zero(self): self.fixture.facts.clear() self.assertEqual(self.measure()['verdict'], 'UNKNOWN') + def test_historical_legacy_facts_verify_without_rewriting_signed_identities(self): + self.due() + vocab = export.vocabulary + def old_term(value): + for prefix in (vocab.CORE, vocab.QUIPU): + if value.startswith(prefix): + return vocab.LEGACY + value[len(prefix):] + return value + before = list(self.fixture.facts.values()) + legacy = [{**row, 'p': old_term(row['p']), + 'value': old_term(row['value']) if row['p'] == fixtures._RDF()+'type' + else row['value']} for row in before] + self.fixture.facts = {(row['ev'], row['p']): row for row in legacy} + self.assertEqual(self.measure()['verdict'], 'SUCCEEDED') + self.assertEqual([r['ev'] for r in legacy], [r['ev'] for r in before]) + self.assertIn('<'+vocab.LEGACY+'inRun>', self.fixture.queries[-1][0]) + self.assertEqual(list(self.fixture.facts.values()), legacy) + + def test_duplicate_legacy_and_canonical_fact_stays_unknown(self): + self.due() + row = next(r for r in self.fixture.facts.values() + if r['p'] == export.vocabulary.term('signature')) + self.extra = [{**row, 'p': export.AEGIS+'signature'}] + self.assertEqual(self.measure()['verdict'], 'UNKNOWN') + def test_every_control_fact_is_required_and_integrity_checked(self): original = dict(self.fixture.facts) for key, row in original.items(): @@ -115,12 +141,12 @@ def test_every_control_fact_is_required_and_integrity_checked(self): def test_tampered_due_is_unknown_not_failed(self): self.due() event = export.event_iri(self.history._run(self.control+60), 3) - self.fixture.facts[(event, export.AEGIS+'signature')]['value'] = '0'*128 + self.fixture.facts[(event, export.vocabulary.term('signature'))]['value'] = '0'*128 self.assertEqual(self.measure()['verdict'], 'UNKNOWN') def test_failed_positive_control_is_not_coverage(self): event = export.event_iri(self.history._run(self.control), 3) - self.fixture.facts[(event, export.AEGIS+'toState')]['value'] = 'failed' + self.fixture.facts[(event, export.vocabulary.term('toState'))]['value'] = 'failed' self.assertEqual(self.measure()['verdict'], 'UNKNOWN') def test_truncated_transport_and_current_eligibility_unknown(self): @@ -147,7 +173,7 @@ def test_scope_registration_mismatch_unknown(self): def test_partial_target_cannot_be_reported_missing(self): self.due() event = export.event_iri(self.history._run(self.control+60), 3) - del self.fixture.facts[(event, export.AEGIS+'sourceKind')] + del self.fixture.facts[(event, export.vocabulary.term('sourceKind'))] self.assertEqual(self.measure()['verdict'], 'UNKNOWN') def test_start_outside_window_unknown_even_with_valid_signature(self): @@ -159,7 +185,7 @@ def test_start_outside_window_unknown_even_with_valid_signature(self): sig = signing.sign_transition(self.fixture.scope.agent, run, 'started', 'queued', 'running', at, require_existing_key=True) self.fixture.facts[(event, export.PROV+'endedAtTime')]['value'] = at - self.fixture.facts[(event, export.AEGIS+'signature')]['value'] = sig + self.fixture.facts[(event, export.vocabulary.term('signature'))]['value'] = sig self.assertEqual(self.measure()['verdict'], 'UNKNOWN') def test_future_or_reversed_end_unknown_even_with_valid_signature(self): @@ -172,7 +198,7 @@ def test_future_or_reversed_end_unknown_even_with_valid_signature(self): export.run_iri(self.history._run(self.control+60)), 'succeeded', 'running', 'succeeded', at, require_existing_key=True) self.fixture.facts[(event, export.PROV+'endedAtTime')]['value'] = at - self.fixture.facts[(event, export.AEGIS+'signature')]['value'] = sig + self.fixture.facts[(event, export.vocabulary.term('signature'))]['value'] = sig self.assertEqual(self.measure()['verdict'], 'UNKNOWN') def test_clock_and_positive_slot_types_are_strict(self): diff --git a/tests/test_stable_vocabulary.py b/tests/test_stable_vocabulary.py new file mode 100644 index 0000000..47ec11c --- /dev/null +++ b/tests/test_stable_vocabulary.py @@ -0,0 +1,65 @@ +"""Fresh exports use published terms; historical signed identities stay intact.""" +import unittest +from unittest import mock +import json +from pathlib import Path + +from shuttle import export, signing + +LEGACY = 'http://aegis.gastown.local/ontology/' +CORE = 'https://w3id.org/intigraph/camayoc/core/ns#' +QUIPU = 'https://w3id.org/intigraph/quipu/ns#' + + +class StableVocabulary(unittest.TestCase): + def test_every_writer_term_matches_the_retained_published_crosswalk(self): + from shuttle import vocabulary as vocab + snapshot = json.loads((Path(__file__).parent / 'fixtures/shuttle-ownership.json').read_text()) + self.assertEqual(len(snapshot['terms']), 19) + self.assertEqual(set(vocab._ALIASES), {row['old'] for row in snapshot['terms']}) + for row in snapshot['terms']: + self.assertEqual(vocab.term(row['old'][len(LEGACY):]), row['new']) + + def test_fresh_definition_start_transition_and_registration_have_no_old_terms(self): + definition = {'name': 'example', 'transitions': [{'step': 'claim'}]} + record = {'run': 'r1', 'definition': 'example', 'state': 'open', + 'at': '2026-11-13T00:00:00Z', 'agent': 'fixture', + 'step': 'claim', 'from': 'open', 'to': 'claimed', 'signature': 'ab' * 64} + with mock.patch.dict('os.environ', {'SHUTTLE_ENTITY_NS': 'urn:shuttle:'}), \ + mock.patch.object(signing, 'public_key_hex', return_value='cd' * 32): + bodies = [export.definition_turtle(definition), export.start_turtle(record), + export.transition_turtle(record, 3), + signing.registration_turtle('fixture', 'urn:shuttle:workflow:example')] + for body in bodies: + with self.subTest(body=body): + self.assertNotIn(LEGACY, body) + self.assertNotIn('scbrown.github.io/quechua', body) + self.assertIn(QUIPU, body) + self.assertIn(CORE, body) + self.assertIn('urn:shuttle:run:r1', bodies[1]) + + def test_mapped_terms_and_legacy_query_paths_are_exact(self): + from shuttle import vocabulary as vocab + self.assertEqual(vocab.term('WorkflowRun'), QUIPU + 'WorkflowRun') + self.assertEqual(vocab.term('currentState'), CORE + 'currentState') + query = vocab.pattern('?event', 'signature', '?signature') + self.assertIn('<' + LEGACY + 'signature>', query) + self.assertIn('<' + CORE + 'signature>', query) + self.assertIn('UNION', query) + self.assertNotIn('|', query) # native literal-object path compatibility + with self.assertRaises(KeyError): + vocab.term('InventedTerm') + + def test_normalization_keeps_instance_identity_and_detectable_alias_duplicates(self): + from shuttle import vocabulary as vocab + subject = LEGACY + 'historical-run' + old = {'ev': subject, 'p': LEGACY + 'inRun', 'value': subject, 'iri': True} + new = {**old, 'p': CORE + 'inRun'} + rows = [vocab.normalize_fact(row) for row in [old, new]] + self.assertEqual(rows, [new, new]) + self.assertEqual(len(rows), 2) # strict consumers must still see the collision + self.assertEqual(old['p'], LEGACY + 'inRun') # source is not mutated + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_state_and_signing.py b/tests/test_state_and_signing.py index 52d389c..1ec2352 100644 --- a/tests/test_state_and_signing.py +++ b/tests/test_state_and_signing.py @@ -91,7 +91,7 @@ def test_malformed_hex_is_could_not_check_not_false(self): def test_the_registration_turtle_names_agent_key_and_workflow(self): t = signing.registration_turtle("agent-a", "urn:shuttle:workflow:triage", self.root) - self.assertIn('aegis:verifier "agent-a"', t) + self.assertIn('core:verifier "agent-a"', t) self.assertIn(signing.public_key_hex("agent-a", self.root), t) self.assertIn("VerifierRegistration", t)