From e702567f6a599fcf4b70c86c0f005fd39daf339e Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Sun, 28 Jun 2026 19:13:20 -0400 Subject: [PATCH 01/12] spec+engine: AI-driven-development autonomy ladder (2026) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Canonical contract and runtime engine for the 2026 reframing of Gartner's "Applying AI to the Development Process": the build-AI ladder becomes the platform stack, the AI-autonomy ladder becomes the governed Agent Choir, and a third axis (governed autonomy) gates every rung through the trust kernel. - specs/ai-driven-development.yaml — the canonical L0..L5 ladder, build-ladder -> layer map, per-level gate + enforced_at repo, and IBM-PE layering analogy. - autonomy.py — validator + fail-closed gate engine. evaluate_autonomy() authorizes by role ceiling then admits by evidence, demoting toward L0. - cli: validate-ai-driven-development, evaluate-autonomy. - 6 tests; full suite passes. --- specs/ai-driven-development.yaml | 184 +++++++++++++++++ src/prophet_mesh/autonomy.py | 328 +++++++++++++++++++++++++++++++ src/prophet_mesh/cli.py | 39 ++++ tests/test_autonomy.py | 78 ++++++++ 4 files changed, 629 insertions(+) create mode 100644 specs/ai-driven-development.yaml create mode 100644 src/prophet_mesh/autonomy.py create mode 100644 tests/test_autonomy.py diff --git a/specs/ai-driven-development.yaml b/specs/ai-driven-development.yaml new file mode 100644 index 0000000..c3c476c --- /dev/null +++ b/specs/ai-driven-development.yaml @@ -0,0 +1,184 @@ +name: prophet-mesh-ai-driven-development +version: 0.1.0 +product: Prophet Mesh +primary_repo: SocioProphet/prophet-mesh +primary_conductor: Michael Agent +purpose: > + 2026 reframing of the Gartner (2018) "Applying AI to the Development Process" + model and the IBM PE Developer Edition layered architecture, expressed as a + canonical prophet-stack contract. Gartner split the field into a left + "build-AI" ladder and a right "AI-does-the-work" autonomy ladder. In 2026 + those two ladders merge: the build-AI ladder is the prophet platform stack, + the AI-does-the-work ladder is the Agent Choir, and the prophet + differentiator is a third axis Gartner never had -- governed autonomy, where + every rung of autonomy is admitted only through the trust kernel. + +provenance: + source_frameworks: + - id: gartner-2018-374252 + title: Applying Artificial Intelligence to the Development Process + note: two-ladder taxonomy (build-AI ladder + AI-autonomy ladder), modernized here for 2026 + - id: ibm-pe-developer-edition + title: IBM Parallel Environment Developer Edition (Fig. 1) + note: local-workbench-over-remote-runtime layered model, used as the layering template + +invariants: + - autonomy_is_governed_not_granted + - every_autonomy_level_has_an_explicit_gate + - no_trust_kernel_bypass + - build_ladder_is_the_platform_stack + - work_ladder_is_the_agent_choir + - verified_compute_is_a_distinct_arm + - evidence_required_scales_with_autonomy + - demotion_is_always_available + +# --------------------------------------------------------------------------- +# AXIS 1 -- the build-AI ladder (Gartner LEFT column), mapped to prophet layers. +# 2018 named vendor categories; 2026 these are owned layers of our own stack. +# --------------------------------------------------------------------------- +build_ladder: + - rung: ai_infrastructure + gartner_2018: "AWS Lambda, IBM Cloud Functions, Google Cloud Platform, Microsoft Azure AI" + prophet_2026: + owner: prophet-platform + realized_by: + - kubernetes_kustomize_argocd # infra/ + - gpu_serving_substrate # GCP L4 reasoning lane + - vector_store_embed_sidecar # Noetica/embed-sidecar (Rust) + - rung: ai_frameworks + gartner_2018: "TensorFlow, Torch, Theano, Apache MXNet, Deeplearning4j, scikit-learn" + prophet_2026: + owner: tritfabric + realized_by: + - model_serving_ray # atlas/serve + - peft_distill_merge # slate/{peft,distill,merge} + - onnx_roundtrip_validation # promotion-gate substrate + - rung: ai_platforms + gartner_2018: "Amazon ML, Microsoft Azure ML, Google Cloud ML Engine, IBM Watson" + prophet_2026: + owner: tritfabric + realized_by: + - atlas_orchestration_daemon # cmd/atlasd + - drf_fairness_scheduler # atlas/scheduler + - eval_fabric # prophet-platform/apps/eval-fabric-api + - rung: ai_services + gartner_2018: "Vision/Image, Speech/Language, Conversation, Knowledge, Empathy, Custom Models" + prophet_2026: + owner: noetica + realized_by: + - governed_chat_surface # Noetica app/ + - mcp_servers # Noetica/components/mcp + - single_model_router_interface # specs/model-router-interface.yaml + - choir_as_a_service # specs/agent-choir.yaml + +# --------------------------------------------------------------------------- +# AXIS 2 -- the AI-autonomy ladder (Gartner RIGHT column), reframed as the +# Agent Choir on an explicit autonomy ladder. The 2018 "virtual roles" became +# actual choir agents. Each level names the gate that must clear to operate. +# --------------------------------------------------------------------------- +autonomy_ladder: + - level: L0 + label: manual + gartner_2018: "(human-only baseline)" + description: human performs the task; AI not in the loop + choir_roles: [] + gate: none + enforced_at: noetica + evidence_required: none + + - level: L1 + label: assisted + gartner_2018: "Assisted Development and Testing -- Virtual Intern, Virtual QA Tester" + description: AI suggests/drafts inline; human drives and commits every step + choir_roles: [coding, writing] + gate: surface_disclosure # operator sees it is AI-assisted; governance trail logged + enforced_at: noetica + evidence_required: trail_log + + - level: L2 + label: automated_unit + gartner_2018: "Automated Testing and Code Generation -- Virtual Software Engineer" + description: AI completes a bounded unit of work (a function, a test, a query) for human review + choir_roles: [coding, analytics] + gate: tests_or_review # coding_routes_require_tests_or_review (model-task-policy) + enforced_at: prophet-platform # eval-fabric replay + svf-validate-change + evidence_required: test_result_or_review_receipt + + - level: L3 + label: automated_design + gartner_2018: "Automated Architecture and Data Science -- Virtual Architect, Virtual Data Scientist" + description: AI produces a multi-step design / analysis artifact (plan, schema, model) under review + choir_roles: [planning, research, analytics] + gate: evidence_grounded # research_routes_require_evidence; verified-compute arm where applicable + enforced_at: prophet-platform # trust-chain admission + evidence-receipts + evidence_required: evidence_dossier + + - level: L4 + label: automated_solution + gartner_2018: "Automated Business Modeling and Solution Development -- Virtual Business Analyst, Virtual Solution Architect" + description: conductor orchestrates the choir end-to-end to deliver a solution; human approves outcomes + choir_roles: [conductor, planning, research, coding, writing, analytics, operations] + gate: channel_governed # channel-governed runtime gate + approval envelope + enforced_at: prophet-platform # channel-governance + conductor-response envelope + evidence_required: conductor_response_envelope + + - level: L5 + label: autonomous_governed + gartner_2018: "(beyond 2018 frame)" + description: standing autonomous operation within a chartered envelope; safe-by-default, dry-run first + choir_roles: [conductor, operations, governance-sentinel] + gate: chartered_envelope # explicit charter + revocation + continuous attestation + enforced_at: tritfabric # atlas/policy SHACL fail-closed + autopilot + evidence_required: continuous_attestation_with_revocation + +# --------------------------------------------------------------------------- +# AXIS 3 -- the prophet differentiator Gartner did not have. Every rung of the +# autonomy ladder is admitted only through the trust kernel, in order. +# --------------------------------------------------------------------------- +governed_autonomy_axis: + trust_kernel_gate_order: + - identity + - policy + - evidence + - attestation + - revocation + - audit + principle: > + Autonomy is not a capability that is granted once; it is a level that must + be re-admitted per task. A choir role may operate at autonomy level N only + if the trust-kernel gate for level N clears. Failure is fail-closed and + demotes to the highest level whose gate still clears. + distinct_arm: + name: verified_compute + note: > + decorrelated from the generative arms; provides the evidence that lets + L3+ design/solution work be trusted rather than merely produced. This is + the empty industry column the prophet stack occupies. + +# --------------------------------------------------------------------------- +# Where each concern lives, so this contract stays consistent with the stack. +# --------------------------------------------------------------------------- +repo_responsibilities: + noetica: + role: developer-edition workbench (Gartner "AI Services" rung + operator-facing autonomy control) + owns: [governed_chat_surface, autonomy_level_display, steering_promote_demote, governance_trail] + prophet-mesh: + role: programming-models / tools-platform layer -- defines roles, ladder, contracts + owns: [this_spec, agent_choir, model_router_interface, model_task_policy] + tritfabric: + role: runtime + promotion discipline (Gartner "AI Platforms/Frameworks" rungs; IBM POE/SCI/LoadLeveler analog) + owns: [atlas_orchestration, drf_scheduler, shacl_onnx_promotion_gates, L5_charter_enforcement] + prophet-platform: + role: remote-host runtime hub (IBM "Remote Host" analog) -- where levels are enforced and measured + owns: [eval_fabric, trust_chain_admission, channel_governance, evidence_receipts, evidence_console] + +ibm_pe_layering_analogy: + local_desktop_laptop: + ibm: "Developer Edition workbench (programming models, debugger, HPC toolkit, tools platform, base platform) over Eclipse" + prophet: "Noetica surface + agent-machine sidecar + embed-sidecar over the MCP/governance shell" + communication_scheduling_spine: + ibm: "POE Runtime, Scalable Communication Infrastructure (SCI), LoadLeveler" + prophet: "Atlas daemon (Ray serve = POE), gRPC fabric (= SCI), DRF scheduler (= LoadLeveler)" + remote_host_substrate: + ibm: "AIX, Linux Power, Linux x86" + prophet: "prophet-platform on Kubernetes (Kustomize + ArgoCD), GPU serving lane, eval-fabric data plane" diff --git a/src/prophet_mesh/autonomy.py b/src/prophet_mesh/autonomy.py new file mode 100644 index 0000000..1de0e0e --- /dev/null +++ b/src/prophet_mesh/autonomy.py @@ -0,0 +1,328 @@ +"""AI-driven-development autonomy ladder: validation + runtime gate engine. + +This is the executable counterpart to ``specs/ai-driven-development.yaml``. It +reframes the Gartner (2018) "Applying AI to the Development Process" autonomy +ladder as the governed Agent Choir, and it does two jobs: + +1. ``validate_ai_driven_development`` — structural conformance of the spec, + matching the other Prophet Mesh validators. +2. ``evaluate_autonomy`` — the fail-closed runtime decision. A choir role + requests an autonomy level; the engine first caps at the role's declared + ceiling (authorization), then admits the highest level whose evidence gate + is satisfied (admission), demoting toward L0 rather than failing open. + +Autonomy is never granted once; it is re-admitted per task through the trust +kernel. This module is the single deterministic source of that decision so +Noetica (surface), tritfabric (SHACL promotion) and prophet-platform (runtime +admission) can all defer to one contract. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +REQUIRED_INVARIANTS = frozenset( + { + "autonomy_is_governed_not_granted", + "every_autonomy_level_has_an_explicit_gate", + "no_trust_kernel_bypass", + "build_ladder_is_the_platform_stack", + "work_ladder_is_the_agent_choir", + "verified_compute_is_a_distinct_arm", + "evidence_required_scales_with_autonomy", + "demotion_is_always_available", + } +) + +REQUIRED_LEVELS = ("L0", "L1", "L2", "L3", "L4", "L5") + +TRUST_KERNEL_GATE_ORDER = ( + "identity", + "policy", + "evidence", + "attestation", + "revocation", + "audit", +) + +REQUIRED_LEVEL_FIELDS = frozenset( + {"level", "label", "description", "choir_roles", "gate", "enforced_at", "evidence_required"} +) + +REQUIRED_BUILD_RUNGS = frozenset( + {"ai_infrastructure", "ai_frameworks", "ai_platforms", "ai_services"} +) + +# Evidence tokens that satisfy no gate. +_NO_GATE = frozenset({"none", "", None}) + + +# --------------------------------------------------------------------------- # +# Spec loading + validation +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class AutonomyValidationResult: + valid: bool + errors: list[str] + + def to_dict(self) -> dict[str, Any]: + return {"valid": self.valid, "errors": self.errors} + + +def load_ai_driven_development(path: str | Path) -> dict[str, Any]: + with Path(path).open("r", encoding="utf-8") as handle: + data = yaml.safe_load(handle) or {} + if not isinstance(data, dict): + raise ValueError("ai-driven-development spec must be a YAML object") + return data + + +def _level_rank(level: str) -> int: + """L0..L5 -> 0..5; -1 if unparseable.""" + try: + return int(str(level).lstrip("Ll")) + except (ValueError, TypeError): + return -1 + + +def validate_ai_driven_development(data: dict[str, Any]) -> AutonomyValidationResult: + errors: list[str] = [] + + if data.get("name") != "prophet-mesh-ai-driven-development": + errors.append("name must be prophet-mesh-ai-driven-development") + if data.get("product") != "Prophet Mesh": + errors.append("product must be Prophet Mesh") + + invariants = data.get("invariants", []) + if not isinstance(invariants, list): + errors.append("invariants must be a list") + invariants = [] + missing_inv = REQUIRED_INVARIANTS - {v for v in invariants if isinstance(v, str)} + if missing_inv: + errors.append("missing invariants: " + ", ".join(sorted(missing_inv))) + + # --- build ladder (Gartner left column -> platform stack) --- + build = data.get("build_ladder", []) + if not isinstance(build, list) or not build: + errors.append("build_ladder must be a non-empty list") + build = [] + build_rungs = {r.get("rung") for r in build if isinstance(r, dict)} + missing_rungs = REQUIRED_BUILD_RUNGS - build_rungs + if missing_rungs: + errors.append("build_ladder missing rungs: " + ", ".join(sorted(missing_rungs))) + + # --- autonomy ladder (Gartner right column -> governed choir) --- + ladder = data.get("autonomy_ladder", []) + if not isinstance(ladder, list) or not ladder: + errors.append("autonomy_ladder must be a non-empty list") + ladder = [] + + seen_levels: list[str] = [] + for index, level in enumerate(ladder): + if not isinstance(level, dict): + errors.append(f"autonomy_ladder[{index}] must be an object") + continue + missing_fields = REQUIRED_LEVEL_FIELDS - set(level) + if missing_fields: + errors.append( + f"autonomy_ladder[{index}] missing fields: " + ", ".join(sorted(missing_fields)) + ) + name = level.get("level") + if isinstance(name, str): + seen_levels.append(name) + roles = level.get("choir_roles") + if not isinstance(roles, list): + errors.append(f"autonomy_ladder[{index}].choir_roles must be a list") + gate = level.get("gate") + # Invariant: every level above L0 must declare a real gate. + if _level_rank(name) >= 1 and (gate in _NO_GATE): + errors.append(f"autonomy_ladder[{index}] ({name}) above L0 must declare a gate") + + missing_levels = set(REQUIRED_LEVELS) - set(seen_levels) + if missing_levels: + errors.append("autonomy_ladder missing levels: " + ", ".join(sorted(missing_levels))) + + # --- governed-autonomy axis (the trust-kernel spine) --- + axis = data.get("governed_autonomy_axis", {}) + if not isinstance(axis, dict): + errors.append("governed_autonomy_axis must be an object") + axis = {} + order = axis.get("trust_kernel_gate_order", []) + if list(order) != list(TRUST_KERNEL_GATE_ORDER): + errors.append( + "trust_kernel_gate_order must be exactly " + + " -> ".join(TRUST_KERNEL_GATE_ORDER) + ) + + return AutonomyValidationResult(valid=not errors, errors=errors) + + +def validate_ai_driven_development_file(path: str | Path) -> AutonomyValidationResult: + return validate_ai_driven_development(load_ai_driven_development(path)) + + +# --------------------------------------------------------------------------- # +# Runtime autonomy gate engine +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class AutonomyLevel: + level: str + label: str + rank: int + choir_roles: tuple[str, ...] + gate: str + evidence_required: str + enforced_at: str + + +@dataclass(frozen=True) +class AutonomyDecision: + role: str + requested_level: str + role_ceiling: str + granted_level: str + granted_label: str + gate: str + evidence_required: str + enforced_at: str + demoted: bool + reason: str + trust_kernel_gate_order: tuple[str, ...] + + def to_dict(self) -> dict[str, Any]: + return { + "role": self.role, + "requested_level": self.requested_level, + "role_ceiling": self.role_ceiling, + "granted_level": self.granted_level, + "granted_label": self.granted_label, + "gate": self.gate, + "evidence_required": self.evidence_required, + "enforced_at": self.enforced_at, + "demoted": self.demoted, + "reason": self.reason, + "trust_kernel_gate_order": list(self.trust_kernel_gate_order), + } + + +class AutonomyLadder: + """In-memory ladder built from the spec, with the gate decision engine.""" + + def __init__(self, levels: list[AutonomyLevel]) -> None: + self._by_rank: dict[int, AutonomyLevel] = {lvl.rank: lvl for lvl in levels} + if 0 not in self._by_rank: + raise ValueError("autonomy ladder must define L0 (manual baseline)") + + @classmethod + def from_spec(cls, data: dict[str, Any]) -> "AutonomyLadder": + levels: list[AutonomyLevel] = [] + for entry in data.get("autonomy_ladder", []): + if not isinstance(entry, dict): + continue + name = entry.get("level") + rank = _level_rank(name) + if rank < 0: + continue + roles = entry.get("choir_roles") or [] + levels.append( + AutonomyLevel( + level=str(name), + label=str(entry.get("label", "")), + rank=rank, + choir_roles=tuple(r for r in roles if isinstance(r, str)), + gate=str(entry.get("gate", "none")), + evidence_required=str(entry.get("evidence_required", "none")), + enforced_at=str(entry.get("enforced_at", "")), + ) + ) + return cls(levels) + + @classmethod + def from_file(cls, path: str | Path) -> "AutonomyLadder": + return cls.from_spec(load_ai_driven_development(path)) + + def level(self, rank: int) -> AutonomyLevel: + return self._by_rank[rank] + + def role_ceiling(self, role: str) -> int: + """Highest rank at which ``role`` is declared in the spec (else L0).""" + ceiling = 0 + for rank, lvl in self._by_rank.items(): + if role in lvl.choir_roles and rank > ceiling: + ceiling = rank + return ceiling + + @staticmethod + def _evidence_satisfied(level: AutonomyLevel, available: set[str]) -> bool: + required = level.evidence_required + return required in _NO_GATE or required in available + + def evaluate( + self, + role: str, + requested_level: str, + available_evidence: set[str] | None = None, + ) -> AutonomyDecision: + """Authorize (role ceiling) then admit (evidence gate), failing closed. + + Returns the highest level <= min(requested, role_ceiling) whose evidence + gate is satisfied. L0 is always grantable, so a decision always exists. + """ + available = available_evidence or set() + requested_rank = _level_rank(requested_level) + if requested_rank < 0: + requested_rank = 0 + ceiling = self.role_ceiling(role) + capped = min(requested_rank, ceiling) + + reasons: list[str] = [] + if requested_rank > ceiling: + reasons.append( + f"role {role!r} not authorized above L{ceiling}; capped from L{requested_rank}" + ) + + granted_rank = 0 + for rank in range(capped, -1, -1): + candidate = self._by_rank.get(rank) + if candidate is None: + continue + if self._evidence_satisfied(candidate, available): + granted_rank = rank + break + reasons.append( + f"L{rank} gate '{candidate.gate}' needs evidence " + f"'{candidate.evidence_required}' (absent) -> demote" + ) + + granted = self._by_rank[granted_rank] + demoted = granted_rank < requested_rank + if not demoted and not reasons: + reasons.append(f"granted at requested level {granted.level}") + + return AutonomyDecision( + role=role, + requested_level=f"L{requested_rank}", + role_ceiling=f"L{ceiling}", + granted_level=granted.level, + granted_label=granted.label, + gate=granted.gate, + evidence_required=granted.evidence_required, + enforced_at=granted.enforced_at, + demoted=demoted, + reason="; ".join(reasons), + trust_kernel_gate_order=TRUST_KERNEL_GATE_ORDER, + ) + + +def evaluate_autonomy_file( + spec_path: str | Path, + role: str, + requested_level: str, + available_evidence: set[str] | None = None, +) -> AutonomyDecision: + ladder = AutonomyLadder.from_file(spec_path) + return ladder.evaluate(role, requested_level, available_evidence) diff --git a/src/prophet_mesh/cli.py b/src/prophet_mesh/cli.py index c19318e..15a385d 100644 --- a/src/prophet_mesh/cli.py +++ b/src/prophet_mesh/cli.py @@ -7,6 +7,10 @@ from pathlib import Path from prophet_mesh.agent_registry import validate_agent_registry_path +from prophet_mesh.autonomy import ( + evaluate_autonomy_file, + validate_ai_driven_development_file, +) from prophet_mesh.choir import validate_choir_spec_file from prophet_mesh.choir_plan import validate_choir_plan_file from prophet_mesh.conductor_response import validate_conductor_response_file @@ -127,6 +131,20 @@ def _cmd_validate_model_policy(args: argparse.Namespace) -> int: return 0 if result.valid else 1 +def _cmd_validate_ai_driven_development(args: argparse.Namespace) -> int: + result = validate_ai_driven_development_file(Path(args.path)) + print(json.dumps(result.to_dict(), indent=2)) + return 0 if result.valid else 1 + + +def _cmd_evaluate_autonomy(args: argparse.Namespace) -> int: + evidence = {token for token in (args.evidence or "").split(",") if token.strip()} + decision = evaluate_autonomy_file(Path(args.spec), args.role, args.level, evidence) + print(json.dumps(decision.to_dict(), indent=2)) + # Fail-closed: non-zero exit when the requested level could not be granted. + return 0 if not decision.demoted else 1 + + def _cmd_validate_router_decision(args: argparse.Namespace) -> int: result = validate_router_decision_file(Path(args.path)) print(json.dumps(result.to_dict(), indent=2)) @@ -227,6 +245,27 @@ def build_parser() -> argparse.ArgumentParser: validate_model_policy.add_argument("path") validate_model_policy.set_defaults(func=_cmd_validate_model_policy) + validate_ai_dev = subcommands.add_parser( + "validate-ai-driven-development", + help="validate the Prophet Mesh AI-driven-development autonomy spec", + ) + validate_ai_dev.add_argument("path", nargs="?", default="specs/ai-driven-development.yaml") + validate_ai_dev.set_defaults(func=_cmd_validate_ai_driven_development) + + evaluate_autonomy = subcommands.add_parser( + "evaluate-autonomy", + help="evaluate a fail-closed autonomy decision for a choir role at a requested level", + ) + evaluate_autonomy.add_argument("--role", required=True, help="choir role requesting autonomy") + evaluate_autonomy.add_argument("--level", required=True, help="requested autonomy level, e.g. L4") + evaluate_autonomy.add_argument( + "--evidence", default="", help="comma-separated evidence tokens present at decision time" + ) + evaluate_autonomy.add_argument( + "--spec", default="specs/ai-driven-development.yaml", help="autonomy spec path" + ) + evaluate_autonomy.set_defaults(func=_cmd_evaluate_autonomy) + validate_router_decision = subcommands.add_parser("validate-router-decision", help="validate a Prophet Mesh router decision artifact") validate_router_decision.add_argument("path") validate_router_decision.set_defaults(func=_cmd_validate_router_decision) diff --git a/tests/test_autonomy.py b/tests/test_autonomy.py new file mode 100644 index 0000000..c891e0e --- /dev/null +++ b/tests/test_autonomy.py @@ -0,0 +1,78 @@ +"""Tests for the AI-driven-development autonomy ladder + gate engine.""" + +from __future__ import annotations + +from pathlib import Path + +from prophet_mesh.autonomy import ( + AutonomyLadder, + validate_ai_driven_development_file, +) + +SPEC = Path("specs/ai-driven-development.yaml") + + +def test_spec_is_valid() -> None: + result = validate_ai_driven_development_file(SPEC) + assert result.valid, result.errors + + +def test_l0_is_always_grantable_without_evidence() -> None: + ladder = AutonomyLadder.from_file(SPEC) + decision = ladder.evaluate(role="anyone", requested_level="L0", available_evidence=set()) + assert decision.granted_level == "L0" + assert decision.demoted is False + + +def test_full_evidence_grants_requested_level() -> None: + ladder = AutonomyLadder.from_file(SPEC) + decision = ladder.evaluate( + role="conductor", + requested_level="L4", + available_evidence={"conductor_response_envelope"}, + ) + assert decision.granted_level == "L4" + assert decision.demoted is False + + +def test_missing_evidence_demotes_fail_closed() -> None: + ladder = AutonomyLadder.from_file(SPEC) + decision = ladder.evaluate( + role="coding", + requested_level="L2", + available_evidence=set(), # no test/review receipt + ) + # L2 needs a test/review receipt; without it we must not grant L2. + assert decision.granted_level != "L2" + assert decision.demoted is True + assert "demote" in decision.reason + + +def test_role_ceiling_caps_authorization() -> None: + ladder = AutonomyLadder.from_file(SPEC) + # 'writing' is in the full L4 choir but never declared at L5; it cannot + # reach standing autonomous operation. + decision = ladder.evaluate( + role="writing", + requested_level="L5", + available_evidence={"continuous_attestation_with_revocation"}, + ) + assert decision.role_ceiling == "L4" + assert decision.granted_level != "L5" + assert "not authorized" in decision.reason + + +def test_unknown_role_floors_at_l0() -> None: + ladder = AutonomyLadder.from_file(SPEC) + decision = ladder.evaluate(role="memory-steward", requested_level="L3") + assert decision.role_ceiling == "L0" + assert decision.granted_level == "L0" + + +def test_l1_admits_with_disclosure_trail() -> None: + ladder = AutonomyLadder.from_file(SPEC) + decision = ladder.evaluate( + role="writing", requested_level="L1", available_evidence={"trail_log"} + ) + assert decision.granted_level == "L1" + assert decision.demoted is False From 0c8018c31479bbe69745fab9dfd6e945a7957059 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Sun, 28 Jun 2026 19:17:11 -0400 Subject: [PATCH 02/12] autonomy: canonical ladder export (single source of truth) Add export-autonomy-ladder CLI + canonical_ladder() that derives a machine-readable ladder (specs/ai-driven-development.ladder.json) from the spec. This is the single source downstream repos (tritfabric, prophet-platform, Noetica) derive from / verify against instead of hand-mirroring the YAML. A drift test asserts the committed export matches the spec. --- specs/ai-driven-development.ladder.json | 91 +++++++++++++++++++++++++ src/prophet_mesh/autonomy.py | 44 ++++++++++++ src/prophet_mesh/cli.py | 31 +++++++++ tests/test_autonomy.py | 14 ++++ 4 files changed, 180 insertions(+) create mode 100644 specs/ai-driven-development.ladder.json diff --git a/specs/ai-driven-development.ladder.json b/specs/ai-driven-development.ladder.json new file mode 100644 index 0000000..178115c --- /dev/null +++ b/specs/ai-driven-development.ladder.json @@ -0,0 +1,91 @@ +{ + "levels": [ + { + "enforced_at": "noetica", + "evidence_required": "none", + "gate": "none", + "label": "manual", + "level": "L0", + "rank": 0, + "roles": [] + }, + { + "enforced_at": "noetica", + "evidence_required": "trail_log", + "gate": "surface_disclosure", + "label": "assisted", + "level": "L1", + "rank": 1, + "roles": [ + "coding", + "writing" + ] + }, + { + "enforced_at": "prophet-platform", + "evidence_required": "test_result_or_review_receipt", + "gate": "tests_or_review", + "label": "automated_unit", + "level": "L2", + "rank": 2, + "roles": [ + "coding", + "analytics" + ] + }, + { + "enforced_at": "prophet-platform", + "evidence_required": "evidence_dossier", + "gate": "evidence_grounded", + "label": "automated_design", + "level": "L3", + "rank": 3, + "roles": [ + "planning", + "research", + "analytics" + ] + }, + { + "enforced_at": "prophet-platform", + "evidence_required": "conductor_response_envelope", + "gate": "channel_governed", + "label": "automated_solution", + "level": "L4", + "rank": 4, + "roles": [ + "conductor", + "planning", + "research", + "coding", + "writing", + "analytics", + "operations" + ] + }, + { + "enforced_at": "tritfabric", + "evidence_required": "continuous_attestation_with_revocation", + "gate": "chartered_envelope", + "label": "autonomous_governed", + "level": "L5", + "rank": 5, + "roles": [ + "conductor", + "operations", + "governance-sentinel" + ] + } + ], + "name": "prophet-mesh-ai-driven-development", + "source": "SocioProphet/prophet-mesh:specs/ai-driven-development.yaml", + "trust_kernel_gate_order": [ + "identity", + "policy", + "evidence", + "attestation", + "revocation", + "audit" + ], + "version": "0.1.0" +} diff --git a/src/prophet_mesh/autonomy.py b/src/prophet_mesh/autonomy.py index 1de0e0e..3d6f236 100644 --- a/src/prophet_mesh/autonomy.py +++ b/src/prophet_mesh/autonomy.py @@ -248,6 +248,10 @@ def from_file(cls, path: str | Path) -> "AutonomyLadder": def level(self, rank: int) -> AutonomyLevel: return self._by_rank[rank] + def ordered_levels(self) -> list[AutonomyLevel]: + """Levels low-to-high by rank.""" + return [self._by_rank[rank] for rank in sorted(self._by_rank)] + def role_ceiling(self, role: str) -> int: """Highest rank at which ``role`` is declared in the spec (else L0).""" ceiling = 0 @@ -326,3 +330,43 @@ def evaluate_autonomy_file( ) -> AutonomyDecision: ladder = AutonomyLadder.from_file(spec_path) return ladder.evaluate(role, requested_level, available_evidence) + + +# --------------------------------------------------------------------------- # +# Canonical export — the single source of truth for downstream repos +# --------------------------------------------------------------------------- # +CANONICAL_LADDER_DEFAULT_OUT = "specs/ai-driven-development.ladder.json" + + +def canonical_ladder(data: dict[str, Any]) -> dict[str, Any]: + """Derive the machine-readable ladder that downstream repos consume. + + This is the single source of truth: tritfabric's gate config, the + prophet-platform runtime engine, and the Noetica surface all derive from / + verify against this export rather than hand-mirroring the YAML. + """ + ladder = AutonomyLadder.from_spec(data) + axis = data.get("governed_autonomy_axis", {}) or {} + order = axis.get("trust_kernel_gate_order") or list(TRUST_KERNEL_GATE_ORDER) + return { + "name": data.get("name"), + "version": data.get("version"), + "source": "SocioProphet/prophet-mesh:specs/ai-driven-development.yaml", + "trust_kernel_gate_order": list(order), + "levels": [ + { + "level": lvl.level, + "rank": lvl.rank, + "label": lvl.label, + "roles": list(lvl.choir_roles), + "gate": lvl.gate, + "evidence_required": lvl.evidence_required, + "enforced_at": lvl.enforced_at, + } + for lvl in ladder.ordered_levels() + ], + } + + +def canonical_ladder_file(spec_path: str | Path) -> dict[str, Any]: + return canonical_ladder(load_ai_driven_development(spec_path)) diff --git a/src/prophet_mesh/cli.py b/src/prophet_mesh/cli.py index 15a385d..b0fa72f 100644 --- a/src/prophet_mesh/cli.py +++ b/src/prophet_mesh/cli.py @@ -8,6 +8,8 @@ from prophet_mesh.agent_registry import validate_agent_registry_path from prophet_mesh.autonomy import ( + CANONICAL_LADDER_DEFAULT_OUT, + canonical_ladder_file, evaluate_autonomy_file, validate_ai_driven_development_file, ) @@ -137,6 +139,19 @@ def _cmd_validate_ai_driven_development(args: argparse.Namespace) -> int: return 0 if result.valid else 1 +def _cmd_export_autonomy_ladder(args: argparse.Namespace) -> int: + ladder = canonical_ladder_file(Path(args.spec)) + payload = json.dumps(ladder, indent=2, sort_keys=True) + "\n" + if args.out: + out_path = Path(args.out) + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_text(payload, encoding="utf-8") + print(json.dumps({"ok": True, "out": str(out_path), "levels": len(ladder["levels"])})) + else: + print(payload, end="") + return 0 + + def _cmd_evaluate_autonomy(args: argparse.Namespace) -> int: evidence = {token for token in (args.evidence or "").split(",") if token.strip()} decision = evaluate_autonomy_file(Path(args.spec), args.role, args.level, evidence) @@ -252,6 +267,22 @@ def build_parser() -> argparse.ArgumentParser: validate_ai_dev.add_argument("path", nargs="?", default="specs/ai-driven-development.yaml") validate_ai_dev.set_defaults(func=_cmd_validate_ai_driven_development) + export_ladder = subcommands.add_parser( + "export-autonomy-ladder", + help="export the canonical machine-readable autonomy ladder (single source of truth)", + ) + export_ladder.add_argument( + "--spec", default="specs/ai-driven-development.yaml", help="autonomy spec path" + ) + export_ladder.add_argument( + "--out", + nargs="?", + const=CANONICAL_LADDER_DEFAULT_OUT, + default=None, + help=f"write JSON here (bare flag -> {CANONICAL_LADDER_DEFAULT_OUT}); omit to print", + ) + export_ladder.set_defaults(func=_cmd_export_autonomy_ladder) + evaluate_autonomy = subcommands.add_parser( "evaluate-autonomy", help="evaluate a fail-closed autonomy decision for a choir role at a requested level", diff --git a/tests/test_autonomy.py b/tests/test_autonomy.py index c891e0e..0ee0ea1 100644 --- a/tests/test_autonomy.py +++ b/tests/test_autonomy.py @@ -4,12 +4,16 @@ from pathlib import Path +import json + from prophet_mesh.autonomy import ( AutonomyLadder, + canonical_ladder_file, validate_ai_driven_development_file, ) SPEC = Path("specs/ai-driven-development.yaml") +CANONICAL = Path("specs/ai-driven-development.ladder.json") def test_spec_is_valid() -> None: @@ -69,6 +73,16 @@ def test_unknown_role_floors_at_l0() -> None: assert decision.granted_level == "L0" +def test_canonical_export_is_committed_and_current() -> None: + # The committed canonical ladder must match what the spec exports, so + # downstream repos (tritfabric, prophet-platform, Noetica) never drift from + # a stale source. Regenerate with: prophet-mesh export-autonomy-ladder --out + assert CANONICAL.exists(), "run: prophet-mesh export-autonomy-ladder --out" + on_disk = json.loads(CANONICAL.read_text(encoding="utf-8")) + fresh = canonical_ladder_file(SPEC) + assert on_disk == fresh, "canonical ladder is stale; regenerate it" + + def test_l1_admits_with_disclosure_trail() -> None: ladder = AutonomyLadder.from_file(SPEC) decision = ladder.evaluate( From 595a727d955a013e3093b90f990c80ca3543d03b Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:59:00 -0400 Subject: [PATCH 03/12] spec(pkg): default personal-knowledge-graph structure (the seed to build over) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ego-centric graph a person starts with, that the Memory Steward grows. Schema over HellGraph (NodeAtom + AssertionClass{Structural/Assertion/Executable} + required provenance + SHACL), scoped by memory-scope.yaml, populated by prophet-workspace (contacts/calendar/mail/drive), and the per-person graph the mesh grounds on. - node_types: Self(anchor) · Person · Place · Organization · Thing · Interest · Event · Document · Communication · Account - edge_types: relatedTo/knows/homeTown/attended/worksAt/owns/interestedIn/ providerFor/participatedIn/authored/communicatedWith/hasAccount - external_link contract (the dashed edges): sameAs → {general_purpose, social, domain_specific, ecommerce}, direction=reference_only (privacy: read external, never leak), confidence + trust_class + provenance required - every node/edge carries provenance + epistemic_mode + P-RET/P-GEN + I-DED/IND/ABD + memory_scope + confidence (ties to the grounding-fidelity moat) - seed = Self + registered external KGs + empty typed slots - Rule #0: exam/holdout NEVER ingested (no contamination via retrieval) --- specs/personal-knowledge-graph.yaml | 115 ++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 specs/personal-knowledge-graph.yaml diff --git a/specs/personal-knowledge-graph.yaml b/specs/personal-knowledge-graph.yaml new file mode 100644 index 0000000..fa82c92 --- /dev/null +++ b/specs/personal-knowledge-graph.yaml @@ -0,0 +1,115 @@ +name: prophet-mesh-personal-knowledge-graph +version: 0.1.0 +product: Prophet Mesh +purpose: >- + The default graph structure for a person — the seed the Memory Steward builds + over. Sits on HellGraph (NodeAtom + AssertionClass + required provenance + + SHACL shapes), is scoped by memory-scope.yaml, owned by the memory-steward + agent, populated by prophet-workspace (contacts/calendar/mail/drive), and is + what the mesh grounds retrieval on per-person. NOT a new store — a schema over + HellGraph. + +invariants: + - self_is_the_single_anchor # exactly one Self node; the whole graph is ego-centric + - every_node_and_edge_has_provenance # HellGraph rule — synthetic provenance if none other + - schema_links_are_structural # type/identity scaffolding = AssertionClass.Structural + - world_claims_are_assertions # facts about the person/world = AssertionClass.Assertion + - open_world # missing assertion = unknown, not false + - external_links_are_reference_only # PKG reads external KGs; it never writes into them (privacy) + - private_by_default # default memory-scope = relationship_context:approved + - inference_is_labeled # P-RET (pointer-bound) vs P-GEN, and I-DED/IND/ABD (see grounding) + +# ── Node types (entity classes) — the vocabulary a person is built from ──────── +node_types: + Self: { class: Person, cardinality: exactly_one, role: the graph anchor (the user) } + Person: { class: Person, qualifiers: [relationship], examples: [Mom, Jamie, colleague, "Mom's dentist"] } + Place: { class: Place, examples: [Hometown, residence, workplace-location] } + Organization: { class: Organization, examples: [High school, employer, "dental practice"] } + Thing: { class: Thing, aka: Possession, examples: [Electric guitar, Acoustic guitar, devices] } + Interest: { class: Topic, aka: [Skill, Hobby, Domain], examples: [guitar, music, a profession] } + Event: { class: Event, source: calendar } + Document: { class: CreativeWork, source: [drive, notes] } + Communication: { class: Message, source: [mail, chat], note: often modeled as an edge with content } + Account: { class: ExternalIdentity, role: anchor for external-KG links, examples: [social handle, login] } + +# ── Edge types (typed relations) — solid edges inside the PKG ────────────────── +edge_types: + # family / social + relatedTo: { subtypes: [parentOf, childOf, siblingOf, spouseOf], class: Assertion } + knows: { subtypes: [friendOf, colleagueOf, acquaintanceOf], class: Assertion } + # place / org + homeTown: { from: Self, to: Place, class: Assertion } + residesIn: { to: Place, class: Assertion } + attended: { to: Organization, class: Assertion } + worksAt: { to: Organization, class: Assertion } + memberOf: { to: Organization, class: Assertion } + # provider relations (2nd-degree, e.g. Mom → dentist) + providerFor: { subtypes: [dentistOf, doctorOf, lawyerOf], class: Assertion } + # things / interests + owns: { to: Thing, class: Assertion } + uses: { to: Thing, class: Assertion } + interestedIn: { to: Interest, class: Assertion } + skilledIn: { to: Interest, class: Assertion } + # activity (workspace-sourced) + participatedIn:{ to: Event, class: Assertion } + authored: { to: Document, class: Assertion } + communicatedWith: { to: Person, carries: [Communication], class: Assertion } + # identity scaffolding + hasAccount: { from: [Self, Person], to: Account, class: Structural } + +# ── External links (the dashed edges) — the cross-KG grounding contract ──────── +# A PKG node may point at an entity in an external graph. One-way by invariant: +# the mesh RESOLVES/reads the external entity for grounding; nothing private +# crosses outward. Every external link is an Assertion with full provenance. +external_link: + edge: sameAs # or externalRef + class: Assertion + direction: reference_only # privacy: read external, never write + required_fields: + - target_kg # which external graph (see external_kgs) + - target_id # the external entity id (e.g. Wikidata Q-id) + - confidence # entity-resolution confidence 0..1 + - trust_class # source trust (mirrors HellGraph source_trust_class) + - provenance # required +external_kgs: + general_purpose: { role: entity resolution / world facts, examples: [Wikidata, KBpedia/KKO], links_from: [Place, Organization, Interest] } + social_network: { role: people + relationships, links_from: [Person, Account], privacy: high } + domain_specific: { role: domain KGs (music, medical, legal, finance), links_from: [Interest, Thing] } + ecommerce_catalog:{ role: products / listings, links_from: [Thing] } + +# ── Provenance + scope carried by EVERY node and edge ────────────────────────── +required_on_every_element: + provenance: { source, captured_at, method } # HellGraph requirement + epistemic_mode: [asserted, imported_rdf, identity] # HellGraph EpistemicMode + assertion_class: [Structural, Assertion, Executable] # HellGraph AssertionClass + provenance_tag: [P-RET, P-GEN] # pointer-bound vs generated (grounding axis) + inference_type: [I-NON, I-DED, I-IND, I-ABD] # assertive/deductive/inductive/abductive + memory_scope: relationship_context:approved # default; see memory-scope.yaml + confidence: { type: number, range: [0, 1] } + +# ── The SEED — what a brand-new person's graph is, before anything is learned ── +# A new PKG is the Self node + this schema + the external-KG link-points +# registered. prophet-workspace ingestion then fills the typed slots: +# contacts → Person + knows/relatedTo ; calendar → Event + participatedIn +# mail/chat → communicatedWith (+ Communication) ; drive/notes → Document + authored +seed: + nodes: + - { type: Self, id: "self", provenance: { source: onboarding, method: declared } } + registered_external_kgs: [general_purpose, social_network, domain_specific, ecommerce_catalog] + empty_slots: # the ontology the steward grows into, no instances yet + people: [] + places: [] + organizations: [] + things: [] + interests: [] + events: [] + documents: [] + accounts: [] + +governance: + owner_agent: memory-steward + scope_policy: memory-scope.yaml + substrate: hellgraph # NodeAtom + AssertionClass + SHACL shapes + grounding: >- + This graph is the per-person hellgraph the mesh grounds on. Exam/holdout + content is NEVER ingested (Rule #0 — no contamination via retrieval). From ef1608f062bc611832a06d8666e367e8081db9f9 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:59:40 -0400 Subject: [PATCH 04/12] =?UTF-8?q?fix(pkg):=20YAML=20=E2=80=94=20space=20af?= =?UTF-8?q?ter=20colon=20on=20participatedIn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- specs/personal-knowledge-graph.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/specs/personal-knowledge-graph.yaml b/specs/personal-knowledge-graph.yaml index fa82c92..700a591 100644 --- a/specs/personal-knowledge-graph.yaml +++ b/specs/personal-knowledge-graph.yaml @@ -51,7 +51,7 @@ edge_types: interestedIn: { to: Interest, class: Assertion } skilledIn: { to: Interest, class: Assertion } # activity (workspace-sourced) - participatedIn:{ to: Event, class: Assertion } + participatedIn: { to: Event, class: Assertion } authored: { to: Document, class: Assertion } communicatedWith: { to: Person, carries: [Communication], class: Assertion } # identity scaffolding From f583b0388918982ddd40e7e9f954cf09d887fb4c Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:00:35 -0400 Subject: [PATCH 05/12] =?UTF-8?q?fix(pkg):=20rewrite=20in=20block=20style?= =?UTF-8?q?=20=E2=80=94=20prose-in-flow-mappings=20was=20breaking=20YAML?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- specs/personal-knowledge-graph.yaml | 168 ++++++++++++++++------------ 1 file changed, 97 insertions(+), 71 deletions(-) diff --git a/specs/personal-knowledge-graph.yaml b/specs/personal-knowledge-graph.yaml index 700a591..b882914 100644 --- a/specs/personal-knowledge-graph.yaml +++ b/specs/personal-knowledge-graph.yaml @@ -3,100 +3,126 @@ version: 0.1.0 product: Prophet Mesh purpose: >- The default graph structure for a person — the seed the Memory Steward builds - over. Sits on HellGraph (NodeAtom + AssertionClass + required provenance + - SHACL shapes), is scoped by memory-scope.yaml, owned by the memory-steward - agent, populated by prophet-workspace (contacts/calendar/mail/drive), and is - what the mesh grounds retrieval on per-person. NOT a new store — a schema over - HellGraph. + over. A schema over HellGraph (NodeAtom + AssertionClass + required provenance + + SHACL shapes), scoped by memory-scope.yaml, owned by the memory-steward + agent, populated by prophet-workspace (contacts / calendar / mail / drive), and + the per-person graph the mesh grounds retrieval on. Not a new store — a schema. invariants: - - self_is_the_single_anchor # exactly one Self node; the whole graph is ego-centric - - every_node_and_edge_has_provenance # HellGraph rule — synthetic provenance if none other - - schema_links_are_structural # type/identity scaffolding = AssertionClass.Structural - - world_claims_are_assertions # facts about the person/world = AssertionClass.Assertion - - open_world # missing assertion = unknown, not false - - external_links_are_reference_only # PKG reads external KGs; it never writes into them (privacy) - - private_by_default # default memory-scope = relationship_context:approved - - inference_is_labeled # P-RET (pointer-bound) vs P-GEN, and I-DED/IND/ABD (see grounding) + - self_is_the_single_anchor + - every_node_and_edge_has_provenance + - schema_links_are_structural + - world_claims_are_assertions + - open_world + - external_links_are_reference_only + - private_by_default + - inference_is_labeled # ── Node types (entity classes) — the vocabulary a person is built from ──────── node_types: - Self: { class: Person, cardinality: exactly_one, role: the graph anchor (the user) } - Person: { class: Person, qualifiers: [relationship], examples: [Mom, Jamie, colleague, "Mom's dentist"] } - Place: { class: Place, examples: [Hometown, residence, workplace-location] } - Organization: { class: Organization, examples: [High school, employer, "dental practice"] } - Thing: { class: Thing, aka: Possession, examples: [Electric guitar, Acoustic guitar, devices] } - Interest: { class: Topic, aka: [Skill, Hobby, Domain], examples: [guitar, music, a profession] } - Event: { class: Event, source: calendar } - Document: { class: CreativeWork, source: [drive, notes] } - Communication: { class: Message, source: [mail, chat], note: often modeled as an edge with content } - Account: { class: ExternalIdentity, role: anchor for external-KG links, examples: [social handle, login] } + Self: + class: Person + cardinality: exactly_one + role: "the graph anchor (the user)" + Person: + class: Person + qualifiers: [relationship] + examples: ["Mom", "Jamie", "colleague", "Mom's dentist"] + Place: + class: Place + examples: ["Hometown", "residence", "workplace location"] + Organization: + class: Organization + examples: ["High school", "employer", "dental practice"] + Thing: + class: Thing + aka: Possession + examples: ["Electric guitar", "Acoustic guitar", "devices"] + Interest: + class: Topic + aka: [Skill, Hobby, Domain] + examples: ["guitar", "music", "a profession"] + Event: + class: Event + source: [calendar] + Document: + class: CreativeWork + source: [drive, notes] + Communication: + class: Message + source: [mail, chat] + note: "often modeled as an edge with content" + Account: + class: ExternalIdentity + role: "anchor for external-KG links" + examples: ["social handle", "login"] # ── Edge types (typed relations) — solid edges inside the PKG ────────────────── edge_types: - # family / social - relatedTo: { subtypes: [parentOf, childOf, siblingOf, spouseOf], class: Assertion } - knows: { subtypes: [friendOf, colleagueOf, acquaintanceOf], class: Assertion } - # place / org - homeTown: { from: Self, to: Place, class: Assertion } - residesIn: { to: Place, class: Assertion } - attended: { to: Organization, class: Assertion } - worksAt: { to: Organization, class: Assertion } - memberOf: { to: Organization, class: Assertion } - # provider relations (2nd-degree, e.g. Mom → dentist) - providerFor: { subtypes: [dentistOf, doctorOf, lawyerOf], class: Assertion } - # things / interests - owns: { to: Thing, class: Assertion } - uses: { to: Thing, class: Assertion } - interestedIn: { to: Interest, class: Assertion } - skilledIn: { to: Interest, class: Assertion } - # activity (workspace-sourced) - participatedIn: { to: Event, class: Assertion } - authored: { to: Document, class: Assertion } + relatedTo: { subtypes: [parentOf, childOf, siblingOf, spouseOf], class: Assertion } + knows: { subtypes: [friendOf, colleagueOf, acquaintanceOf], class: Assertion } + homeTown: { from: Self, to: Place, class: Assertion } + residesIn: { to: Place, class: Assertion } + attended: { to: Organization, class: Assertion } + worksAt: { to: Organization, class: Assertion } + memberOf: { to: Organization, class: Assertion } + providerFor: { subtypes: [dentistOf, doctorOf, lawyerOf], class: Assertion } + owns: { to: Thing, class: Assertion } + uses: { to: Thing, class: Assertion } + interestedIn: { to: Interest, class: Assertion } + skilledIn: { to: Interest, class: Assertion } + participatedIn: { to: Event, class: Assertion } + authored: { to: Document, class: Assertion } communicatedWith: { to: Person, carries: [Communication], class: Assertion } - # identity scaffolding - hasAccount: { from: [Self, Person], to: Account, class: Structural } + hasAccount: { from: [Self, Person], to: Account, class: Structural } # ── External links (the dashed edges) — the cross-KG grounding contract ──────── # A PKG node may point at an entity in an external graph. One-way by invariant: # the mesh RESOLVES/reads the external entity for grounding; nothing private # crosses outward. Every external link is an Assertion with full provenance. external_link: - edge: sameAs # or externalRef + edge: sameAs class: Assertion - direction: reference_only # privacy: read external, never write - required_fields: - - target_kg # which external graph (see external_kgs) - - target_id # the external entity id (e.g. Wikidata Q-id) - - confidence # entity-resolution confidence 0..1 - - trust_class # source trust (mirrors HellGraph source_trust_class) - - provenance # required + direction: reference_only + required_fields: [target_kg, target_id, confidence, trust_class, provenance] external_kgs: - general_purpose: { role: entity resolution / world facts, examples: [Wikidata, KBpedia/KKO], links_from: [Place, Organization, Interest] } - social_network: { role: people + relationships, links_from: [Person, Account], privacy: high } - domain_specific: { role: domain KGs (music, medical, legal, finance), links_from: [Interest, Thing] } - ecommerce_catalog:{ role: products / listings, links_from: [Thing] } + general_purpose: + role: "entity resolution and world facts" + examples: [Wikidata, "KBpedia/KKO"] + links_from: [Place, Organization, Interest] + social_network: + role: "people and relationships" + links_from: [Person, Account] + privacy: high + domain_specific: + role: "domain KGs — music, medical, legal, finance" + links_from: [Interest, Thing] + ecommerce_catalog: + role: "products and listings" + links_from: [Thing] # ── Provenance + scope carried by EVERY node and edge ────────────────────────── required_on_every_element: - provenance: { source, captured_at, method } # HellGraph requirement - epistemic_mode: [asserted, imported_rdf, identity] # HellGraph EpistemicMode - assertion_class: [Structural, Assertion, Executable] # HellGraph AssertionClass - provenance_tag: [P-RET, P-GEN] # pointer-bound vs generated (grounding axis) - inference_type: [I-NON, I-DED, I-IND, I-ABD] # assertive/deductive/inductive/abductive - memory_scope: relationship_context:approved # default; see memory-scope.yaml - confidence: { type: number, range: [0, 1] } + provenance: [source, captured_at, method] + epistemic_mode: [asserted, imported_rdf, identity] + assertion_class: [Structural, Assertion, Executable] + provenance_tag: [P-RET, P-GEN] + inference_type: [I-NON, I-DED, I-IND, I-ABD] + memory_scope: "relationship_context:approved" + confidence: "number in [0,1]" # ── The SEED — what a brand-new person's graph is, before anything is learned ── -# A new PKG is the Self node + this schema + the external-KG link-points -# registered. prophet-workspace ingestion then fills the typed slots: -# contacts → Person + knows/relatedTo ; calendar → Event + participatedIn -# mail/chat → communicatedWith (+ Communication) ; drive/notes → Document + authored +# Self + this schema + the external-KG link-points. prophet-workspace ingestion +# then fills the typed slots: contacts → Person + knows/relatedTo; calendar → +# Event + participatedIn; mail/chat → communicatedWith (+ Communication); +# drive/notes → Document + authored. seed: nodes: - - { type: Self, id: "self", provenance: { source: onboarding, method: declared } } + - type: Self + id: self + provenance: { source: onboarding, method: declared } registered_external_kgs: [general_purpose, social_network, domain_specific, ecommerce_catalog] - empty_slots: # the ontology the steward grows into, no instances yet + empty_slots: people: [] places: [] organizations: [] @@ -109,7 +135,7 @@ seed: governance: owner_agent: memory-steward scope_policy: memory-scope.yaml - substrate: hellgraph # NodeAtom + AssertionClass + SHACL shapes + substrate: hellgraph grounding: >- - This graph is the per-person hellgraph the mesh grounds on. Exam/holdout + This graph is the per-person hellgraph the mesh grounds on. Exam / holdout content is NEVER ingested (Rule #0 — no contamination via retrieval). From 50a886d9c91c6c670180b8a422d7be147ef0716f Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:50:11 -0400 Subject: [PATCH 06/12] feat(pkg): seed builder + workspace ingestion + SHACL-lite validator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Makes the default person-graph buildable-over, reading the spec vocabulary: - seed_graph(self_id) → the Self anchor (Structural) + registered external KGs. - ingestion: ingest_contact (family→relatedTo / social→knows), ingest_event (→participatedIn), ingest_message (→communicatedWith), ingest_document (→authored) — every element carries workspace provenance. - link_external → reference-only external-KG link (privacy: read, never leak). - validate() = SHACL-lite: node types + relations in the spec vocabulary, single Self invariant, external KG known, provenance.source required on every element. Brain-agnostic (data model; the mesh grounds on it, memory-steward owns it). 5/5 tests. --- src/prophet_mesh/pkg.py | 174 ++++++++++++++++++++++++++++++++++++++++ tests/test_pkg.py | 72 +++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 src/prophet_mesh/pkg.py create mode 100644 tests/test_pkg.py diff --git a/src/prophet_mesh/pkg.py b/src/prophet_mesh/pkg.py new file mode 100644 index 0000000..f86a842 --- /dev/null +++ b/src/prophet_mesh/pkg.py @@ -0,0 +1,174 @@ +"""Personal Knowledge Graph — the default graph a person is built over. + +Materializes the seed from specs/personal-knowledge-graph.yaml, ingests +prophet-workspace data (contacts / calendar / mail / drive) into typed nodes + +edges, and validates every element against the spec vocabulary + provenance +requirement (a SHACL-lite check). Brain-agnostic: this is the data model; the +mesh grounds on it, the Memory Steward owns it. +""" +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Any + +import yaml + +SPEC_PATH = Path("specs/personal-knowledge-graph.yaml") + + +def load_spec(path: str | Path = SPEC_PATH) -> dict[str, Any]: + return yaml.safe_load(Path(path).read_text()) + + +def valid_node_types(spec: dict) -> set[str]: + return set(spec["node_types"].keys()) + + +def valid_relations(spec: dict) -> set[str]: + """Top-level edge types + all declared subtypes.""" + rels: set[str] = set() + for name, body in spec["edge_types"].items(): + rels.add(name) + for sub in (body or {}).get("subtypes", []) or []: + rels.add(sub) + return rels + + +@dataclass(frozen=True) +class Provenance: + source: str # e.g. "onboarding", "workspace:contacts" + method: str # declared | imported | inferred | ingested + captured_at: str = "" # ISO-8601, optional + + +@dataclass(frozen=True) +class ExternalLink: + target_kg: str # general_purpose | social_network | domain_specific | ecommerce_catalog + target_id: str # e.g. a Wikidata Q-id + confidence: float = 0.0 + trust_class: str = "unverified" + + +@dataclass(frozen=True) +class Node: + id: str + type: str + label: str + provenance: Provenance + assertion_class: str = "Assertion" # Structural | Assertion | Executable + provenance_tag: str = "P-RET" # P-RET (pointer-bound) | P-GEN + inference_type: str = "I-NON" # I-NON | I-DED | I-IND | I-ABD + memory_scope: str = "relationship_context:approved" + confidence: float = 1.0 + external: tuple[ExternalLink, ...] = () + + +@dataclass(frozen=True) +class Edge: + src: str + dst: str + rel: str + provenance: Provenance + assertion_class: str = "Assertion" + provenance_tag: str = "P-RET" + inference_type: str = "I-NON" + memory_scope: str = "relationship_context:approved" + confidence: float = 1.0 + + +@dataclass +class PKG: + self_id: str + nodes: dict[str, Node] = field(default_factory=dict) + edges: list[Edge] = field(default_factory=list) + external_kgs: list[str] = field(default_factory=list) + + def add_node(self, n: Node) -> None: + self.nodes[n.id] = n + + def add_edge(self, e: Edge) -> None: + self.edges.append(e) + + +# ── Seed ────────────────────────────────────────────────────────────────────── +def seed_graph(self_id: str = "self", spec: dict | None = None) -> PKG: + """A brand-new person's graph: the Self anchor + registered external KGs.""" + spec = spec or load_spec() + g = PKG(self_id=self_id, external_kgs=list(spec["seed"]["registered_external_kgs"])) + g.add_node(Node( + id=self_id, type="Self", label="Self", + provenance=Provenance(source="onboarding", method="declared"), + assertion_class="Structural", + )) + return g + + +# ── Ingestion: prophet-workspace → typed nodes + edges ──────────────────────── +def _prov(app: str) -> Provenance: + return Provenance(source=f"workspace:{app}", method="ingested") + + +def ingest_contact(g: PKG, name: str, relationship: str | None = None) -> str: + """A contact → a Person node + an edge from Self (relatedTo family / knows social).""" + nid = f"person:{name.strip().lower().replace(' ', '_')}" + g.add_node(Node(id=nid, type="Person", label=name, provenance=_prov("contacts"))) + family = {"parentOf", "childOf", "siblingOf", "spouseOf", "mom", "dad", "mother", "father", "sister", "brother"} + rel = "relatedTo" if (relationship or "").lower() in family else "knows" + g.add_edge(Edge(src=g.self_id, dst=nid, rel=rel, provenance=_prov("contacts"))) + return nid + + +def ingest_event(g: PKG, title: str) -> str: + nid = f"event:{title.strip().lower().replace(' ', '_')}" + g.add_node(Node(id=nid, type="Event", label=title, provenance=_prov("calendar"))) + g.add_edge(Edge(src=g.self_id, dst=nid, rel="participatedIn", provenance=_prov("calendar"))) + return nid + + +def ingest_message(g: PKG, peer_node_id: str) -> None: + """A message → a communicatedWith edge (the peer Person must already exist).""" + g.add_edge(Edge(src=g.self_id, dst=peer_node_id, rel="communicatedWith", provenance=_prov("mail"))) + + +def ingest_document(g: PKG, title: str) -> str: + nid = f"doc:{title.strip().lower().replace(' ', '_')}" + g.add_node(Node(id=nid, type="Document", label=title, provenance=_prov("drive"))) + g.add_edge(Edge(src=g.self_id, dst=nid, rel="authored", provenance=_prov("drive"))) + return nid + + +def link_external(g: PKG, node_id: str, link: ExternalLink) -> None: + """Attach a reference-only external-KG link to a node (privacy: read, never leak).""" + n = g.nodes[node_id] + g.nodes[node_id] = replace(n, external=n.external + (link,)) + + +# ── Validation (SHACL-lite): vocabulary + provenance + invariants ───────────── +def validate(g: PKG, spec: dict | None = None) -> list[str]: + spec = spec or load_spec() + ntypes, rels = valid_node_types(spec), valid_relations(spec) + kgs = set(spec["external_kgs"].keys()) + errs: list[str] = [] + + selves = [n for n in g.nodes.values() if n.type == "Self"] + if len(selves) != 1: + errs.append(f"invariant self_is_the_single_anchor: found {len(selves)} Self nodes") + + for n in g.nodes.values(): + if n.type not in ntypes: + errs.append(f"node {n.id}: unknown type {n.type!r}") + if not n.provenance.source: + errs.append(f"node {n.id}: missing provenance.source") + for x in n.external: + if x.target_kg not in kgs: + errs.append(f"node {n.id}: external link to unknown KG {x.target_kg!r}") + + for e in g.edges: + if e.rel not in rels: + errs.append(f"edge {e.src}->{e.dst}: unknown relation {e.rel!r}") + if e.src not in g.nodes or e.dst not in g.nodes: + errs.append(f"edge {e.src}->{e.dst} ({e.rel}): dangling endpoint") + if not e.provenance.source: + errs.append(f"edge {e.src}->{e.dst}: missing provenance.source") + return errs diff --git a/tests/test_pkg.py b/tests/test_pkg.py new file mode 100644 index 0000000..505702e --- /dev/null +++ b/tests/test_pkg.py @@ -0,0 +1,72 @@ +"""Tests for the Personal Knowledge Graph seed + ingestion + validation.""" +from __future__ import annotations + +from prophet_mesh.pkg import ( + ExternalLink, + ingest_contact, + ingest_document, + ingest_event, + ingest_message, + link_external, + load_spec, + seed_graph, + validate, +) + +SPEC = load_spec() + + +def test_seed_is_single_self_anchor_plus_external_kgs(): + g = seed_graph() + assert len(g.nodes) == 1 + self_node = g.nodes["self"] + assert self_node.type == "Self" + assert self_node.assertion_class == "Structural" + assert self_node.provenance.source == "onboarding" + assert set(g.external_kgs) == set(SPEC["seed"]["registered_external_kgs"]) + assert validate(g, SPEC) == [] + + +def test_ingest_contact_family_vs_social_relation(): + g = seed_graph() + mom = ingest_contact(g, "Mom", relationship="mother") + jamie = ingest_contact(g, "Jamie", relationship="friend") + assert g.nodes[mom].type == "Person" + rels = {(e.src, e.dst): e.rel for e in g.edges} + assert rels[("self", mom)] == "relatedTo" # family + assert rels[("self", jamie)] == "knows" # social + # every ingested element carries provenance + assert all(e.provenance.source.startswith("workspace:") for e in g.edges) + assert validate(g, SPEC) == [] + + +def test_full_workspace_ingest_validates(): + g = seed_graph() + jamie = ingest_contact(g, "Jamie", relationship="friend") + ingest_message(g, jamie) + ingest_event(g, "Band practice") + ingest_document(g, "Setlist") + errs = validate(g, SPEC) + assert errs == [], errs + # spot-check the typed shape + types = sorted({n.type for n in g.nodes.values()}) + assert types == ["Document", "Event", "Person", "Self"] + + +def test_external_link_is_reference_only_and_kg_checked(): + g = seed_graph() + p = ingest_contact(g, "Jamie", relationship="friend") + link_external(g, p, ExternalLink(target_kg="social_network", target_id="sn:jamie", confidence=0.9, trust_class="platform")) + assert validate(g, SPEC) == [] + # an unknown external KG is rejected by the validator + link_external(g, p, ExternalLink(target_kg="darkweb", target_id="x")) + assert any("unknown KG" in e for e in validate(g, SPEC)) + + +def test_validator_catches_missing_provenance_and_bad_vocab(): + from dataclasses import replace + from prophet_mesh.pkg import Provenance + g = seed_graph() + # break provenance on a node + g.nodes["self"] = replace(g.nodes["self"], provenance=Provenance(source="", method="declared")) + assert any("missing provenance.source" in e for e in validate(g, SPEC)) From 84d9ef588d953b54d48ba19642a6818e4c75933f Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:59:25 -0400 Subject: [PATCH 07/12] feat(pkg): hellgraph writer + prophet-workspace adapters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Persist and populate the Personal Knowledge Graph on real substrate: pkg_hellgraph.py — projects a PKG into HellGraph's canonical wire shape (GraphNode {id,labels,properties,createdAt} + GraphEdge with the mandatory epistemic fields epistemicClass/confidence/promotionState/createdAt). Maps PKG inference/method → EpistemicClass (workspace imports = confirmed_relation/ confirmed, labelled inferences = inferred_relation/candidate); external links stay reference_only in properties. Emits to the TS HellGraphStore façade; the Rust kernel is untouched (external writers go through the TS projection). pkg_workspace.py — adapters over the real workspace contracts: contact.schema (→ Person + worksAt Organization + social sameAs), calendar-event.schema (→ Event + participatedIn for Self and attendees-with-contactRef), mail-message.schema (→ communicatedWith from the correspondent), office-artifact .schema (→ Document + authored). sourceAdapter/accountRef/timestamps flow into provenance; ingest_workspace() batches a whole export in dependency order so mail/calendar edges resolve onto Person nodes from contacts (no dup). 15 PKG tests pass (5 spec/builder + 5 writer + 5 adapters). --- src/prophet_mesh/pkg_hellgraph.py | 121 ++++++++++++++++++++++++ src/prophet_mesh/pkg_workspace.py | 148 ++++++++++++++++++++++++++++++ tests/test_pkg_hellgraph.py | 62 +++++++++++++ tests/test_pkg_workspace.py | 104 +++++++++++++++++++++ 4 files changed, 435 insertions(+) create mode 100644 src/prophet_mesh/pkg_hellgraph.py create mode 100644 src/prophet_mesh/pkg_workspace.py create mode 100644 tests/test_pkg_hellgraph.py create mode 100644 tests/test_pkg_workspace.py diff --git a/src/prophet_mesh/pkg_hellgraph.py b/src/prophet_mesh/pkg_hellgraph.py new file mode 100644 index 0000000..f634e63 --- /dev/null +++ b/src/prophet_mesh/pkg_hellgraph.py @@ -0,0 +1,121 @@ +"""PKG → HellGraph writer. + +Projects a Personal Knowledge Graph (pkg.PKG) into HellGraph's canonical wire +shape so it persists on the real substrate. HellGraph's write path is the TS +``HellGraphStore`` façade (``@socioprophet/hellgraph``): + + g.addNode(id, labels[], properties) + g.addEdge(label, from, to, properties) + +with GraphNode = {id, labels[], properties, createdAt} and every edge carrying +the mandatory epistemic fields {epistemicClass, confidence, promotionState, +createdAt}. This module emits exactly those payloads (JSON-serializable dicts) +from the Python side; a thin TS ingester replays them through the façade. We do +NOT touch the Rust kernel — external writers go through the TS projection. + +Idempotent by construction: HellGraph structural atoms are content-addressed, so +re-emitting the same PKG collapses to the same atoms. +""" +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from prophet_mesh.pkg import PKG, Edge, ExternalLink, Node + +# ── Epistemic mapping: PKG inference/method → HellGraph EpistemicClass ───────── +# HellGraph EpistemicClass (ts/src/types.ts): extracted_relation | inferred_relation +# | confirmed_relation | graph_extraction | semantic. +# A PKG element sourced from onboarding or workspace ingestion is a confirmed +# import (source of truth); a labelled inference is inferred_relation. +_INFERRED = {"I-DED", "I-IND", "I-ABD"} + + +def _epistemic_class(inference_type: str, method: str) -> str: + if inference_type in _INFERRED: + return "inferred_relation" + if method in {"declared", "imported", "ingested"}: + return "confirmed_relation" + return "extracted_relation" + + +def _promotion_state(inference_type: str) -> str: + # Confirmed imports are accepted; inferences enter as candidates for review. + return "candidate" if inference_type in _INFERRED else "confirmed" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _created_at(captured_at: str) -> str: + return captured_at or _now() + + +def _external_props(external: tuple[ExternalLink, ...]) -> dict[str, Any]: + """External links are reference-only; carry them as flat, auditable props.""" + props: dict[str, Any] = {} + for i, x in enumerate(external): + props[f"sameAs.{i}.kg"] = x.target_kg + props[f"sameAs.{i}.id"] = x.target_id + props[f"sameAs.{i}.confidence"] = x.confidence + props[f"sameAs.{i}.trust"] = x.trust_class + props[f"sameAs.{i}.direction"] = "reference_only" + return props + + +def node_payload(n: Node) -> dict[str, Any]: + """One GraphNode. Labels = [type, assertion_class]; provenance in properties.""" + props: dict[str, Any] = { + "label": n.label, + "provenance.source": n.provenance.source, + "provenance.method": n.provenance.method, + "provenance.captured_at": n.provenance.captured_at, + "assertion_class": n.assertion_class, + "provenance_tag": n.provenance_tag, # P-RET | P-GEN + "inference_type": n.inference_type, # I-NON | I-DED | I-IND | I-ABD + "memory_scope": n.memory_scope, + "confidence": n.confidence, + } + props.update(_external_props(n.external)) + return { + "id": n.id, + "labels": [n.type, n.assertion_class], + "properties": props, + "createdAt": _created_at(n.provenance.captured_at), + } + + +def edge_payload(e: Edge) -> dict[str, Any]: + """One GraphEdge. rel → label; PKG provenance → mandatory epistemic fields.""" + return { + "label": e.rel, + "from": e.src, + "to": e.dst, + "properties": { + # HellGraph-mandatory epistemic fields: + "epistemicClass": _epistemic_class(e.inference_type, e.provenance.method), + "confidence": e.confidence, + "promotionState": _promotion_state(e.inference_type), + "createdAt": _created_at(e.provenance.captured_at), + # PKG provenance / scope carried through for audit + replay: + "provenance.source": e.provenance.source, + "provenance.method": e.provenance.method, + "assertion_class": e.assertion_class, + "provenance_tag": e.provenance_tag, + "inference_type": e.inference_type, + "memory_scope": e.memory_scope, + }, + } + + +def to_hellgraph(g: PKG) -> dict[str, list[dict[str, Any]]]: + """The full PKG as a HellGraph ingest bundle: {nodes:[...], edges:[...]}. + + Feed to the TS ingester, which replays each through HellGraphStore.addNode / + .addEdge. Deterministic order (nodes before edges) so edge endpoints exist. + """ + return { + "nodes": [node_payload(n) for n in g.nodes.values()], + "edges": [edge_payload(e) for e in g.edges], + } diff --git a/src/prophet_mesh/pkg_workspace.py b/src/prophet_mesh/pkg_workspace.py new file mode 100644 index 0000000..f454c04 --- /dev/null +++ b/src/prophet_mesh/pkg_workspace.py @@ -0,0 +1,148 @@ +"""prophet-workspace → PKG adapters. + +Maps real prophet-workspace records into the Personal Knowledge Graph, one +adapter per contract: + + contact.schema.json → Person (+ worksAt Organization, social sameAs) + calendar-event.schema.json → Event + participatedIn (Self + attendees) + mail-message.schema.json → communicatedWith (Self ↔ the correspondent) + office-artifact.schema.json→ Document + authored + +Every emitted node/edge carries workspace provenance: source = +``workspace::``, method ``ingested``, captured_at from the +record's own timestamp (updatedAt / receivedAt / createdAt). External identities +(socialProfiles) become reference-only links to the social_network KG — never a +data copy. These consume the canonical records; they do not invent shape. +""" +from __future__ import annotations + +from typing import Any + +from prophet_mesh.pkg import PKG, Edge, ExternalLink, Node, Provenance, link_external + +# Contact.labels / groupRefs hints that reclassify knows → relatedTo (family). +FAMILY_HINTS = { + "family", "mom", "dad", "mother", "father", "sister", "brother", + "parent", "child", "sibling", "spouse", "wife", "husband", "son", "daughter", +} + + +def _prov(rec: dict[str, Any], surface: str, when_key: str = "updatedAt") -> Provenance: + adapter = rec.get("sourceAdapter") or rec.get("accountRef") or rec.get("workroomId") or "unknown" + captured = rec.get(when_key) or rec.get("createdAt") or "" + return Provenance(source=f"workspace:{surface}:{adapter}", method="ingested", captured_at=captured) + + +def _person_id(ref: str) -> str: + return f"person:{ref}" + + +# ── contact.schema.json → Person ────────────────────────────────────────────── +def adapt_contact(g: PKG, rec: dict[str, Any]) -> str: + if rec.get("contactClass") == "organization": + return adapt_organization(g, rec) + cid = rec["contactId"] + nid = _person_id(cid) + label = rec.get("displayName") or " ".join( + p for p in (rec.get("givenName"), rec.get("familyName")) if p + ) or cid + prov = _prov(rec, "contacts") + g.add_node(Node(id=nid, type="Person", label=label, provenance=prov)) + + hints = {h.lower() for h in (rec.get("labels") or [])} | {h.lower() for h in (rec.get("groupRefs") or [])} + rel = "relatedTo" if hints & FAMILY_HINTS else "knows" + g.add_edge(Edge(src=g.self_id, dst=nid, rel=rel, provenance=prov)) + + if rec.get("organizationRef"): + oid = f"org:{rec['organizationRef']}" + if oid not in g.nodes: + g.add_node(Node(id=oid, type="Organization", label=rec["organizationRef"], provenance=prov)) + g.add_edge(Edge(src=nid, dst=oid, rel="worksAt", provenance=prov)) + + for sp in rec.get("socialProfiles") or []: + target = sp.get("url") or f"{sp.get('platform')}:{sp.get('handle')}" + link_external(g, nid, ExternalLink( + target_kg="social_network", target_id=target, + confidence=0.6, trust_class="platform", + )) + return nid + + +def adapt_organization(g: PKG, rec: dict[str, Any]) -> str: + oid = f"org:{rec['contactId']}" + prov = _prov(rec, "contacts") + g.add_node(Node(id=oid, type="Organization", label=rec.get("displayName") or rec["contactId"], provenance=prov)) + return oid + + +# ── calendar-event.schema.json → Event + participatedIn ─────────────────────── +def adapt_event(g: PKG, rec: dict[str, Any]) -> str: + eid = rec["eventId"] + nid = f"event:{eid}" + prov = _prov(rec, "calendar") + g.add_node(Node(id=nid, type="Event", label=rec.get("title") or eid, provenance=prov)) + g.add_edge(Edge(src=g.self_id, dst=nid, rel="participatedIn", provenance=prov)) + + for att in rec.get("attendees") or []: + if att.get("isSelf"): + continue + cref = att.get("contactRef") + if not cref: + continue + pid = _person_id(cref) + if pid not in g.nodes: + g.add_node(Node(id=pid, type="Person", label=att.get("name") or cref, provenance=prov)) + g.add_edge(Edge(src=pid, dst=nid, rel="participatedIn", provenance=prov)) + return nid + + +# ── mail-message.schema.json → communicatedWith ─────────────────────────────── +def adapt_message(g: PKG, rec: dict[str, Any]) -> str | None: + """A message → a communicatedWith edge Self↔correspondent (the `from` party).""" + prov = _prov(rec, "mail", when_key="receivedAt") + frm = rec.get("from") or {} + if frm.get("isSelf"): + return None # outbound; the correspondent is on `to` — kept simple for now + cref = frm.get("contactRef") + if cref: + pid = _person_id(cref) + label = frm.get("name") or cref + else: + email = frm.get("email", "unknown") + pid = _person_id(f"email:{email}") + label = frm.get("name") or email + if pid not in g.nodes: + g.add_node(Node(id=pid, type="Person", label=label, provenance=prov)) + g.add_edge(Edge(src=g.self_id, dst=pid, rel="communicatedWith", provenance=prov)) + return pid + + +# ── office-artifact.schema.json → Document + authored ───────────────────────── +def adapt_artifact(g: PKG, rec: dict[str, Any]) -> str: + aid = rec["artifactId"] + nid = f"doc:{aid}" + prov = _prov(rec, "office", when_key="createdAt") + g.add_node(Node(id=nid, type="Document", label=rec.get("title") or aid, provenance=prov)) + g.add_edge(Edge(src=g.self_id, dst=nid, rel="authored", provenance=prov)) + return nid + + +# ── batch ingest ────────────────────────────────────────────────────────────── +def ingest_workspace( + g: PKG, + contacts: list[dict] | None = None, + events: list[dict] | None = None, + messages: list[dict] | None = None, + artifacts: list[dict] | None = None, +) -> PKG: + """Ingest whole workspace exports in dependency order (people first so mail/ + calendar edges resolve to existing Person nodes).""" + for rec in contacts or []: + adapt_contact(g, rec) + for rec in artifacts or []: + adapt_artifact(g, rec) + for rec in events or []: + adapt_event(g, rec) + for rec in messages or []: + adapt_message(g, rec) + return g diff --git a/tests/test_pkg_hellgraph.py b/tests/test_pkg_hellgraph.py new file mode 100644 index 0000000..b0e5134 --- /dev/null +++ b/tests/test_pkg_hellgraph.py @@ -0,0 +1,62 @@ +"""Tests for the PKG → HellGraph writer (canonical wire-shape conformance).""" +from __future__ import annotations + +from prophet_mesh.pkg import ExternalLink, ingest_contact, link_external, seed_graph +from prophet_mesh.pkg_hellgraph import edge_payload, node_payload, to_hellgraph + +REQUIRED_EDGE_EPISTEMIC = {"epistemicClass", "confidence", "promotionState", "createdAt"} +EPISTEMIC_CLASSES = { + "extracted_relation", "inferred_relation", "confirmed_relation", + "graph_extraction", "semantic", +} +PROMOTION_STATES = {"candidate", "confirmed", "contested", "superseded", "vetoed"} + + +def test_node_payload_has_graphnode_shape(): + g = seed_graph() + p = node_payload(g.nodes["self"]) + assert set(p.keys()) == {"id", "labels", "properties", "createdAt"} + assert p["id"] == "self" + assert p["labels"][0] == "Self" and "Structural" in p["labels"] + assert p["properties"]["provenance.source"] == "onboarding" + assert p["createdAt"] # never empty + + +def test_edge_payload_carries_mandatory_epistemic_fields(): + g = seed_graph() + ingest_contact(g, "Mom", relationship="mother") + e = g.edges[0] + p = edge_payload(e) + assert set(p.keys()) == {"label", "from", "to", "properties"} + assert REQUIRED_EDGE_EPISTEMIC <= set(p["properties"]) + assert p["properties"]["epistemicClass"] in EPISTEMIC_CLASSES + assert p["properties"]["promotionState"] in PROMOTION_STATES + + +def test_ingested_workspace_edge_is_confirmed_import(): + g = seed_graph() + ingest_contact(g, "Jamie", relationship="friend") + props = edge_payload(g.edges[0])["properties"] + # method="ingested", I-NON → a confirmed import, not a speculative candidate + assert props["epistemicClass"] == "confirmed_relation" + assert props["promotionState"] == "confirmed" + + +def test_external_link_is_reference_only_in_props(): + g = seed_graph() + p = ingest_contact(g, "Jamie", relationship="friend") + link_external(g, p, ExternalLink("social_network", "sn:jamie", 0.9, "platform")) + props = node_payload(g.nodes[p])["properties"] + assert props["sameAs.0.kg"] == "social_network" + assert props["sameAs.0.direction"] == "reference_only" + + +def test_full_bundle_nodes_before_edges_and_endpoints_exist(): + g = seed_graph() + jamie = ingest_contact(g, "Jamie", relationship="friend") + bundle = to_hellgraph(g) + node_ids = {n["id"] for n in bundle["nodes"]} + assert "self" in node_ids and jamie in node_ids + # every edge endpoint is present as a node in the same bundle + for e in bundle["edges"]: + assert e["from"] in node_ids and e["to"] in node_ids diff --git a/tests/test_pkg_workspace.py b/tests/test_pkg_workspace.py new file mode 100644 index 0000000..e11d6e2 --- /dev/null +++ b/tests/test_pkg_workspace.py @@ -0,0 +1,104 @@ +"""Tests for the prophet-workspace → PKG adapters (real record shapes).""" +from __future__ import annotations + +from prophet_mesh.pkg import load_spec, seed_graph, validate +from prophet_mesh.pkg_workspace import ( + adapt_contact, + adapt_event, + adapt_message, + ingest_workspace, +) + +SPEC = load_spec() + +CONTACT_MOM = { + "schemaVersion": "v0.1", "contactId": "c-mom", "accountRef": "acct-1", + "contactClass": "person", "displayName": "Mom", "givenName": "Ada", + "labels": ["family"], "sourceAdapter": "google_contacts", + "externalId": "people/c123", "updatedAt": "2026-06-28T14:22:00Z", + "socialProfiles": [{"platform": "linkedin", "handle": "ada", "url": "https://linkedin.com/in/ada"}], +} +CONTACT_JAMIE = { + "schemaVersion": "v0.1", "contactId": "c-jamie", "accountRef": "acct-1", + "contactClass": "person", "displayName": "Jamie", "organizationRef": "Acme Music", + "labels": ["bandmate"], "sourceAdapter": "carddav", "updatedAt": "2026-06-01T00:00:00Z", +} +EVENT_PRACTICE = { + "schemaVersion": "v0.1", "eventId": "e-practice", "calendarRef": "cal-1", + "title": "Band practice", "status": "confirmed", "organizerRef": "me@x.com", + "start": {"dateTime": "2026-07-15T18:00:00"}, "end": {"dateTime": "2026-07-15T20:00:00"}, + "sourceAdapter": "google_calendar", "updatedAt": "2026-07-10T00:00:00Z", + "attendees": [ + {"email": "me@x.com", "isSelf": True, "isOrganizer": True}, + {"email": "jamie@x.com", "name": "Jamie", "contactRef": "c-jamie", "responseStatus": "accepted"}, + ], +} +MAIL_FROM_JAMIE = { + "schemaVersion": "v0.1", "messageId": "m-1", "accountRef": "acct-mail-1", "threadId": "t-1", + "from": {"email": "jamie@x.com", "name": "Jamie", "contactRef": "c-jamie"}, + "to": [{"email": "me@x.com", "isSelf": True}], "subject": "setlist", + "receivedAt": "2026-06-28T09:45:00Z", "routingState": "imbox", "inboxSlot": "primary", +} +ARTIFACT_SETLIST = { + "schemaVersion": "v0.1", "artifactId": "a-setlist", "workroomId": "wr-band", + "artifactType": "document", "title": "Setlist", "format": "docx", "status": "approved", + "createdAt": "2026-06-20T00:00:00Z", +} + + +def test_contact_family_hint_becomes_relatedTo(): + g = seed_graph() + nid = adapt_contact(g, CONTACT_MOM) + rel = {(e.src, e.dst): e.rel for e in g.edges}[("self", nid)] + assert rel == "relatedTo" + # sourceAdapter flows into provenance; social profile is a reference-only link + n = g.nodes[nid] + assert n.provenance.source == "workspace:contacts:google_contacts" + assert n.provenance.captured_at == "2026-06-28T14:22:00Z" + assert n.external[0].target_kg == "social_network" + assert validate(g, SPEC) == [] + + +def test_contact_with_org_makes_worksAt(): + g = seed_graph() + nid = adapt_contact(g, CONTACT_JAMIE) + rels = {(e.src, e.dst): e.rel for e in g.edges} + assert rels[("self", nid)] == "knows" + assert rels[(nid, "org:Acme Music")] == "worksAt" + assert g.nodes["org:Acme Music"].type == "Organization" + assert validate(g, SPEC) == [] + + +def test_event_attendee_with_contactRef_participates(): + g = seed_graph() + adapt_event(g, EVENT_PRACTICE) + rels = {(e.src, e.dst): e.rel for e in g.edges} + assert rels[("self", "event:e-practice")] == "participatedIn" + assert rels[("person:c-jamie", "event:e-practice")] == "participatedIn" # attendee, not self + assert validate(g, SPEC) == [] + + +def test_mail_from_contact_is_communicatedWith(): + g = seed_graph() + pid = adapt_message(g, MAIL_FROM_JAMIE) + assert pid == "person:c-jamie" + e = g.edges[-1] + assert (e.src, e.dst, e.rel) == ("self", "person:c-jamie", "communicatedWith") + assert e.provenance.captured_at == "2026-06-28T09:45:00Z" # receivedAt + assert validate(g, SPEC) == [] + + +def test_full_workspace_export_ingests_and_validates(): + g = seed_graph() + ingest_workspace( + g, + contacts=[CONTACT_MOM, CONTACT_JAMIE], + events=[EVENT_PRACTICE], + messages=[MAIL_FROM_JAMIE], + artifacts=[ARTIFACT_SETLIST], + ) + assert validate(g, SPEC) == [] + types = sorted({n.type for n in g.nodes.values()}) + assert types == ["Document", "Event", "Organization", "Person", "Self"] + # mail resolved onto the SAME Jamie node created from contacts (no dup) + assert sum(1 for n in g.nodes.values() if n.label == "Jamie") == 1 From 8da919b5e218440fd0423637c04637b3b8f19034 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Thu, 2 Jul 2026 04:16:11 -0400 Subject: [PATCH 08/12] =?UTF-8?q?feat(pkg):=20CRDT=20Slice=201=20=E2=80=94?= =?UTF-8?q?=20EventEnvelope=20op-emitter=20over=20PKG=20mutations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Non-invasive EmittingPKG wrapper dual-writes every add_node/add_edge as a sourceos-spec v2 EventEnvelope onto an append-only hash-linked log (Fig-18 railroad track). Causal + locus/attestation metadata rides in the open payload; no schema change, pkg.py untouched. Includes deterministic fold() round-trip and verify_chain() tamper check. Emit-only — merge is Slice 3, gate is Slice 4. 5 new tests; full suite 65 green. --- src/prophet_mesh/pkg_ops.py | 195 ++++++++++++++++++++++++++++++++++++ tests/test_pkg_ops.py | 74 ++++++++++++++ 2 files changed, 269 insertions(+) create mode 100644 src/prophet_mesh/pkg_ops.py create mode 100644 tests/test_pkg_ops.py diff --git a/src/prophet_mesh/pkg_ops.py b/src/prophet_mesh/pkg_ops.py new file mode 100644 index 0000000..37c4e32 --- /dev/null +++ b/src/prophet_mesh/pkg_ops.py @@ -0,0 +1,195 @@ +"""CRDT Slice 1 — op-emitter over the Personal Knowledge Graph. + +Dual-writes every PKG mutation as a sourceos-spec v2 ``EventEnvelope`` op onto an +append-only, hash-linked log (the Figure-18 "railroad track"). This is Slice 1 of +the CRDT-over-evidence-fabric design: **emit only, no merge yet**. It gives us: + + * an EventEnvelope-conformant op for every add_node / add_edge, + * a content-hash DAG (parents chain) so history is verifiable, and + * a deterministic ``fold`` that rebuilds graph state from the log alone + (the round-trip proof that the log is sufficient — the substrate Slice 3 + merge will union). + +Causal + trust metadata rides in the open ``payload`` (EventEnvelope keeps +``additionalProperties: false`` at the top level, so we add nothing there). +The merge gate that consumes ``crdt.locus`` / ``crdt.attestationRef`` is Slice 4; +here we only record them. + +Conforms to: EventEnvelope.json, and reuses the SyncCycleReceipt.locus vocabulary. +Accretive: does not modify pkg.py. +""" +from __future__ import annotations + +import hashlib +import json +import uuid +from dataclasses import asdict +from datetime import datetime, timezone +from typing import Any + +from .pkg import PKG, Node, Edge, Provenance + +SPEC_VERSION = "2.0.0" + +# SyncCycleReceipt.locus vocabulary — the "distance-to-execution" axis. +LOCI = ("local", "trusted_private", "attested_fog", "burst_cloud") + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _canonical(obj: Any) -> str: + """Stable JSON for content hashing (sorted keys, no whitespace drift).""" + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def _content_hash(event_type: str, object_id: str, payload: dict) -> str: + """Merkle-DAG node hash. Covers semantics + causal parents, NOT wall-clock + (occurredAt is display-only and never trusted for ordering).""" + material = _canonical({"eventType": event_type, "objectId": object_id, "payload": payload}) + return "sha256:" + hashlib.sha256(material.encode("utf-8")).hexdigest() + + +class OpLog: + """Append-only EventEnvelope log with a single-replica hash chain. + + Slice 1 is single-replica, so ``parents`` is the linear chain head. Slice 3 + generalises ``parents`` to the concurrent (multi-head) case for OR-Set merge. + """ + + def __init__(self, replica_id: str, subject_id: str, locus: str = "local") -> None: + if locus not in LOCI: + raise ValueError(f"locus must be one of {LOCI}, got {locus!r}") + self.replica_id = replica_id + self.subject_id = subject_id + self.locus = locus + self.ops: list[dict] = [] + self._head: str | None = None + self._lamport = 0 + + def append( + self, + event_type: str, + object_id: str, + core: dict, + *, + attestation_ref: str | None = None, + trust_level: str = "trusted-workspace-source", + ) -> dict: + self._lamport += 1 + event_id = f"urn:srcos:event:{uuid.uuid4()}" + crdt = { + "replicaId": self.replica_id, + "lamport": self._lamport, + "parents": [self._head] if self._head else [], + "locus": self.locus, + "attestationRef": attestation_ref, # resolved by the Slice-4 merge gate + "trustLevel": trust_level, + "tag": event_id, # OR-Set unique add-tag + } + payload = {"op": event_type, **core, "crdt": crdt} + event_hash = _content_hash(event_type, object_id, payload) + envelope = { + "eventId": event_id, + "eventType": event_type, + "specVersion": SPEC_VERSION, + "occurredAt": _now(), + "actor": {"subjectId": self.subject_id}, + "objectId": object_id, + "payload": payload, + "integrity": {"eventHash": event_hash, "signature": None}, + } + self._head = event_hash + self.ops.append(envelope) + return envelope + + +def _node_urn(node_id: str) -> str: + return f"urn:srcos:pkg-node:{node_id}" + + +def _edge_urn(e: Edge) -> str: + return f"urn:srcos:pkg-edge:{e.src}|{e.rel}|{e.dst}" + + +class EmittingPKG: + """Composes a PKG with an OpLog: every mutation is applied AND emitted. + + Non-invasive wrapper — the underlying PKG is untouched, so all existing + ingestion (ingest_contact, workspace adapters, hellgraph writer) can be + driven through this to start producing ops with zero changes to pkg.py. + """ + + def __init__(self, graph: PKG, *, replica_id: str, locus: str = "local") -> None: + self.g = graph + self.log = OpLog(replica_id=replica_id, subject_id=f"urn:srcos:replica:{replica_id}", locus=locus) + + @classmethod + def seeded(cls, self_id: str = "self", *, replica_id: str, locus: str = "local") -> "EmittingPKG": + """Fresh graph whose Self anchor is emitted as the genesis op, so the + op-log is a *complete* source of truth (a fresh replica can fold it and + recover the anchor). external_kgs registration as an op is a later slice. + """ + epkg = cls(PKG(self_id=self_id), replica_id=replica_id, locus=locus) + epkg.add_node(Node( + id=self_id, type="Self", label="Self", + provenance=Provenance(source="onboarding", method="declared"), + assertion_class="Structural", + )) + return epkg + + def add_node(self, n: Node, *, attestation_ref: str | None = None) -> dict: + self.g.add_node(n) + return self.log.append( + "GraphNodeAsserted", _node_urn(n.id), {"node": asdict(n)}, + attestation_ref=attestation_ref, + ) + + def add_edge(self, e: Edge, *, attestation_ref: str | None = None) -> dict: + self.g.add_edge(e) + return self.log.append( + "GraphEdgeAsserted", _edge_urn(e), {"edge": asdict(e)}, + attestation_ref=attestation_ref, + ) + + +def fold(ops: list[dict], self_id: str = "self") -> PKG: + """Deterministically rebuild graph state from the op-log alone. + + Slice-1 semantics: asserts only (add-wins, no retracts yet). Proves the log + is a sufficient source of truth — the property Slice 3's merge relies on. + """ + g = PKG(self_id=self_id) + for env in ops: + p = env["payload"] + op = p["op"] + if op == "GraphNodeAsserted": + nd = dict(p["node"]) + nd["provenance"] = _Prov(**nd["provenance"]) + nd["external"] = tuple(_Ext(**x) for x in nd.get("external", ())) + g.add_node(Node(**nd)) + elif op == "GraphEdgeAsserted": + ed = dict(p["edge"]) + ed["provenance"] = _Prov(**ed["provenance"]) + g.add_edge(Edge(**ed)) + return g + + +# local re-imports kept at bottom to avoid confusing the module header +from .pkg import Provenance as _Prov, ExternalLink as _Ext # noqa: E402 + + +def verify_chain(ops: list[dict]) -> bool: + """Every op's recorded eventHash matches its recomputed content hash, and + parents point at the immediately preceding head (single-replica chain).""" + prev: str | None = None + for env in ops: + p = env["payload"] + recomputed = _content_hash(env["eventType"], env["objectId"], p) + if recomputed != env["integrity"]["eventHash"]: + return False + if p["crdt"]["parents"] != ([prev] if prev else []): + return False + prev = env["integrity"]["eventHash"] + return True diff --git a/tests/test_pkg_ops.py b/tests/test_pkg_ops.py new file mode 100644 index 0000000..9c17278 --- /dev/null +++ b/tests/test_pkg_ops.py @@ -0,0 +1,74 @@ +"""CRDT Slice 1 tests — op-emitter conformance, hash chain, and fold round-trip.""" +from __future__ import annotations + +import re + +from prophet_mesh.pkg import PKG, Node, Edge, Provenance +from prophet_mesh.pkg_ops import EmittingPKG, fold, verify_chain, LOCI, SPEC_VERSION + +_EVENT_ID = re.compile(r"^urn:srcos:event:") +_REQUIRED = {"eventId", "eventType", "specVersion", "occurredAt", "actor", "objectId", "payload"} + + +def _build() -> EmittingPKG: + # seeded() emits the Self anchor as the genesis op → the log is complete. + epkg = EmittingPKG.seeded("self", replica_id="device-a", locus="trusted_private") + epkg.add_node(Node( + id="person:ada", type="Person", label="Ada", + provenance=Provenance(source="workspace:contacts", method="ingested"), + )) + epkg.add_edge(Edge( + src="self", dst="person:ada", rel="knows", + provenance=Provenance(source="workspace:contacts", method="ingested"), + )) + return epkg + + +def test_every_mutation_emits_a_conformant_envelope(): + epkg = _build() + assert len(epkg.log.ops) == 3 # genesis Self + Person node + edge + for env in epkg.log.ops: + # EventEnvelope required fields (additionalProperties:false — only allowed keys) + assert _REQUIRED <= set(env.keys()) + assert set(env.keys()) <= _REQUIRED | {"integrity"} + assert _EVENT_ID.match(env["eventId"]) + assert env["specVersion"] == SPEC_VERSION + assert env["actor"]["subjectId"].startswith("urn:srcos:replica:") + assert env["eventType"] in ("GraphNodeAsserted", "GraphEdgeAsserted") + + +def test_crdt_metadata_records_locus_and_causal_order(): + epkg = _build() + self_op, node_op, edge_op = epkg.log.ops + for env in (self_op, node_op, edge_op): + crdt = env["payload"]["crdt"] + assert crdt["locus"] in LOCI and crdt["locus"] == "trusted_private" + assert crdt["tag"] == env["eventId"] # OR-Set add-tag identity + assert crdt["trustLevel"] == "trusted-workspace-source" + # lamport increases; edge causally follows the node (parents = node's hash) + assert self_op["payload"]["crdt"]["lamport"] == 1 + assert node_op["payload"]["crdt"]["lamport"] == 2 + assert edge_op["payload"]["crdt"]["lamport"] == 3 + assert edge_op["payload"]["crdt"]["parents"] == [node_op["integrity"]["eventHash"]] + + +def test_hash_chain_verifies(): + epkg = _build() + assert verify_chain(epkg.log.ops) is True + + +def test_tamper_breaks_the_chain(): + epkg = _build() + epkg.log.ops[0]["payload"]["node"]["label"] = "Mallory" # mutate content, not hash + assert verify_chain(epkg.log.ops) is False + + +def test_fold_reconstructs_graph_state_from_log_alone(): + epkg = _build() + rebuilt: PKG = fold(epkg.log.ops) + # nodes + edges recovered from the op-log match the live graph + assert set(rebuilt.nodes) == set(epkg.g.nodes) + assert rebuilt.nodes["person:ada"].label == "Ada" + live_edges = {(e.src, e.rel, e.dst) for e in epkg.g.edges} + fold_edges = {(e.src, e.rel, e.dst) for e in rebuilt.edges} + assert live_edges == fold_edges From 687f5f42f9e480f693970c4592ce2f118977940e Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Thu, 2 Jul 2026 11:51:26 -0400 Subject: [PATCH 09/12] =?UTF-8?q?feat(pkg):=20CRDT=20Slice=202/3=20?= =?UTF-8?q?=E2=80=94=20reconstruct=20+=20converge=20(OR-Set=20merge)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit materialize(): deterministic add-wins OR-Set fold with observed-remove retracts + LWW(lamport,eventId) value resolution — order-independent reconstruction from a log. merge(): dedup-by-eventId union of replicas' logs; commutative/associative/idempotent → strong eventual consistency, zero coordination. Closes local-first ideals 2 (multi- device) + 4 (collaboration) without touching 5/6/7. Adds retract_node/edge (observed- remove) + apply_remote (ingest without re-emit) to EmittingPKG. 9 new tests: commutativity, associativity, idempotence, offline convergence, random- shuffle order-independence, add-wins-over-concurrent-retract, retract-wins. Suite 74 green. --- src/prophet_mesh/pkg_ops.py | 215 ++++++++++++++++++++++++++---------- tests/test_pkg_merge.py | 129 ++++++++++++++++++++++ 2 files changed, 287 insertions(+), 57 deletions(-) create mode 100644 tests/test_pkg_merge.py diff --git a/src/prophet_mesh/pkg_ops.py b/src/prophet_mesh/pkg_ops.py index 37c4e32..e77ee69 100644 --- a/src/prophet_mesh/pkg_ops.py +++ b/src/prophet_mesh/pkg_ops.py @@ -1,21 +1,23 @@ -"""CRDT Slice 1 — op-emitter over the Personal Knowledge Graph. +"""CRDT over the Personal Knowledge Graph — Slices 1–3. Dual-writes every PKG mutation as a sourceos-spec v2 ``EventEnvelope`` op onto an -append-only, hash-linked log (the Figure-18 "railroad track"). This is Slice 1 of -the CRDT-over-evidence-fabric design: **emit only, no merge yet**. It gives us: - - * an EventEnvelope-conformant op for every add_node / add_edge, - * a content-hash DAG (parents chain) so history is verifiable, and - * a deterministic ``fold`` that rebuilds graph state from the log alone - (the round-trip proof that the log is sufficient — the substrate Slice 3 - merge will union). - -Causal + trust metadata rides in the open ``payload`` (EventEnvelope keeps -``additionalProperties: false`` at the top level, so we add nothing there). -The merge gate that consumes ``crdt.locus`` / ``crdt.attestationRef`` is Slice 4; -here we only record them. - -Conforms to: EventEnvelope.json, and reuses the SyncCycleReceipt.locus vocabulary. +append-only, hash-linked log (the Figure-18 "railroad track"), then **reconstructs +and converges** graph state from those logs across replicas. + + * Slice 1 — emit: an EventEnvelope-conformant op per add/retract, content-hash DAG. + * Slice 2 — reconstruct: deterministic ``materialize`` folds a log back to a PKG, + with add-wins OR-Set semantics + observed-remove retracts. + * Slice 3 — converge: ``merge`` unions two replicas' logs; ``materialize`` over the + union is commutative / associative / idempotent → strong eventual consistency + with zero coordination. This is what closes local-first ideals 2 (multi-device) + and 4 (collaboration) without touching 5/6/7. + +Causal + trust metadata (parents, lamport, locus, attestationRef, OR-Set tags/removes) +rides in the open ``payload`` — EventEnvelope keeps ``additionalProperties: false`` at +the top level, so no schema change. The locus/attestation *merge gate* (which converged +sub-graph is authoritative) is Slice 4; here every op still converges. + +Conforms to: EventEnvelope.json; reuses the SyncCycleReceipt.locus vocabulary. Accretive: does not modify pkg.py. """ from __future__ import annotations @@ -25,15 +27,17 @@ import uuid from dataclasses import asdict from datetime import datetime, timezone -from typing import Any +from typing import Any, Iterable -from .pkg import PKG, Node, Edge, Provenance +from .pkg import PKG, Node, Edge, Provenance, ExternalLink SPEC_VERSION = "2.0.0" # SyncCycleReceipt.locus vocabulary — the "distance-to-execution" axis. LOCI = ("local", "trusted_private", "attested_fog", "burst_cloud") +EdgeKey = tuple[str, str, str] # (src, rel, dst) + def _now() -> str: return datetime.now(timezone.utc).isoformat() @@ -54,9 +58,8 @@ def _content_hash(event_type: str, object_id: str, payload: dict) -> str: class OpLog: """Append-only EventEnvelope log with a single-replica hash chain. - Slice 1 is single-replica, so ``parents`` is the linear chain head. Slice 3 - generalises ``parents`` to the concurrent (multi-head) case for OR-Set merge. - """ + Single-replica ``parents`` is the linear chain head; after a ``merge`` the DAG + is multi-head (``verify_chain`` is therefore a per-replica, pre-merge check).""" def __init__(self, replica_id: str, subject_id: str, locus: str = "local") -> None: if locus not in LOCI: @@ -76,6 +79,7 @@ def append( *, attestation_ref: str | None = None, trust_level: str = "trusted-workspace-source", + removes: Iterable[str] | None = None, ) -> dict: self._lamport += 1 event_id = f"urn:srcos:event:{uuid.uuid4()}" @@ -88,6 +92,8 @@ def append( "trustLevel": trust_level, "tag": event_id, # OR-Set unique add-tag } + if removes is not None: + crdt["removes"] = list(removes) # OR-Set observed-remove tags payload = {"op": event_type, **core, "crdt": crdt} event_hash = _content_hash(event_type, object_id, payload) envelope = { @@ -113,24 +119,38 @@ def _edge_urn(e: Edge) -> str: return f"urn:srcos:pkg-edge:{e.src}|{e.rel}|{e.dst}" +def _node_from(nd: dict) -> Node: + nd = dict(nd) + nd["provenance"] = Provenance(**nd["provenance"]) + nd["external"] = tuple(ExternalLink(**x) for x in nd.get("external", ())) + return Node(**nd) + + +def _edge_from(ed: dict) -> Edge: + ed = dict(ed) + ed["provenance"] = Provenance(**ed["provenance"]) + return Edge(**ed) + + class EmittingPKG: """Composes a PKG with an OpLog: every mutation is applied AND emitted. Non-invasive wrapper — the underlying PKG is untouched, so all existing - ingestion (ingest_contact, workspace adapters, hellgraph writer) can be - driven through this to start producing ops with zero changes to pkg.py. - """ + ingestion (ingest_contact, workspace adapters, hellgraph writer) can be driven + through this to start producing ops with zero changes to pkg.py. Tracks the + live add-tags per element so retracts are *observed-removes* (add-wins).""" def __init__(self, graph: PKG, *, replica_id: str, locus: str = "local") -> None: self.g = graph self.log = OpLog(replica_id=replica_id, subject_id=f"urn:srcos:replica:{replica_id}", locus=locus) + self._node_tags: dict[str, set[str]] = {} + self._edge_tags: dict[EdgeKey, set[str]] = {} @classmethod def seeded(cls, self_id: str = "self", *, replica_id: str, locus: str = "local") -> "EmittingPKG": - """Fresh graph whose Self anchor is emitted as the genesis op, so the - op-log is a *complete* source of truth (a fresh replica can fold it and - recover the anchor). external_kgs registration as an op is a later slice. - """ + """Fresh graph whose Self anchor is emitted as the genesis op, so the op-log + is a *complete* source of truth (a fresh replica can materialize it and + recover the anchor). external_kgs registration as an op is a later slice.""" epkg = cls(PKG(self_id=self_id), replica_id=replica_id, locus=locus) epkg.add_node(Node( id=self_id, type="Self", label="Self", @@ -139,55 +159,136 @@ def seeded(cls, self_id: str = "self", *, replica_id: str, locus: str = "local") )) return epkg + # ── local mutations (apply + emit) ────────────────────────────────────── def add_node(self, n: Node, *, attestation_ref: str | None = None) -> dict: self.g.add_node(n) - return self.log.append( - "GraphNodeAsserted", _node_urn(n.id), {"node": asdict(n)}, - attestation_ref=attestation_ref, - ) + env = self.log.append("GraphNodeAsserted", _node_urn(n.id), {"node": asdict(n)}, attestation_ref=attestation_ref) + self._node_tags.setdefault(n.id, set()).add(env["payload"]["crdt"]["tag"]) + return env def add_edge(self, e: Edge, *, attestation_ref: str | None = None) -> dict: self.g.add_edge(e) - return self.log.append( - "GraphEdgeAsserted", _edge_urn(e), {"edge": asdict(e)}, - attestation_ref=attestation_ref, - ) + env = self.log.append("GraphEdgeAsserted", _edge_urn(e), {"edge": asdict(e)}, attestation_ref=attestation_ref) + self._edge_tags.setdefault((e.src, e.rel, e.dst), set()).add(env["payload"]["crdt"]["tag"]) + return env + + def retract_node(self, node_id: str, *, attestation_ref: str | None = None) -> dict: + """Observed-remove: retract exactly the add-tags this replica has seen, so a + concurrent re-add elsewhere (a tag we didn't observe) survives — add-wins.""" + observed = sorted(self._node_tags.get(node_id, set())) + env = self.log.append("GraphNodeRetracted", _node_urn(node_id), {"nodeId": node_id}, + attestation_ref=attestation_ref, removes=observed) + self.g.nodes.pop(node_id, None) + self._node_tags.pop(node_id, None) + return env + + def retract_edge(self, key: EdgeKey, *, attestation_ref: str | None = None) -> dict: + observed = sorted(self._edge_tags.get(key, set())) + env = self.log.append("GraphEdgeRetracted", f"urn:srcos:pkg-edge:{key[0]}|{key[1]}|{key[2]}", + {"edgeKey": list(key)}, attestation_ref=attestation_ref, removes=observed) + self.g.edges = [e for e in self.g.edges if (e.src, e.rel, e.dst) != key] + self._edge_tags.pop(key, None) + return env + + # ── remote op ingestion (apply without re-emitting) ───────────────────── + def apply_remote(self, env: dict) -> None: + """Ingest another replica's op into local view + tag trackers, WITHOUT + appending it to our own log. Lets this replica observe a remote add so a + later local retract removes the right tag.""" + p, crdt = env["payload"], env["payload"]["crdt"] + op = p["op"] + if op == "GraphNodeAsserted": + self.g.add_node(_node_from(p["node"])) + self._node_tags.setdefault(p["node"]["id"], set()).add(crdt["tag"]) + elif op == "GraphEdgeAsserted": + ed = p["edge"]; key = (ed["src"], ed["rel"], ed["dst"]) + self.g.add_edge(_edge_from(ed)) + self._edge_tags.setdefault(key, set()).add(crdt["tag"]) + elif op == "GraphNodeRetracted": + nid, rem = p["nodeId"], set(crdt.get("removes", [])) + self._node_tags[nid] = self._node_tags.get(nid, set()) - rem + if not self._node_tags.get(nid): + self.g.nodes.pop(nid, None) + self._node_tags.pop(nid, None) + elif op == "GraphEdgeRetracted": + key = tuple(p["edgeKey"]); rem = set(crdt.get("removes", [])) + self._edge_tags[key] = self._edge_tags.get(key, set()) - rem + if not self._edge_tags.get(key): + self.g.edges = [e for e in self.g.edges if (e.src, e.rel, e.dst) != key] + self._edge_tags.pop(key, None) + + +# ── Slice 2/3: reconstruct + converge ────────────────────────────────────── +def _dedupe(ops: Iterable[dict]) -> list[dict]: + """Idempotent op-set keyed by eventId (re-applying the same op is a no-op).""" + seen: dict[str, dict] = {} + for env in ops: + seen[env["eventId"]] = env + return list(seen.values()) -def fold(ops: list[dict], self_id: str = "self") -> PKG: - """Deterministically rebuild graph state from the op-log alone. +def materialize(ops: Iterable[dict], self_id: str = "self") -> PKG: + """Deterministically reconstruct graph state from an op-log (Slice 2). - Slice-1 semantics: asserts only (add-wins, no retracts yet). Proves the log - is a sufficient source of truth — the property Slice 3's merge relies on. - """ - g = PKG(self_id=self_id) - for env in ops: - p = env["payload"] + Add-wins OR-Set: an element is present iff it has ≥1 add-tag not covered by an + observed-remove. Among surviving adds, value is last-writer-wins by + (lamport, eventId) — a total, replica-independent tiebreak. Pure fold over the + op *set*, so it is order-independent → the property Slice-3 merge relies on.""" + events = _dedupe(ops) + node_adds: dict[str, list[tuple[int, str, dict, str]]] = {} + edge_adds: dict[EdgeKey, list[tuple[int, str, dict, str]]] = {} + removed: set[str] = set() + + for env in events: + p, crdt = env["payload"], env["payload"]["crdt"] op = p["op"] if op == "GraphNodeAsserted": - nd = dict(p["node"]) - nd["provenance"] = _Prov(**nd["provenance"]) - nd["external"] = tuple(_Ext(**x) for x in nd.get("external", ())) - g.add_node(Node(**nd)) + node_adds.setdefault(p["node"]["id"], []).append( + (crdt["lamport"], env["eventId"], p["node"], crdt["tag"])) elif op == "GraphEdgeAsserted": - ed = dict(p["edge"]) - ed["provenance"] = _Prov(**ed["provenance"]) - g.add_edge(Edge(**ed)) + e = p["edge"]; key = (e["src"], e["rel"], e["dst"]) + edge_adds.setdefault(key, []).append( + (crdt["lamport"], env["eventId"], e, crdt["tag"])) + elif op in ("GraphNodeRetracted", "GraphEdgeRetracted"): + removed.update(crdt.get("removes", [])) + + g = PKG(self_id=self_id) + for adds in node_adds.values(): + survivors = [a for a in adds if a[3] not in removed] + if survivors: + g.add_node(_node_from(max(survivors, key=lambda a: (a[0], a[1]))[2])) + for adds in edge_adds.values(): + survivors = [a for a in adds if a[3] not in removed] + if survivors: + g.add_edge(_edge_from(max(survivors, key=lambda a: (a[0], a[1]))[2])) return g -# local re-imports kept at bottom to avoid confusing the module header -from .pkg import Provenance as _Prov, ExternalLink as _Ext # noqa: E402 +def fold(ops: list[dict], self_id: str = "self") -> PKG: + """Slice-1 alias — reconstruct from the log (now backed by ``materialize``).""" + return materialize(ops, self_id=self_id) + + +def merge(*logs: "OpLog | list[dict]") -> list[dict]: + """Union replicas' op-logs into one deduped, deterministically-ordered log (Slice 3). + + Commutative / associative / idempotent by construction (set union keyed by + eventId). Ordering by (lamport, eventId) is only for a stable *serialized* form; + ``materialize`` is order-independent, so convergence does not depend on it.""" + merged: list[dict] = [] + for log in logs: + merged.extend(log.ops if isinstance(log, OpLog) else log) + return sorted(_dedupe(merged), key=lambda e: (e["payload"]["crdt"]["lamport"], e["eventId"])) def verify_chain(ops: list[dict]) -> bool: - """Every op's recorded eventHash matches its recomputed content hash, and - parents point at the immediately preceding head (single-replica chain).""" + """Every op's recorded eventHash matches its recomputed content hash, and parents + point at the immediately preceding head. Single-replica / pre-merge check — after + a merge the DAG is multi-head and this no longer applies.""" prev: str | None = None for env in ops: p = env["payload"] - recomputed = _content_hash(env["eventType"], env["objectId"], p) - if recomputed != env["integrity"]["eventHash"]: + if _content_hash(env["eventType"], env["objectId"], p) != env["integrity"]["eventHash"]: return False if p["crdt"]["parents"] != ([prev] if prev else []): return False diff --git a/tests/test_pkg_merge.py b/tests/test_pkg_merge.py new file mode 100644 index 0000000..f55983c --- /dev/null +++ b/tests/test_pkg_merge.py @@ -0,0 +1,129 @@ +"""CRDT Slice 2/3 — reconstruct + converge. + +Proves the OR-Set merge is commutative / associative / idempotent, that offline +replicas converge, and that concurrent add-vs-retract resolves add-wins. +""" +from __future__ import annotations + +import random + +from prophet_mesh.pkg import PKG, Node, Edge, Provenance +from prophet_mesh.pkg_ops import EmittingPKG, materialize, merge, fold + + +def _prov(src="workspace:contacts"): + return Provenance(source=src, method="ingested") + + +def _node(nid, label): + return Node(id=nid, type="Person", label=label, provenance=_prov()) + + +def _edge(src, rel, dst): + return Edge(src=src, dst=dst, rel=rel, provenance=_prov()) + + +def _state(g: PKG): + """Comparable convergence fingerprint: node ids+labels and edge triples.""" + nodes = frozenset((n.id, n.label) for n in g.nodes.values()) + edges = frozenset((e.src, e.rel, e.dst) for e in g.edges) + return nodes, edges + + +def _replica_a(): + a = EmittingPKG.seeded("self", replica_id="device-a", locus="trusted_private") + a.add_node(_node("person:ada", "Ada")) + a.add_edge(_edge("self", "knows", "person:ada")) + return a + + +def _replica_b(): + b = EmittingPKG.seeded("self", replica_id="device-b", locus="trusted_private") + b.add_node(_node("person:bob", "Bob")) + b.add_edge(_edge("self", "knows", "person:bob")) + return b + + +# ── Slice 2: reconstruct ──────────────────────────────────────────────────── +def test_materialize_matches_live_graph(): + a = _replica_a() + assert _state(materialize(a.log.ops)) == _state(a.g) + + +def test_retract_removes_element_but_keeps_the_rest(): + a = _replica_a() + a.retract_node("person:ada") + g = materialize(a.log.ops) + ids = {n.id for n in g.nodes.values()} + assert "person:ada" not in ids and "self" in ids + + +# ── Slice 3: converge ─────────────────────────────────────────────────────── +def test_merge_is_commutative(): + a, b = _replica_a(), _replica_b() + assert _state(materialize(merge(a.log, b.log))) == _state(materialize(merge(b.log, a.log))) + + +def test_merge_is_idempotent(): + a = _replica_a() + once = merge(a.log) + twice = merge(a.log, a.log) + assert {e["eventId"] for e in once} == {e["eventId"] for e in twice} + assert _state(materialize(twice)) == _state(materialize(a.log.ops)) + + +def test_merge_is_associative(): + a, b = _replica_a(), _replica_b() + c = EmittingPKG.seeded("self", replica_id="device-c", locus="local") + c.add_node(_node("person:cyd", "Cyd")) + left = merge(merge(a.log, b.log), c.log) + right = merge(a.log, merge(b.log, c.log)) + assert _state(materialize(left)) == _state(materialize(right)) + + +def test_two_offline_replicas_converge(): + a, b = _replica_a(), _replica_b() # each edited offline, never saw the other + converged = materialize(merge(a.log, b.log)) + nodes, edges = _state(converged) + ids = {nid for nid, _ in nodes} + assert ids == {"self", "person:ada", "person:bob"} + assert ("self", "knows", "person:ada") in edges + assert ("self", "knows", "person:bob") in edges + + +def test_order_independent_under_random_shuffle(): + a, b = _replica_a(), _replica_b() + union = merge(a.log, b.log) + baseline = _state(materialize(union)) + rng = random.Random(1729) + for _ in range(20): + shuffled = union[:] + rng.shuffle(shuffled) + assert _state(materialize(shuffled)) == baseline + + +def test_add_wins_over_concurrent_retract(): + # A adds Ada; B observes that add, then retracts it. Concurrently A re-adds Ada + # (a NEW tag B never saw) with an updated label. Merge → Ada present, updated. + a = EmittingPKG.seeded("self", replica_id="device-a", locus="local") + add1 = a.add_node(_node("person:ada", "Ada")) + + b = EmittingPKG.seeded("self", replica_id="device-b", locus="local") + b.apply_remote(add1) # B observed Ada@tag1 + b.retract_node("person:ada") # removes only tag1 + + a.add_node(_node("person:ada", "Ada (updated)")) # concurrent re-add, tag2 + + converged = materialize(merge(a.log, b.log)) + ada = [n for n in converged.nodes.values() if n.id == "person:ada"] + assert len(ada) == 1 # add-wins: survives the concurrent retract + assert ada[0].label == "Ada (updated)" + + +def test_retract_wins_when_all_observed_tags_removed(): + # If the retract observed every add-tag, the element is genuinely gone. + a = EmittingPKG.seeded("self", replica_id="device-a", locus="local") + a.add_node(_node("person:ada", "Ada")) + a.retract_node("person:ada") # observed the only tag + ids = {n.id for n in materialize(a.log.ops).nodes.values()} + assert "person:ada" not in ids From d2608a2d86f0a1d979a9e22db612cf0a0ac1f7d4 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Fri, 3 Jul 2026 00:43:13 -0400 Subject: [PATCH 10/12] =?UTF-8?q?feat(pkg):=20CRDT=20Slice=204=20=E2=80=94?= =?UTF-8?q?=20locus/attestation=20merge=20gate=20(the=20correctness=20moat?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit gate() splits a converged op-log into a canonical view (ops from local/trusted_private, or attested_fog/burst_cloud WITH a passing correctness receipt) vs. a quarantine view (everything else, retained not authoritative). Fail-closed: unattested gated-locus ops are quarantined, so an untrusted retract cannot silently delete canonical data. Emits a SyncCycleReceipt-shaped receipt per cycle. This is where TEE-confidentiality ends and correctness-attestation begins — the moat, made executable. 7 tests incl. fail-closed admission, attested-admit, the untrusted-retract security property, longevity (nothing lost), and SyncCycleReceipt v2 conformance. Suite 81 green. --- src/prophet_mesh/pkg_gate.py | 155 +++++++++++++++++++++++++++++++++++ tests/test_pkg_gate.py | 125 ++++++++++++++++++++++++++++ 2 files changed, 280 insertions(+) create mode 100644 src/prophet_mesh/pkg_gate.py create mode 100644 tests/test_pkg_gate.py diff --git a/src/prophet_mesh/pkg_gate.py b/src/prophet_mesh/pkg_gate.py new file mode 100644 index 0000000..b6f80ba --- /dev/null +++ b/src/prophet_mesh/pkg_gate.py @@ -0,0 +1,155 @@ +"""CRDT Slice 4 — the locus / attestation merge gate. + +Slices 1–3 make the graph *converge* (everyone sees the same ops). Slice 4 decides +*which converged sub-graph is authoritative*. This is where the correctness moat +plugs in: TEE/confidentiality proves an op ran privately; it says nothing about +whether the computation behind it is correct. The gate admits an op into the +**canonical view** iff: + + * its ``locus`` is ``local`` or ``trusted_private`` (authored where we already + trust the execution locus), OR + * its ``locus`` is ``attested_fog`` / ``burst_cloud`` **and** its + ``attestationRef`` resolves to a **passing correctness receipt**. + +Everything else lands in the **quarantine view** — still converged and retained +(longevity), just not authoritative. Fail-closed: a gated-locus op with no / +unresolved attestation is quarantined, and an untrusted retract therefore cannot +silently delete canonical data. + +Emits a ``SyncCycleReceipt``-shaped receipt (sourceos-spec v2) per gate cycle. +Accretive: layers over pkg_ops, touches nothing below it. +""" +from __future__ import annotations + +import hashlib +import json +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Callable, Iterable + +from .pkg import PKG +from .pkg_ops import materialize, OpLog + +# Loci that require a passing correctness receipt to reach the canonical view. +GATED_LOCI = ("attested_fog", "burst_cloud") +TRUSTED_LOCI = ("local", "trusted_private") + +# A resolver answers: does this attestationRef name a *passing* correctness receipt? +Resolver = Callable[[str], bool] + + +def deny_all(_ref: str) -> bool: + """Default resolver — fail-closed. Nothing from a gated locus is admitted.""" + return False + + +def passing(*refs: str) -> Resolver: + """Resolver that admits exactly the given (passing) attestation refs.""" + allowed = set(refs) + return lambda ref: ref in allowed + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _canon(obj) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def graph_version(g: PKG) -> str: + """Content hash of materialized graph state — the SyncCycleReceipt version.""" + nodes = sorted((n.id, n.label) for n in g.nodes.values()) + edges = sorted((e.src, e.rel, e.dst) for e in g.edges) + return "sha256:" + hashlib.sha256(_canon({"nodes": nodes, "edges": edges}).encode("utf-8")).hexdigest() + + +def admit(env: dict, resolve: Resolver) -> bool: + """Admission rule for a single op (fail-closed on gated loci).""" + crdt = env["payload"]["crdt"] + locus = crdt["locus"] + if locus in TRUSTED_LOCI: + return True + if locus in GATED_LOCI: + ref = crdt.get("attestationRef") + return bool(ref) and resolve(ref) + return False # unknown locus → fail-closed + + +@dataclass +class GateResult: + canonical: PKG # authoritative graph + quarantine: PKG # held-back view (retained, not authoritative) + canonical_ops: list[dict] + quarantine_ops: list[dict] + receipt: dict # SyncCycleReceipt-shaped + + @property + def admitted(self) -> int: + return len(self.canonical_ops) + + @property + def quarantined(self) -> int: + return len(self.quarantine_ops) + + +def gate( + ops: Iterable[dict], + *, + resolve: Resolver = deny_all, + self_id: str = "self", + org: str = "self", + content_view: str | None = None, + lifecycle_env: str = "candidate", + run_locus: str = "trusted_private", + from_version: str | None = None, + cycle_id: str | None = None, + agentplane_run_ref: str | None = None, +) -> GateResult: + """Split a converged op-log into canonical vs. quarantine and emit a receipt.""" + ops = ops.ops if isinstance(ops, OpLog) else list(ops) + canonical_ops = [e for e in ops if admit(e, resolve)] + quarantine_ops = [e for e in ops if not admit(e, resolve)] + + canonical = materialize(canonical_ops, self_id) + quarantine = materialize(quarantine_ops, self_id) + to_version = graph_version(canonical) + + n_admit, n_quar = len(canonical_ops), len(quarantine_ops) + if not ops: + outcome, gate_val, reason = "skipped", "no-op", "empty op-log" + elif n_admit == 0: + outcome, gate_val, reason = "denied", "denied", f"all {n_quar} op(s) held pending correctness attestation" + elif n_quar: + outcome, gate_val, reason = "applied", "partial", f"{n_admit} admitted, {n_quar} quarantined pending attestation" + else: + outcome, gate_val, reason = "applied", "allowed", f"{n_admit} op(s) admitted" + + steps = [{"step": "admit", "status": "ok", "reason": f"{n_admit} op(s) → canonical"}] + if n_quar: + steps.append({"step": "quarantine", "status": "skipped", + "reason": f"{n_quar} op(s) → quarantine (unattested gated-locus)"}) + + receipt = { + "id": f"urn:srcos:sync-receipt:{uuid.uuid4()}", + "type": "SyncCycleReceipt", + "specVersion": "2.0.0", + "cycleId": cycle_id or f"pkg-gate-{uuid.uuid4()}", + "engineId": "sourceos.sync.pkg-crdt-gate", + "org": org, + "contentView": content_view or f"pkg-{self_id}", + "fromVersion": from_version, + "toVersion": to_version, + "lifecycleEnv": lifecycle_env, + "locus": run_locus, + "outcome": outcome, + "policyGate": gate_val, + "policyReason": reason, + "steps": steps, + "issuedAt": _now(), + "auditId": f"urn:srcos:audit:{uuid.uuid4()}", + "agentplaneRunRef": agentplane_run_ref, + } + + return GateResult(canonical, quarantine, canonical_ops, quarantine_ops, receipt) diff --git a/tests/test_pkg_gate.py b/tests/test_pkg_gate.py new file mode 100644 index 0000000..320392e --- /dev/null +++ b/tests/test_pkg_gate.py @@ -0,0 +1,125 @@ +"""CRDT Slice 4 — locus / attestation merge gate. + +Proves: trusted loci admitted, gated loci fail-closed without a passing receipt, +attested gated ops admitted, an untrusted retract cannot delete canonical data, +nothing is lost, and the emitted receipt conforms to SyncCycleReceipt v2. +""" +from __future__ import annotations + +from prophet_mesh.pkg import PKG, Node, Edge, Provenance +from prophet_mesh.pkg_ops import EmittingPKG, merge +from prophet_mesh.pkg_gate import gate, passing, deny_all + +PASS_REF = "urn:srcos:reasoning-receipt:pass-001" + + +def _prov(src="workspace:contacts"): + return Provenance(source=src, method="ingested") + + +def _node(nid, label, src="workspace:contacts"): + return Node(id=nid, type="Person", label=label, provenance=_prov(src)) + + +def _trusted(): + t = EmittingPKG.seeded("self", replica_id="device-a", locus="trusted_private") + t.add_node(_node("person:ada", "Ada")) + return t + + +def _cloud(): + """A burst_cloud replica emitting one attested + one unattested claim.""" + c = EmittingPKG(PKG(self_id="self"), replica_id="cloud-1", locus="burst_cloud") + c.add_node(_node("claim:cpi", "CPI figure", src="cloud:analysis"), attestation_ref=PASS_REF) + c.add_node(_node("claim:gdp", "GDP figure", src="cloud:analysis")) # no attestation + return c + + +def _ids(g: PKG): + return {n.id for n in g.nodes.values()} + + +# ── admission ─────────────────────────────────────────────────────────────── +def test_trusted_loci_are_admitted(): + t = _trusted() + res = gate(t.log.ops) # default deny_all resolver — but these are trusted_private + assert _ids(res.canonical) == {"self", "person:ada"} + assert res.quarantined == 0 + + +def test_unattested_gated_op_is_quarantined_fail_closed(): + merged = merge(_trusted().log, _cloud().log) + res = gate(merged, resolve=deny_all) # nothing from cloud can pass + assert "claim:gdp" not in _ids(res.canonical) + assert "claim:cpi" not in _ids(res.canonical) # even attested — resolver denies + assert "claim:gdp" in _ids(res.quarantine) + + +def test_attested_gated_op_is_admitted(): + merged = merge(_trusted().log, _cloud().log) + res = gate(merged, resolve=passing(PASS_REF)) + assert "claim:cpi" in _ids(res.canonical) # attested → canonical + assert "claim:gdp" not in _ids(res.canonical) # unattested → still quarantined + assert "claim:gdp" in _ids(res.quarantine) + + +# ── security property ───────────────────────────────────────────────────────── +def test_untrusted_retract_cannot_delete_canonical_data(): + t = _trusted() # self + Ada (trusted_private) + ada_add = [e for e in t.log.ops if e["payload"].get("node", {}).get("id") == "person:ada"][0] + + attacker = EmittingPKG(PKG(self_id="self"), replica_id="cloud-evil", locus="burst_cloud") + attacker.apply_remote(ada_add) # observes Ada's tag + attacker.retract_node("person:ada") # unattested burst_cloud retract + + merged = merge(t.log, attacker.log) + res = gate(merged, resolve=deny_all) + assert "person:ada" in _ids(res.canonical) # the untrusted retract was quarantined + assert res.quarantined >= 1 + + +# ── longevity: nothing lost ────────────────────────────────────────────────── +def test_nothing_is_lost(): + merged = merge(_trusted().log, _cloud().log) + res = gate(merged, resolve=passing(PASS_REF)) + assert res.admitted + res.quarantined == len(merged) + + +# ── receipt conformance ────────────────────────────────────────────────────── +_ALLOWED = { # SyncCycleReceipt.json top-level keys (additionalProperties:false) + "id", "type", "specVersion", "cycleId", "engineId", "org", "contentView", + "fromVersion", "toVersion", "lifecycleEnv", "locus", "outcome", "policyGate", + "policyReason", "steps", "nixCacheUrl", "flakeRef", "durationMs", "issuedAt", + "auditId", "agentplaneRunRef", +} +_REQUIRED = {"id", "type", "specVersion", "cycleId", "engineId", "org", "contentView", + "toVersion", "lifecycleEnv", "locus", "outcome", "steps", "issuedAt", "auditId"} +_OUTCOMES = {"planned", "dry_run", "applied", "skipped", "denied", "failed"} +_STEP_STATUS = {"dry_run", "ok", "failed", "skipped", "timeout"} +_LOCI = {"local", "trusted_private", "attested_fog", "burst_cloud"} + + +def test_receipt_is_synccyclereceipt_conformant(): + res = gate(merge(_trusted().log, _cloud().log), resolve=passing(PASS_REF)) + r = res.receipt + assert set(r.keys()) <= _ALLOWED + assert _REQUIRED <= set(r.keys()) + assert r["type"] == "SyncCycleReceipt" + assert r["id"].startswith("urn:srcos:sync-receipt:") + assert r["engineId"].startswith("sourceos.sync.") + assert r["auditId"].startswith("urn:srcos:audit:") + assert r["outcome"] in _OUTCOMES + assert r["locus"] in _LOCI + assert r["toVersion"].startswith("sha256:") + for step in r["steps"]: + assert {"step", "status"} <= set(step.keys()) + assert step["status"] in _STEP_STATUS + + +def test_receipt_outcome_reflects_gate_decision(): + # all quarantined → denied; mixed → applied/partial + denied = gate(_cloud().log, resolve=deny_all).receipt + assert denied["outcome"] == "denied" and denied["policyGate"] == "denied" + + mixed = gate(merge(_trusted().log, _cloud().log), resolve=passing(PASS_REF)).receipt + assert mixed["outcome"] == "applied" and mixed["policyGate"] == "partial" From 54be26c880284ded40e4678caa8c4cffc11d7008 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Fri, 3 Jul 2026 00:47:39 -0400 Subject: [PATCH 11/12] =?UTF-8?q?feat(pkg):=20CRDT=20Slice=205=20=E2=80=94?= =?UTF-8?q?=20live=20replicas=20+=20HellGraph=20ingester=20(design=20compl?= =?UTF-8?q?ete)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replica: a live device/agent that syncs with peers by anti-entropy (exchange only missing ops), converging two live participants with no coordinator. sync() is bidirectional + idempotent; Lamport advances on receive so later local edits win. replay_to_hellgraph(): the ingester the writer promised but never had (gap #4) — replays a materialized PKG through a HellGraphStore-shaped sink (addNode/addEdge), idempotent via content-addressing. persist_canonical() composes it with the Slice-4 gate so ONLY the proven-correct sub-graph reaches the store; unattested cloud ops converge into every log (longevity) but are never persisted. 6 tests: live convergence, sync idempotence, post-sync propagation, replay + idempotent replay, and the end-to-end 'only canonical graph reaches HellGraph'. Suite 87 green. All 5 CRDT slices landed. --- src/prophet_mesh/pkg_replica.py | 129 ++++++++++++++++++++++++++++++++ tests/test_pkg_replica.py | 126 +++++++++++++++++++++++++++++++ 2 files changed, 255 insertions(+) create mode 100644 src/prophet_mesh/pkg_replica.py create mode 100644 tests/test_pkg_replica.py diff --git a/src/prophet_mesh/pkg_replica.py b/src/prophet_mesh/pkg_replica.py new file mode 100644 index 0000000..3ab89b8 --- /dev/null +++ b/src/prophet_mesh/pkg_replica.py @@ -0,0 +1,129 @@ +"""CRDT Slice 5 — live replicas + the HellGraph ingester. + +Slices 1–4 are pure functions over op-logs. Slice 5 makes them *live*: + + * ``Replica`` — a device/agent that holds a PKG + op-log and syncs with peers by + **anti-entropy** (exchange only the ops the other side is missing), converging + two live participants without any central coordinator. + * ``replay_to_hellgraph`` — the ingester the writer docstring promised but never + had (gap #4): replays a materialized PKG through a ``HellGraphStore``-shaped + sink (``addNode`` / ``addEdge``), so the op-log is no longer emit-only — it + lands on the real substrate. + +The two compose into the punchline: sync converges *everything* (longevity), the +Slice-4 gate keeps only the proven-correct sub-graph, and **only that canonical +graph is replayed to HellGraph.** Unattested cloud ops live in every replica's log +but never reach the authoritative store. + +Accretive: reuses pkg_ops (merge/materialize/apply_remote), pkg_gate (gate), and +pkg_hellgraph (to_hellgraph). Nothing below is modified. +""" +from __future__ import annotations + +from typing import Iterable, Protocol, runtime_checkable + +from .pkg import PKG, Node, Edge +from .pkg_ops import EmittingPKG, EdgeKey, materialize +from .pkg_gate import gate, GateResult, Resolver, deny_all +from .pkg_hellgraph import to_hellgraph + + +class Replica: + """A live CRDT participant. Local edits emit ops; ``sync`` reconciles with a + peer via anti-entropy. The authoritative view is always ``gate(self.ops)``.""" + + def __init__(self, replica_id: str, *, locus: str = "local", self_id: str = "self", seed: bool = True) -> None: + self.replica_id = replica_id + if seed: + self.epkg = EmittingPKG.seeded(self_id, replica_id=replica_id, locus=locus) + else: + self.epkg = EmittingPKG(PKG(self_id=self_id), replica_id=replica_id, locus=locus) + self._seen: set[str] = {e["eventId"] for e in self.epkg.log.ops} + + # ── local mutations (delegate + track) ────────────────────────────────── + def _track(self, env: dict) -> dict: + self._seen.add(env["eventId"]) + return env + + def add_node(self, n: Node, **kw) -> dict: + return self._track(self.epkg.add_node(n, **kw)) + + def add_edge(self, e: Edge, **kw) -> dict: + return self._track(self.epkg.add_edge(e, **kw)) + + def retract_node(self, node_id: str, **kw) -> dict: + return self._track(self.epkg.retract_node(node_id, **kw)) + + def retract_edge(self, key: EdgeKey, **kw) -> dict: + return self._track(self.epkg.retract_edge(key, **kw)) + + @property + def ops(self) -> list[dict]: + return self.epkg.log.ops + + # ── anti-entropy ───────────────────────────────────────────────────────── + def delta_for(self, have: set[str]) -> list[dict]: + """Ops this replica holds that the peer (``have`` = its seen-set) lacks.""" + return [e for e in self.ops if e["eventId"] not in have] + + def receive(self, ops: Iterable[dict]) -> int: + """Ingest a peer's ops we haven't seen: apply to view + tag trackers and + incorporate into our log (multi-head DAG). Returns the count newly applied.""" + new = [e for e in ops if e["eventId"] not in self._seen] + max_lamport = self.epkg.log._lamport + for env in new: + self.epkg.apply_remote(env) + self.epkg.log.ops.append(env) + self._seen.add(env["eventId"]) + max_lamport = max(max_lamport, env["payload"]["crdt"]["lamport"]) + self.epkg.log._lamport = max_lamport # Lamport advance so later local edits win + return len(new) + + def materialized(self, self_id: str = "self") -> PKG: + """Fully converged view (pre-gate).""" + return materialize(self.ops, self_id=self_id) + + def gated(self, *, resolve: Resolver = deny_all, **kw) -> GateResult: + """Authoritative view + quarantine + receipt (Slice-4 gate over our log).""" + return gate(self.ops, resolve=resolve, self_id=self.epkg.g.self_id, **kw) + + +def sync(a: Replica, b: Replica) -> tuple[int, int]: + """Bidirectional anti-entropy: after this, both replicas hold the same op-set + and therefore ``materialize`` to the same state. Returns (a→b, b→a) counts.""" + to_b = a.delta_for(b._seen) + to_a = b.delta_for(a._seen) + n_ab = b.receive(to_b) + n_ba = a.receive(to_a) + return n_ab, n_ba + + +# ── the live HellGraph ingester (closes gap #4) ───────────────────────────── +@runtime_checkable +class HellGraphSink(Protocol): + """The subset of the TS ``HellGraphStore`` façade an ingester needs. The real + implementation is an HTTP client to ``@socioprophet/hellgraph``; tests use an + in-memory double.""" + + def addNode(self, id: str, labels: list[str], properties: dict, createdAt: str | None = None) -> None: ... + def addEdge(self, label: str, frm: str, to: str, properties: dict) -> None: ... + + +def replay_to_hellgraph(g: PKG, sink: HellGraphSink) -> dict[str, int]: + """Replay a materialized PKG through the façade sink (nodes before edges, so + endpoints exist). Idempotent because HellGraph atoms are content-addressed.""" + bundle = to_hellgraph(g) + for n in bundle["nodes"]: + sink.addNode(n["id"], n["labels"], n["properties"], n.get("createdAt")) + for e in bundle["edges"]: + sink.addEdge(e["label"], e["from"], e["to"], e["properties"]) + return {"nodes": len(bundle["nodes"]), "edges": len(bundle["edges"])} + + +def persist_canonical(replica: Replica, sink: HellGraphSink, *, resolve: Resolver = deny_all, **gate_kw) -> tuple[GateResult, dict[str, int]]: + """End-to-end: gate the replica's converged log, then replay ONLY the canonical + (proven-correct) graph to HellGraph. The quarantine view is retained in the log + but never persisted. Returns (gate result incl. receipt, replay counts).""" + res = replica.gated(resolve=resolve, **gate_kw) + counts = replay_to_hellgraph(res.canonical, sink) + return res, counts diff --git a/tests/test_pkg_replica.py b/tests/test_pkg_replica.py new file mode 100644 index 0000000..b6ce67f --- /dev/null +++ b/tests/test_pkg_replica.py @@ -0,0 +1,126 @@ +"""CRDT Slice 5 — live replicas + HellGraph ingester. + +Proves two live replicas converge via anti-entropy, the ingester replays a PKG +through a HellGraphStore-shaped sink (idempotently), and end-to-end only the +gated-canonical graph reaches the store. +""" +from __future__ import annotations + +from prophet_mesh.pkg import PKG, Node, Edge, Provenance +from prophet_mesh.pkg_replica import ( + Replica, sync, replay_to_hellgraph, persist_canonical, HellGraphSink, +) +from prophet_mesh.pkg_gate import passing, deny_all + +PASS_REF = "urn:srcos:reasoning-receipt:pass-001" + + +def _prov(src="workspace:contacts"): + return Provenance(source=src, method="ingested") + + +def _node(nid, label, src="workspace:contacts"): + return Node(id=nid, type="Person", label=label, provenance=_prov(src)) + + +def _edge(src, rel, dst): + return Edge(src=src, dst=dst, rel=rel, provenance=_prov()) + + +def _state(g: PKG): + return (frozenset((n.id, n.label) for n in g.nodes.values()), + frozenset((e.src, e.rel, e.dst) for e in g.edges)) + + +class FakeHellGraphStore: + """In-memory double of the TS HellGraphStore façade. Content-addressed: + nodes keyed by id, edges by (from, label, to) — so replay is idempotent.""" + + def __init__(self): + self.nodes: dict[str, dict] = {} + self.edges: dict[tuple, dict] = {} + + def addNode(self, id, labels, properties, createdAt=None): + self.nodes[id] = {"labels": labels, "properties": properties, "createdAt": createdAt} + + def addEdge(self, label, frm, to, properties): + self.edges[(frm, label, to)] = properties + + +# ── live convergence ───────────────────────────────────────────────────────── +def test_two_live_replicas_converge_via_sync(): + a = Replica("device-a", locus="trusted_private") + b = Replica("device-b", locus="trusted_private") + a.add_node(_node("person:ada", "Ada")); a.add_edge(_edge("self", "knows", "person:ada")) + b.add_node(_node("person:bob", "Bob")); b.add_edge(_edge("self", "knows", "person:bob")) + + n_ab, n_ba = sync(a, b) + assert n_ab > 0 and n_ba > 0 + assert a._seen == b._seen # anti-entropy complete + assert _state(a.materialized()) == _state(b.materialized()) + ids = {n.id for n in a.materialized().nodes.values()} + assert ids == {"self", "person:ada", "person:bob"} + + +def test_sync_is_idempotent_second_pass_transfers_nothing(): + a = Replica("device-a", locus="trusted_private") + b = Replica("device-b", locus="trusted_private") + a.add_node(_node("person:ada", "Ada")) + sync(a, b) + assert sync(a, b) == (0, 0) # nothing left to exchange + + +def test_local_edit_after_sync_propagates_back(): + a = Replica("device-a", locus="trusted_private") + b = Replica("device-b", locus="trusted_private") + sync(a, b) + b.add_node(_node("person:cyd", "Cyd")) # B edits after first sync + sync(a, b) + assert "person:cyd" in {n.id for n in a.materialized().nodes.values()} + + +# ── the ingester (gap #4) ──────────────────────────────────────────────────── +def test_replay_persists_to_hellgraph_sink(): + a = Replica("device-a", locus="trusted_private") + a.add_node(_node("person:ada", "Ada")); a.add_edge(_edge("self", "knows", "person:ada")) + store = FakeHellGraphStore() + assert isinstance(store, HellGraphSink) # structural conformance + counts = replay_to_hellgraph(a.materialized(), store) + assert "person:ada" in store.nodes and "self" in store.nodes + assert ("self", "knows", "person:ada") in store.edges + assert counts["nodes"] == len(store.nodes) + + +def test_replay_is_idempotent(): + a = Replica("device-a", locus="trusted_private") + a.add_node(_node("person:ada", "Ada")); a.add_edge(_edge("self", "knows", "person:ada")) + store = FakeHellGraphStore() + replay_to_hellgraph(a.materialized(), store) + n1, e1 = len(store.nodes), len(store.edges) + replay_to_hellgraph(a.materialized(), store) # replay again + assert (len(store.nodes), len(store.edges)) == (n1, e1) + + +# ── end-to-end: only the canonical graph is persisted ──────────────────────── +def test_only_gated_canonical_graph_reaches_the_store(): + a = Replica("device-a", locus="trusted_private") + a.add_node(_node("person:ada", "Ada")) + + cloud = Replica("cloud-1", locus="burst_cloud", seed=False) + cloud.add_node(_node("claim:cpi", "CPI", src="cloud:analysis"), attestation_ref=PASS_REF) + cloud.add_node(_node("claim:gdp", "GDP", src="cloud:analysis")) # unattested + + sync(a, cloud) # everything converges into both logs + # both unattested + attested claims are present in the converged (pre-gate) view + conv_ids = {n.id for n in a.materialized().nodes.values()} + assert {"claim:cpi", "claim:gdp"} <= conv_ids + + store = FakeHellGraphStore() + res, counts = persist_canonical(a, store, resolve=passing(PASS_REF)) + + persisted = set(store.nodes) + assert "person:ada" in persisted # trusted → persisted + assert "claim:cpi" in persisted # attested → persisted + assert "claim:gdp" not in persisted # unattested → quarantined, NOT persisted + assert res.quarantined >= 1 + assert res.receipt["type"] == "SyncCycleReceipt" From 19b8af412b5af7427c63e329671574eb2368e26d Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Sat, 4 Jul 2026 12:41:30 -0400 Subject: [PATCH 12/12] lint: satisfy ruff on pkg_gate/pkg_ops + pkg tests (F401 + E702) The test job (ruff check src tests) was red: 4 unused imports (F401) in pkg_gate.py and the pkg tests, and 7 semicolon-joined statements (E702) in pkg_ops.py and test_pkg_replica.py. Removed the unused imports and split the compound statements onto their own lines. ruff clean + pytest green, no behaviour change. --- src/prophet_mesh/pkg_gate.py | 2 +- src/prophet_mesh/pkg_ops.py | 9 ++++++--- tests/test_pkg_gate.py | 2 +- tests/test_pkg_merge.py | 2 +- tests/test_pkg_replica.py | 14 +++++++++----- 5 files changed, 18 insertions(+), 11 deletions(-) diff --git a/src/prophet_mesh/pkg_gate.py b/src/prophet_mesh/pkg_gate.py index b6f80ba..82d9eac 100644 --- a/src/prophet_mesh/pkg_gate.py +++ b/src/prophet_mesh/pkg_gate.py @@ -24,7 +24,7 @@ import hashlib import json import uuid -from dataclasses import dataclass, field +from dataclasses import dataclass from datetime import datetime, timezone from typing import Callable, Iterable diff --git a/src/prophet_mesh/pkg_ops.py b/src/prophet_mesh/pkg_ops.py index e77ee69..6194212 100644 --- a/src/prophet_mesh/pkg_ops.py +++ b/src/prophet_mesh/pkg_ops.py @@ -201,7 +201,8 @@ def apply_remote(self, env: dict) -> None: self.g.add_node(_node_from(p["node"])) self._node_tags.setdefault(p["node"]["id"], set()).add(crdt["tag"]) elif op == "GraphEdgeAsserted": - ed = p["edge"]; key = (ed["src"], ed["rel"], ed["dst"]) + ed = p["edge"] + key = (ed["src"], ed["rel"], ed["dst"]) self.g.add_edge(_edge_from(ed)) self._edge_tags.setdefault(key, set()).add(crdt["tag"]) elif op == "GraphNodeRetracted": @@ -211,7 +212,8 @@ def apply_remote(self, env: dict) -> None: self.g.nodes.pop(nid, None) self._node_tags.pop(nid, None) elif op == "GraphEdgeRetracted": - key = tuple(p["edgeKey"]); rem = set(crdt.get("removes", [])) + key = tuple(p["edgeKey"]) + rem = set(crdt.get("removes", [])) self._edge_tags[key] = self._edge_tags.get(key, set()) - rem if not self._edge_tags.get(key): self.g.edges = [e for e in self.g.edges if (e.src, e.rel, e.dst) != key] @@ -246,7 +248,8 @@ def materialize(ops: Iterable[dict], self_id: str = "self") -> PKG: node_adds.setdefault(p["node"]["id"], []).append( (crdt["lamport"], env["eventId"], p["node"], crdt["tag"])) elif op == "GraphEdgeAsserted": - e = p["edge"]; key = (e["src"], e["rel"], e["dst"]) + e = p["edge"] + key = (e["src"], e["rel"], e["dst"]) edge_adds.setdefault(key, []).append( (crdt["lamport"], env["eventId"], e, crdt["tag"])) elif op in ("GraphNodeRetracted", "GraphEdgeRetracted"): diff --git a/tests/test_pkg_gate.py b/tests/test_pkg_gate.py index 320392e..d3d4096 100644 --- a/tests/test_pkg_gate.py +++ b/tests/test_pkg_gate.py @@ -6,7 +6,7 @@ """ from __future__ import annotations -from prophet_mesh.pkg import PKG, Node, Edge, Provenance +from prophet_mesh.pkg import PKG, Node, Provenance from prophet_mesh.pkg_ops import EmittingPKG, merge from prophet_mesh.pkg_gate import gate, passing, deny_all diff --git a/tests/test_pkg_merge.py b/tests/test_pkg_merge.py index f55983c..6868828 100644 --- a/tests/test_pkg_merge.py +++ b/tests/test_pkg_merge.py @@ -8,7 +8,7 @@ import random from prophet_mesh.pkg import PKG, Node, Edge, Provenance -from prophet_mesh.pkg_ops import EmittingPKG, materialize, merge, fold +from prophet_mesh.pkg_ops import EmittingPKG, materialize, merge def _prov(src="workspace:contacts"): diff --git a/tests/test_pkg_replica.py b/tests/test_pkg_replica.py index b6ce67f..dad9ff7 100644 --- a/tests/test_pkg_replica.py +++ b/tests/test_pkg_replica.py @@ -10,7 +10,7 @@ from prophet_mesh.pkg_replica import ( Replica, sync, replay_to_hellgraph, persist_canonical, HellGraphSink, ) -from prophet_mesh.pkg_gate import passing, deny_all +from prophet_mesh.pkg_gate import passing PASS_REF = "urn:srcos:reasoning-receipt:pass-001" @@ -51,8 +51,10 @@ def addEdge(self, label, frm, to, properties): def test_two_live_replicas_converge_via_sync(): a = Replica("device-a", locus="trusted_private") b = Replica("device-b", locus="trusted_private") - a.add_node(_node("person:ada", "Ada")); a.add_edge(_edge("self", "knows", "person:ada")) - b.add_node(_node("person:bob", "Bob")); b.add_edge(_edge("self", "knows", "person:bob")) + a.add_node(_node("person:ada", "Ada")) + a.add_edge(_edge("self", "knows", "person:ada")) + b.add_node(_node("person:bob", "Bob")) + b.add_edge(_edge("self", "knows", "person:bob")) n_ab, n_ba = sync(a, b) assert n_ab > 0 and n_ba > 0 @@ -82,7 +84,8 @@ def test_local_edit_after_sync_propagates_back(): # ── the ingester (gap #4) ──────────────────────────────────────────────────── def test_replay_persists_to_hellgraph_sink(): a = Replica("device-a", locus="trusted_private") - a.add_node(_node("person:ada", "Ada")); a.add_edge(_edge("self", "knows", "person:ada")) + a.add_node(_node("person:ada", "Ada")) + a.add_edge(_edge("self", "knows", "person:ada")) store = FakeHellGraphStore() assert isinstance(store, HellGraphSink) # structural conformance counts = replay_to_hellgraph(a.materialized(), store) @@ -93,7 +96,8 @@ def test_replay_persists_to_hellgraph_sink(): def test_replay_is_idempotent(): a = Replica("device-a", locus="trusted_private") - a.add_node(_node("person:ada", "Ada")); a.add_edge(_edge("self", "knows", "person:ada")) + a.add_node(_node("person:ada", "Ada")) + a.add_edge(_edge("self", "knows", "person:ada")) store = FakeHellGraphStore() replay_to_hellgraph(a.materialized(), store) n1, e1 = len(store.nodes), len(store.edges)