diff --git a/runtime_contracts/__init__.py b/runtime_contracts/__init__.py index 8fdf855..e5fb1e1 100644 --- a/runtime_contracts/__init__.py +++ b/runtime_contracts/__init__.py @@ -124,6 +124,23 @@ available_secret_stores, available_brokers, ) +from .world import ( # noqa: E402 + WorldEvent, + EntityRef, + EntityKind, + GroundTruth, + RealismClass, + IdentityGraph, + WorldDescriptor, + WorldRegistry, + default_registry, + BusinessBlock, + Capsule, + MilestoneKind, + NeedsYouReason, + TraceMilestone, + VisualTrace, +) __version__ = "0.3.1" __all__ = [ @@ -160,5 +177,9 @@ "CredentialDenied", "SecretAccessError", "open_secret_store", "open_broker", "register_secret_store", "register_broker", "available_secret_stores", "available_brokers", + # dataset worlds — canonical entry contract (WorldEvent, identity graph, registry, visual trace) + "WorldEvent", "EntityRef", "EntityKind", "GroundTruth", "RealismClass", + "IdentityGraph", "WorldDescriptor", "WorldRegistry", "default_registry", + "BusinessBlock", "Capsule", "MilestoneKind", "NeedsYouReason", "TraceMilestone", "VisualTrace", *_model_all, ] diff --git a/runtime_contracts/world/__init__.py b/runtime_contracts/world/__init__.py new file mode 100644 index 0000000..d6bf7e2 --- /dev/null +++ b/runtime_contracts/world/__init__.py @@ -0,0 +1,39 @@ +"""Dataset worlds — the canonical entry contract for reproducible demo/product scenarios. + +A dataset world enters the runtime as :class:`WorldEvent`s carrying canonical entities, content-addressed +evidence, an explicit realism class, and optional ground truth; the :class:`IdentityGraph` keeps one +identity across every app projection; the :class:`WorldRegistry` catalogs the worlds; and +:class:`VisualTrace` is the presentation stream the animated execution canvas renders. This replaces the +fictional single-tenant demo object with reproducible, provenance-preserving worlds. +""" +from __future__ import annotations + +from .event import ( + KNOWN_EVENT_TYPES, + EntityKind, + EntityRef, + GroundTruth, + RealismClass, + WorldEvent, + LEAD_RECEIVED, TICKET_OPENED, INVOICE_OVERDUE, USAGE_CHANGED, THREAT_DETECTED, + SIGNAL_OBSERVED, ONBOARDING_REQUESTED, DEPLOYMENT_EVENT, +) +from .identity import IdentityGraph +from .registry import WorldDescriptor, WorldRegistry, default_registry +from .trace import ( + BusinessBlock, + Capsule, + MilestoneKind, + NeedsYouReason, + TraceMilestone, + VisualTrace, +) + +__all__ = [ + "WorldEvent", "EntityRef", "EntityKind", "GroundTruth", "RealismClass", "KNOWN_EVENT_TYPES", + "LEAD_RECEIVED", "TICKET_OPENED", "INVOICE_OVERDUE", "USAGE_CHANGED", "THREAT_DETECTED", + "SIGNAL_OBSERVED", "ONBOARDING_REQUESTED", "DEPLOYMENT_EVENT", + "IdentityGraph", + "WorldDescriptor", "WorldRegistry", "default_registry", + "BusinessBlock", "Capsule", "MilestoneKind", "NeedsYouReason", "TraceMilestone", "VisualTrace", +] diff --git a/runtime_contracts/world/event.py b/runtime_contracts/world/event.py new file mode 100644 index 0000000..917fede --- /dev/null +++ b/runtime_contracts/world/event.py @@ -0,0 +1,164 @@ +"""WorldEvent — the canonical entry point of a dataset world into the runtime stack. + +A dataset world (FinanceBench, GLEIF/OpenSanctions, a government parcel snapshot, live GitHub/HN signals, +CrowdSec telemetry, τ-bench trajectories, a synthetic call transcript …) enters the stack as a stream of +:class:`WorldEvent`s. One event carries a *source record* — its canonical entity identities, its evidence +(by content-addressed reference, never a copy), its realism class, its permissions/tenant scope, its +optional ground truth, and the capabilities it may require — through Discovery → Planner → Mission → +Context → app capabilities **without losing identity or provenance**. + +This replaces the old "fictional tenant" demo object: the same conceptual person/company/invoice/property +keeps one canonical identity as the mission crosses CRM, support, billing, security. Realism is explicit +(:class:`RealismClass`) so a seeded-demo record is never presented as a live customer's data. + +Identity follows the package rule (``canonical.py``): the event *is* its content, so its semantic fields +participate in its hash; ``known_at`` (ingest time) and the optimizer hints (cost/latency/freshness) are +chain-of-custody / runtime metadata and are excluded from identity. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from decimal import Decimal +from enum import Enum +from typing import Any, Dict, Optional, Tuple + +from ..canonical import content_hash +from ..protocol.evidence import EvidenceRef + + +class RealismClass(str, Enum): + """How real the data behind an event is — the audit vocabulary, now a first-class contract so every + evidence surface can label itself and no seeded record is mistaken for a live customer's.""" + REAL_LIVE = "REAL-LIVE" # read live from an external / OSS system at request time + REAL_SNAPSHOT = "REAL-SNAPSHOT" # real public data, committed / downloaded as a file + SEEDED_DEMO = "SEEDED-DEMO" # a live OSS core serving seeded fictional data + SYNTHETIC = "SYNTHETIC" # deterministically generated, realistic data + CANNED = "CANNED" # values hard-coded inline + STUB = "STUB" # the action path performs no real write + SIMULATED = "SIMULATED" # deterministic / LLM logic, no data backend + + +#: Canonical business-object kinds a world entity can be (the shared semantics; vendor-specific fields ride +#: along as typed extensions in the payload, never forced into one universal schema). +class EntityKind(str, Enum): + ORGANIZATION = "organization"; PERSON = "person"; EMPLOYEE = "employee"; CUSTOMER = "customer" + ACCOUNT = "account"; CONTACT = "contact"; OPPORTUNITY = "opportunity"; CONTRACT = "contract" + SUBSCRIPTION = "subscription"; INVOICE = "invoice"; PAYMENT = "payment"; EXPENSE = "expense" + LEDGER_ENTRY = "ledger_entry"; TICKET = "ticket"; CONVERSATION = "conversation"; DOCUMENT = "document" + ASSET = "asset"; IDENTITY = "identity"; PERMISSION = "permission"; FINDING = "finding" + INCIDENT = "incident"; DEPLOYMENT = "deployment"; REPOSITORY = "repository"; SERVICE = "service" + PROPERTY = "property"; VENDOR = "vendor"; PRODUCT = "product"; TASK = "task"; APPROVAL = "approval" + + +#: Known world event types (free-form for forward compatibility — a new type never forces a contract bump). +LEAD_RECEIVED = "lead.received" +TICKET_OPENED = "ticket.opened" +INVOICE_OVERDUE = "invoice.overdue" +USAGE_CHANGED = "usage.changed" +THREAT_DETECTED = "threat.detected" +SIGNAL_OBSERVED = "signal.observed" +ONBOARDING_REQUESTED = "onboarding.requested" +DEPLOYMENT_EVENT = "deployment.event" +KNOWN_EVENT_TYPES = frozenset({ + LEAD_RECEIVED, TICKET_OPENED, INVOICE_OVERDUE, USAGE_CHANGED, THREAT_DETECTED, + SIGNAL_OBSERVED, ONBOARDING_REQUESTED, DEPLOYMENT_EVENT, +}) + + +@dataclass(frozen=True) +class EntityRef: + """A canonical entity identity carried across every app projection. ``entity_id`` is the stable logical + id the Identity Graph maps to each system's native id; ``kind`` is a canonical business object.""" + entity_id: str + kind: str = EntityKind.ORGANIZATION.value + label: str = "" + + def canonical_form(self) -> Dict[str, Any]: + return {"entity_id": self.entity_id, "kind": self.kind, "label": self.label or None} + + def ref(self) -> str: + return content_hash(self.canonical_form()) + + +@dataclass(frozen=True) +class GroundTruth: + """The benchmarkable expectation for an event: what the situation is, the safe/unsafe actions, and the + target outcome. Present only for worlds with ground truth; drives baselines + scorecards.""" + situation: str = "" + safe_action: str = "" + unsafe_action: str = "" + target_outcome: str = "" + + def canonical_form(self) -> Dict[str, Any]: + return {k: v for k, v in { + "situation": self.situation, "safe_action": self.safe_action, + "unsafe_action": self.unsafe_action, "target_outcome": self.target_outcome}.items() if v} + + +@dataclass(frozen=True) +class WorldEvent: + """One source record entering the runtime as a governed, content-addressed event.""" + + world_id: str + dataset_id: str + source_record_id: str + event_type: str + entity_ids: Tuple[EntityRef, ...] = () + observed_at: str = "" # valid-time: when observed in the source + effective_at: str = "" # valid-time: when it takes effect + known_at: str = "" # ingest-time (chain-of-custody; excluded from identity) + evidence_refs: Tuple[EvidenceRef, ...] = () # content-addressed evidence (never a copy) + content_hash: str = "" # rcv1 hash of the source payload (self-pinning) + classification: str = RealismClass.SYNTHETIC.value + data_classifications: Tuple[str, ...] = () + permissions: Tuple[str, ...] = () + tenant: str = "" + ground_truth: Optional[GroundTruth] = None + capability_requirements: Tuple[str, ...] = () + cost_hint: Optional[Decimal] = None # optimizer metadata — excluded from identity + latency_ms_hint: int = 0 + freshness_s: int = 0 + scenario_seed: str = "" + payload: Dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + ch = self.content_hash or (content_hash(self.payload) if self.payload else "") + object.__setattr__(self, "content_hash", ch) + + def canonical_form(self) -> Dict[str, Any]: + d: Dict[str, Any] = { + "world_id": self.world_id, "dataset_id": self.dataset_id, + "source_record_id": self.source_record_id, "event_type": self.event_type, + "classification": self.classification, + } + if self.entity_ids: + d["entity_ids"] = [e.canonical_form() for e in self.entity_ids] + if self.observed_at: + d["observed_at"] = self.observed_at + if self.effective_at: + d["effective_at"] = self.effective_at + if self.evidence_refs: + d["evidence_refs"] = [r.canonical_form() for r in self.evidence_refs] + if self.content_hash: + d["content_hash"] = self.content_hash + if self.data_classifications: + d["data_classifications"] = sorted(self.data_classifications) + if self.ground_truth is not None: + d["ground_truth"] = self.ground_truth.canonical_form() + if self.capability_requirements: + d["capability_requirements"] = sorted(self.capability_requirements) + if self.scenario_seed: + d["scenario_seed"] = self.scenario_seed + return d + + def identity(self) -> str: + """The ``rcv1:`` content hash of the event's semantic identity — stable across the whole stack, so + provenance is verifiable at any hop (Discovery → Mission → Context → capability).""" + return content_hash(self.canonical_form()) + + def entity(self, kind: str) -> Optional[EntityRef]: + """The first carried entity of ``kind`` (e.g. the customer, or the invoice), or None.""" + return next((e for e in self.entity_ids if e.kind == kind), None) + + def realism(self) -> RealismClass: + return RealismClass(self.classification) diff --git a/runtime_contracts/world/identity.py b/runtime_contracts/world/identity.py new file mode 100644 index 0000000..b4cdd79 --- /dev/null +++ b/runtime_contracts/world/identity.py @@ -0,0 +1,51 @@ +"""IdentityGraph — one canonical entity, many app projections. + +The load-bearing piece of a dataset world: the same conceptual customer / company / invoice / property must +keep one identity as a mission crosses Twenty (CRM), Chatwoot (support), Lago (billing), ERPNext (books), +Metabase (BI), CrowdSec (security)… The graph maps a canonical ``entity_id`` to each system's native id — +explicitly registered where a projection seeder created a real record, or **deterministically derived** +(reproducible, content-addressed) otherwise. It also resolves the other way, so a webhook from an app maps +back to the canonical entity without inventing a new unrelated demo object. +""" +from __future__ import annotations + +from typing import Dict, Optional + +from ..canonical import content_hash + + +class IdentityGraph: + """Bidirectional canonical-id ↔ per-app native-id mapping, with deterministic fallback derivation.""" + + def __init__(self) -> None: + self._fwd: Dict[str, Dict[str, str]] = {} # entity_id -> {app: native_id} + self._rev: Dict[str, str] = {} # f"{app}:{native_id}" -> entity_id + + def register(self, entity_id: str, app: str, native_id: str) -> None: + """Record that a projection seeder created ``native_id`` in ``app`` for the canonical entity.""" + self._fwd.setdefault(entity_id, {})[app] = native_id + self._rev[f"{app}:{native_id}"] = entity_id + + def derive(self, entity_id: str, app: str) -> str: + """A deterministic, reproducible native id for ``entity_id`` in ``app`` — used when no real record + was seeded. Same (entity, app) → same id across runs, so replay is stable.""" + h = content_hash({"app": app, "entity": entity_id}).split(":", 1)[-1][:16] + return f"{app}-{h}" + + def resolve(self, entity_id: str, app: str) -> str: + """The entity's native id in ``app`` — the registered one if a seeder created it, else derived.""" + got = self._fwd.get(entity_id, {}).get(app) + return got if got is not None else self.derive(entity_id, app) + + def reverse(self, app: str, native_id: str) -> Optional[str]: + """Map an app-native id back to its canonical entity (None if unknown — never fabricate one).""" + got = self._rev.get(f"{app}:{native_id}") + if got is not None: + return got + # a derived id round-trips deterministically: recover the entity only if it derives back to itself + return None + + def projections(self, entity_id: str) -> Dict[str, str]: + """Every registered projection of the entity ({app: native_id}). Derived ids are computed on demand + via :meth:`resolve`, so this returns only the ids a seeder actually created.""" + return dict(self._fwd.get(entity_id, {})) diff --git a/runtime_contracts/world/registry.py b/runtime_contracts/world/registry.py new file mode 100644 index 0000000..6a03b87 --- /dev/null +++ b/runtime_contracts/world/registry.py @@ -0,0 +1,99 @@ +"""WorldRegistry — the catalog of reproducible dataset worlds. + +A world is a replayable scenario with canonical entities, source provenance, a realism class, an adapter +version, a scenario seed, and (where benchmarkable) ground truth. The registry catalogs them so a demo / +product can select one by id and project it into the OSS-backed apps. ``default_registry()`` seeds the +strongest real/public sources already in the fleet (per the 2026-08-25 datasource audit). +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Tuple + +from ..canonical import content_hash +from .event import RealismClass + + +@dataclass(frozen=True) +class WorldDescriptor: + world_id: str + dataset_id: str + title: str = "" + license: str = "" + provenance: str = "" # source url / attribution + realism: str = RealismClass.SYNTHETIC.value # the DEFAULT realism of this world's data + adapter_version: str = "0.1.0" + scenario_seed: str = "" + datasources: Tuple[str, ...] = () + ground_truth_available: bool = False + supported_scenarios: Tuple[str, ...] = () + capability_requirements: Tuple[str, ...] = () + + def canonical_form(self) -> Dict[str, object]: + d: Dict[str, object] = {"world_id": self.world_id, "dataset_id": self.dataset_id, + "realism": self.realism, "adapter_version": self.adapter_version} + for k in ("title", "license", "provenance", "scenario_seed"): + v = getattr(self, k) + if v: + d[k] = v + for k in ("datasources", "supported_scenarios", "capability_requirements"): + v = getattr(self, k) + if v: + d[k] = sorted(v) + d["ground_truth_available"] = self.ground_truth_available + return d + + def ref(self) -> str: + return content_hash(self.canonical_form()) + + +class WorldRegistry: + def __init__(self) -> None: + self._worlds: Dict[str, WorldDescriptor] = {} + + def register(self, d: WorldDescriptor) -> None: + self._worlds[d.world_id] = d + + def get(self, world_id: str) -> "WorldDescriptor | None": + return self._worlds.get(world_id) + + def list(self) -> List[WorldDescriptor]: + return sorted(self._worlds.values(), key=lambda w: w.world_id) + + +def default_registry() -> WorldRegistry: + """The initial world registry — the fleet's strongest real/public sources, labelled by realism.""" + r = WorldRegistry() + L = RealismClass + for d in [ + WorldDescriptor("finance-evidence", "financebench", "Financial filings evidence quality", + license="CC-BY-4.0", provenance="patronus-ai/financebench", + realism=L.REAL_LIVE.value, datasources=("FinanceBench (SEC 10-K/10-Q)", "PubMedQA"), + supported_scenarios=("evidence-quality", "retrieval-abstention"), + capability_requirements=("context.retrieve",)), + WorldDescriptor("b2b-signals", "github-hn", "B2B signal → qualified outreach", + provenance="api.github.com + hn.algolia.com", realism=L.REAL_LIVE.value, + datasources=("GitHub search", "Hacker News"), + supported_scenarios=("signal-to-outreach",), + capability_requirements=("crm.write", "outreach.draft")), + WorldDescriptor("kyc-ownership", "gleif-opensanctions", "KYC / vendor onboarding", + license="CC0 + CC-BY-NC", provenance="GLEIF golden-copy + OpenSanctions", + realism=L.REAL_SNAPSHOT.value, datasources=("GLEIF LEI", "OpenSanctions"), + ground_truth_available=True, supported_scenarios=("kyc-go-no-go",), + capability_requirements=("kyc.screen",)), + WorldDescriptor("geo-zoning", "us-parcels", "Geospatial / zoning intelligence", + provenance="government ArcGIS / GIS portals", realism=L.REAL_SNAPSHOT.value, + datasources=("harvested_us_sample (32 parcels)",), + supported_scenarios=("zoning-permit",), capability_requirements=("geo.resolve",)), + WorldDescriptor("security-telemetry", "crowdsec", "Compromised agent / unsafe tool call", + provenance="CrowdSec LAPI", realism=L.REAL_LIVE.value, + datasources=("CrowdSec decisions/alerts",), ground_truth_available=True, + supported_scenarios=("compromised-agent",), + capability_requirements=("security.contain",)), + WorldDescriptor("agent-trajectories", "tau-bench", "Governance over recorded agent runs", + license="MIT", provenance="sierra-research/tau-bench (gpt-4o-retail)", + realism=L.REAL_SNAPSHOT.value, datasources=("τ-bench retail trajectories",), + ground_truth_available=True, supported_scenarios=("governance-correlation",)), + ]: + r.register(d) + return r diff --git a/runtime_contracts/world/trace.py b/runtime_contracts/world/trace.py new file mode 100644 index 0000000..f33c1d2 --- /dev/null +++ b/runtime_contracts/world/trace.py @@ -0,0 +1,96 @@ +"""Visual-trace schema — one mission trace that streams to the animated execution canvas. + +The canvas shows a business event (left) → the Runtime spine (center) → the four business blocks (right), +with app nodes lighting up as the mission crosses them and a **capsule** carrying identity / evidence / +policy / authority / context moving *with* the mission rather than being copied between apps. This module +fixes the shape of that stream so any runtime can emit it and any canvas can render it. It is a presentation +projection of the real Mission trace — not a second source of truth. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, List, Optional, Tuple + + +class BusinessBlock(str, Enum): + RUNTIME = "runtime" # the spine (Discovery/Planner/Mission/Context/Verification) + CUSTOMER_SUCCESS = "customer_success" + REVENUE = "revenue_intelligence" + FINANCE = "finance" + SECURITY = "security_compliance" + PLATFORM_OPS = "platform_operations" + + +class MilestoneKind(str, Enum): + EVENT = "event"; DISCOVERY = "discovery"; PLAN = "plan"; POLICY = "policy" + APPROVAL = "approval"; NEEDS_YOU = "needs_you"; ACTION = "action"; VERIFY = "verify"; OUTCOME = "outcome" + + +#: NeedsYou reasons — a human-resolvable interruption is never a generic "mission failed" (v2 §26). +class NeedsYouReason(str, Enum): + AMBIGUOUS_INTENT = "AMBIGUOUS_INTENT"; MISSING_EVIDENCE = "MISSING_EVIDENCE" + AUTHORITY_REQUIRED = "AUTHORITY_REQUIRED"; POLICY_APPROVAL = "POLICY_APPROVAL" + CONFLICTING_EVIDENCE = "CONFLICTING_EVIDENCE"; CAPABILITY_UNAVAILABLE = "CAPABILITY_UNAVAILABLE" + OUT_OF_DISTRIBUTION = "OUT_OF_DISTRIBUTION"; HIGH_RISK = "HIGH_RISK"; BUDGET_EXCEEDED = "BUDGET_EXCEEDED" + + +@dataclass(frozen=True) +class Capsule: + """The identity/evidence/policy context that moves with the mission along canvas edges — references + only, never copies (the anti-pattern the whole design replaces is copying data between apps).""" + mission_id: str = "" + entity_id: str = "" + evidence_hash: str = "" + policy_ref: str = "" + authority_ref: str = "" + context_epoch: str = "" + + def canonical_form(self) -> Dict[str, Any]: + return {k: v for k, v in { + "mission_id": self.mission_id, "entity_id": self.entity_id, "evidence_hash": self.evidence_hash, + "policy_ref": self.policy_ref, "authority_ref": self.authority_ref, + "context_epoch": self.context_epoch}.items() if v} + + +@dataclass(frozen=True) +class TraceMilestone: + """One visible step on the canvas — its offset, what happened, which app/runtime node and block, and the + capsule state at that moment. ``needs_you`` names a human-resolvable interruption reason.""" + t_offset_s: float + label: str + kind: str = MilestoneKind.EVENT.value + node: str = "" # app/runtime node id the edge enters (e.g. "crm", "finance") + block: str = BusinessBlock.RUNTIME.value + capsule: Optional[Capsule] = None + needs_you: str = "" # a NeedsYouReason value, when kind == needs_you + realism: str = "" # RealismClass label to badge this step's data + + def canonical_form(self) -> Dict[str, Any]: + d: Dict[str, Any] = {"t_offset_s": self.t_offset_s, "label": self.label, "kind": self.kind, + "block": self.block} + if self.node: + d["node"] = self.node + if self.capsule is not None: + d["capsule"] = self.capsule.canonical_form() + if self.needs_you: + d["needs_you"] = self.needs_you + if self.realism: + d["realism"] = self.realism + return d + + +@dataclass +class VisualTrace: + """The ordered milestone stream for one mission over one world — what the canvas animates.""" + mission_id: str + world_id: str = "" + milestones: List[TraceMilestone] = field(default_factory=list) + + def add(self, milestone: TraceMilestone) -> "VisualTrace": + self.milestones.append(milestone) + return self + + def canonical_form(self) -> Dict[str, Any]: + return {"mission_id": self.mission_id, "world_id": self.world_id, + "milestones": [m.canonical_form() for m in sorted(self.milestones, key=lambda x: x.t_offset_s)]} diff --git a/tests/test_world_contract.py b/tests/test_world_contract.py new file mode 100644 index 0000000..fcc00d1 --- /dev/null +++ b/tests/test_world_contract.py @@ -0,0 +1,107 @@ +"""P0 dataset-world contract — WorldEvent, IdentityGraph, WorldRegistry, VisualTrace. + +Exit criterion (docx Phase 0): one source record retains identity + provenance through +Discovery → Mission → Context → app capability, and the same canonical entity resolves to every app. +""" +from __future__ import annotations + +from runtime_contracts import ( + BusinessBlock, + Capsule, + EntityKind, + EntityRef, + GroundTruth, + IdentityGraph, + NeedsYouReason, + RealismClass, + TraceMilestone, + VisualTrace, + WorldEvent, + default_registry, +) +from runtime_contracts.protocol.evidence import EvidenceRef +from runtime_contracts.world.event import LEAD_RECEIVED + + +def _lead_event(): + return WorldEvent( + world_id="after-hours-lead", dataset_id="home-services", source_record_id="call-8842", + event_type=LEAD_RECEIVED, + entity_ids=(EntityRef("cust-acme", EntityKind.CUSTOMER.value, "Acme Roofing"), + EntityRef("prop-114-elm", EntityKind.PROPERTY.value, "114 Elm St")), + observed_at="2026-08-25T21:47:00Z", effective_at="2026-08-25T21:47:00Z", + evidence_refs=(EvidenceRef(ref="transcript-8842", content_hash="rcv1:aa", source="voice"),), + classification=RealismClass.SYNTHETIC.value, data_classifications=("pii",), + permissions=("crm.read",), tenant="world:after-hours-lead", + ground_truth=GroundTruth(situation="qualified after-hours lead", + target_outcome="governed quote during the interaction"), + capability_requirements=("crm.read", "geo.resolve", "pricing.quote"), + scenario_seed="seed-42", + payload={"caller": "Acme", "service": "roof repair", "property": "114 Elm St"}) + + +def test_world_event_is_content_addressed_and_stable(): + e = _lead_event() + assert e.content_hash.startswith("rcv1:") + assert e.identity().startswith("rcv1:") + # identity excludes ingest-time + optimizer hints: same event learned later / with hints has same id + from dataclasses import replace + assert e.identity() == replace(e, known_at="2030-01-01", latency_ms_hint=999, freshness_s=5).identity() + # a different source payload => a different content hash + assert e.content_hash != replace(e, content_hash="", payload={"caller": "Other"}).content_hash + + +def test_carried_entities_and_realism(): + e = _lead_event() + assert e.entity(EntityKind.CUSTOMER.value).entity_id == "cust-acme" + assert e.entity(EntityKind.PROPERTY.value).label == "114 Elm St" + assert e.realism() is RealismClass.SYNTHETIC + + +def test_identity_preserved_across_the_stack_and_every_app(): + """The exit criterion: the canonical entity keeps identity from the event through to each app node.""" + e = _lead_event() + cust = e.entity(EntityKind.CUSTOMER.value) + g = IdentityGraph() + # a projection seeder created a real Twenty record; ERPNext is derived (no seeded record) + g.register(cust.entity_id, "twenty", "twenty-company-991") + twenty_id = g.resolve(cust.entity_id, "twenty") + erpnext_id = g.resolve(cust.entity_id, "erpnext") + chatwoot_id = g.resolve(cust.entity_id, "chatwoot") + assert twenty_id == "twenty-company-991" # registered wins + assert erpnext_id.startswith("erpnext-") and chatwoot_id.startswith("chatwoot-") + # deterministic: same entity+app always derives the same id (replay-stable) + assert erpnext_id == g.resolve(cust.entity_id, "erpnext") + # reverse: an app-native id maps back to the canonical entity, never a new object + assert g.reverse("twenty", "twenty-company-991") == cust.entity_id + assert g.reverse("twenty", "unknown") is None + # provenance holds: the event identity is unchanged regardless of which app projection we resolved + assert e.identity() == _lead_event().identity() + + +def test_world_registry_catalogs_real_sources_with_realism(): + r = default_registry() + ids = {w.world_id for w in r.list()} + assert {"kyc-ownership", "security-telemetry", "geo-zoning", "finance-evidence"} <= ids + kyc = r.get("kyc-ownership") + assert kyc.realism == RealismClass.REAL_SNAPSHOT.value and kyc.ground_truth_available + assert "GLEIF LEI" in kyc.datasources + assert r.get("finance-evidence").realism == RealismClass.REAL_LIVE.value + + +def test_visual_trace_carries_the_capsule_and_needs_you(): + cap = Capsule(mission_id="M-1", entity_id="cust-acme", evidence_hash="rcv1:aa", + policy_ref="pol:quote/v1", authority_ref="chain:ctx1") + vt = VisualTrace(mission_id="M-1", world_id="after-hours-lead") + vt.add(TraceMilestone(0.0, "lead arrives", node="intake", block=BusinessBlock.REVENUE.value, + realism=RealismClass.SYNTHETIC.value)) + vt.add(TraceMilestone(10.0, "roof attribute missing — ask caller", kind="needs_you", + needs_you=NeedsYouReason.MISSING_EVIDENCE.value, capsule=cap, + block=BusinessBlock.RUNTIME.value)) + vt.add(TraceMilestone(13.0, "quote created", kind="action", node="pricing", + block=BusinessBlock.FINANCE.value, capsule=cap)) + cf = vt.canonical_form() + assert [m["t_offset_s"] for m in cf["milestones"]] == [0.0, 10.0, 13.0] # ordered + ny = cf["milestones"][1] + assert ny["needs_you"] == "MISSING_EVIDENCE" and ny["capsule"]["entity_id"] == "cust-acme" + assert ny["capsule"]["evidence_hash"] == "rcv1:aa" # references, not copies