diff --git a/docs/estate-rollout.md b/docs/estate-rollout.md new file mode 100644 index 0000000..04af59a --- /dev/null +++ b/docs/estate-rollout.md @@ -0,0 +1,199 @@ +# SourceOS Estate Rollout Guide + +**Scope:** Deploying `sourceos-syncd` across a fleet of managed workstations +**Implements:** sourceos-syncd#2 + +--- + +## Overview + +`sourceos-syncd` is deployed as a systemd service on each managed device. +It owns the local State Integrity lane, syncs content views via Katello/Foreman, and surfaces operator narratives to the SocioSphere dashboard. + +The rollout follows a three-phase model: canary → ring-A → ring-B/full estate. + +--- + +## Prerequisites + +| Requirement | Notes | +|---|---| +| NixOS ≥ 24.05 (aarch64 or x86_64) | Tested on Asahi and standard x86 | +| Katello/Foreman reachable at `$KATELLO_URL` | HTTPS, self-signed cert supported via `--no-verify-ssl` for dev | +| `sourceos-syncd` v0.3.x RPM in the content view | Built by the `sourceos-builder-aarch64` view | +| `KATELLO_PASSWORD_FILE` set or `KATELLO_PASSWORD` env | Password at rest in a secret store, not in the unit file | +| `sourceos-syncd status --store-root /var/lib/sourceos-syncd` exits 0 | Pre-check before promotion | + +--- + +## Step 1 — Prepare the content view + +```bash +# Plan what will be applied +sourceos-syncd sync plan \ + --katello-url "$KATELLO_URL" \ + --org SocioProphet \ + --content-view sourceos-builder-aarch64 \ + --lifecycle-env candidate +``` + +Review the output. `policy_gate` must be `"allowed"` before proceeding. +Gate on the signing key if you have one: + +```bash +sourceos-syncd sync plan \ + --signing-public-key "RWS..." \ + ... +``` + +--- + +## Step 2 — Deploy to canary (1–3 devices) + +Choose canary hosts. On each: + +```bash +# Dry-run first (no changes) +sourceos-syncd sync apply \ + --katello-url "$KATELLO_URL" \ + --org SocioProphet \ + --content-view sourceos-builder-aarch64 \ + --lifecycle-env candidate \ + --store-root /var/lib/sourceos-syncd + +# Promote when dry-run looks good +sourceos-syncd sync apply --execute \ + --katello-url "$KATELLO_URL" \ + --org SocioProphet \ + --content-view sourceos-builder-aarch64 \ + --lifecycle-env candidate \ + --store-root /var/lib/sourceos-syncd +``` + +Enable and start the daemon: + +```bash +systemctl enable --now sourceos-syncd +``` + +Verify the canary: + +```bash +sourceos-syncd sync check-health --store-root /var/lib/sourceos-syncd +sourceos-syncd sync status --store-root /var/lib/sourceos-syncd +sourceos-syncd health snapshot +``` + +All three must return exit 0 before ring-A promotion. + +--- + +## Step 3 — Ring-A rollout (10–20% of estate) + +Promote the content view to the `stable` lifecycle env: + +```bash +sourceos-syncd sync plan \ + --lifecycle-env stable \ + ... +``` + +Roll out to ring-A hosts using your configuration management tool (Ansible, Puppet, NixOS colmena). +The daemon can also be driven purely from environment variables for automation: + +```bash +KATELLO_URL=https://katello.internal:8443 \ +KATELLO_USER=admin \ +KATELLO_PASSWORD_FILE=/run/secrets/katello-password \ +KATELLO_ORG=SocioProphet \ +KATELLO_CONTENT_VIEW=sourceos-builder-aarch64 \ +KATELLO_LIFECYCLE_ENV=stable \ +sourceos-syncd sync daemon --from-env +``` + +Monitor ring-A health: + +```bash +# On each host +sourceos-syncd sync status +sourceos-syncd authority report +sourceos-syncd coherence status +``` + +Check that `allRequiredReachable` is `true` in the authority report before full rollout. + +--- + +## Step 4 — Full estate rollout + +Once ring-A is stable for ≥ 24 hours with no repair-required events: + +1. Promote the content view to `production` (your final lifecycle env). +2. Roll out to remaining hosts in ring-B using the same pattern. +3. After rollout, run the noise budget check: + +```bash +sourceos-syncd noise-budget status +``` + +`budgetUtilization` above 0.5 indicates event churn worth investigating before considering the rollout complete. + +--- + +## Monitoring and operator cards + +The SocioSphere dashboard consumes operator narrative cards from the `/narrativez` endpoint: + +``` +GET http://127.0.0.1:8765/narrativez +``` + +For estate-level aggregation, a collector scrapes each device's `/narrativez` and `/authorityz` endpoints. The coherence plane snapshot is available at `/coherencez`. + +Prometheus metrics are at `/metrics` (Prometheus text format). The key gauges: + +| Metric | Meaning | +|---|---| +| `sourceos_syncd_ready` | 1 when ready, 0 otherwise | +| `sourceos_syncd_initialized` | 1 when local store is initialized | +| `sourceos_syncd_corrupted_objects` | Count of corrupted store objects | +| `sourceos_syncd_replay_lag_events` | Journal replay lag | + +--- + +## Rollback + +If a host fails health checks after the apply: + +```bash +# Show the last receipt and its outcome +sourceos-syncd receipts last --store-root /var/lib/sourceos-syncd + +# Re-apply the previous version +sourceos-syncd sync apply --execute \ + --current-version \ + --lifecycle-env stable \ + ... +``` + +The daemon tolerates network blips and will retry on the next poll cycle (default 300 s). + +--- + +## Security notes + +- The daemon binds only to `127.0.0.1:8765`; the HTTP service is local-only. +- All harvest envelopes produced by `sourceos-syncd harvest wrap` have `sensitive_data_excluded: true` set unconditionally. Raw prompts, credentials, and session tokens are never included in harvest payloads. +- The Katello password must come from `KATELLO_PASSWORD_FILE` or `KATELLO_PASSWORD`. Never pass it as a CLI argument in production (it appears in `ps` output). + +--- + +## Troubleshooting + +| Symptom | Check | +|---|---| +| `sourceos_syncd_ready 0` in Prometheus | `sourceos-syncd health explain` — look at `diagnosis.status` | +| Daemon not starting | `journalctl -u sourceos-syncd -n 50` | +| `katello_reachable: false` in check-health | Network / TLS cert issue; `--no-verify-ssl` for dev only | +| `allRequiredReachable: false` in authority report | Identity or policy authority is offline; expected at first boot until identity is confirmed | +| High `budgetUtilization` | `sourceos-syncd noise-budget status` then flush: `sourceos-syncd noise-budget flush` | diff --git a/src/sourceos_syncd/authority.py b/src/sourceos_syncd/authority.py new file mode 100644 index 0000000..f7ec78c --- /dev/null +++ b/src/sourceos_syncd/authority.py @@ -0,0 +1,112 @@ +"""Authority dependency binding for sourceos-syncd. + +Binds authority dependencies from the hybrid cybernetic control plane +to SourceOS state-integrity and local repair posture. Implements sourceos-syncd#27. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +def _now() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +AUTHORITY_KINDS = frozenset([ + "identity-authority", + "policy-authority", + "capability-authority", + "memory-authority", + "evidence-authority", + "source-channel-authority", +]) + + +@dataclass(frozen=True) +class AuthorityDependency: + """A declared authority dependency for a state integrity lane.""" + + authority_ref: str + authority_kind: str + lane_ref: str + required: bool + reachable: bool + last_confirmed: str | None + note: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "authorityRef": self.authority_ref, + "authorityKind": self.authority_kind, + "laneRef": self.lane_ref, + "required": self.required, + "reachable": self.reachable, + "lastConfirmed": self.last_confirmed, + "note": self.note, + } + + +@dataclass +class AuthorityReport: + """Report of authority dependency health for the current device.""" + + device_ref: str + dependencies: list[AuthorityDependency] = field(default_factory=list) + all_required_reachable: bool = False + generated_at: str = field(default_factory=_now) + note: str = "Stub — live authority checks deferred to sourceos-syncd daemon per sourceos-syncd#27." + + def to_dict(self) -> dict[str, Any]: + return { + "deviceRef": self.device_ref, + "allRequiredReachable": self.all_required_reachable, + "dependencies": [d.to_dict() for d in self.dependencies], + "generatedAt": self.generated_at, + "note": self.note, + } + + +KNOWN_AUTHORITIES = { + "identity-authority": "urn:srcos:authority:sovereign-identity:default", + "policy-authority": "urn:srcos:authority:policy-fabric:default", + "capability-authority": "urn:srcos:authority:capability-broker:default", + "memory-authority": "urn:srcos:authority:memory-mesh:default", + "evidence-authority": "urn:srcos:authority:agentplane:default", + "source-channel-authority": "urn:srcos:authority:source-channel:default", +} + + +def stub_report(device_ref: str = "unknown") -> AuthorityReport: + """Return a stub authority report when the daemon is not yet running.""" + deps = [ + AuthorityDependency( + authority_ref=ref, + authority_kind=kind, + lane_ref="urn:srcos:lane:state-integrity:default", + required=kind in ("identity-authority", "policy-authority"), + reachable=False, + last_confirmed=None, + note="Stub — live authority probe deferred per sourceos-syncd#27.", + ) + for kind, ref in KNOWN_AUTHORITIES.items() + ] + return AuthorityReport(device_ref=device_ref, dependencies=deps) + + +def check_authority(authority_kind: str, device_ref: str = "unknown") -> AuthorityDependency: + """Probe a single authority dependency. Stub until daemon is live.""" + if authority_kind not in AUTHORITY_KINDS: + raise ValueError(f"Unknown authority kind: {authority_kind}") + ref = KNOWN_AUTHORITIES.get(authority_kind, f"urn:srcos:authority:{authority_kind}:default") + return AuthorityDependency( + authority_ref=ref, + authority_kind=authority_kind, + lane_ref="urn:srcos:lane:state-integrity:default", + required=authority_kind in ("identity-authority", "policy-authority"), + reachable=False, + last_confirmed=None, + note="Stub — live authority probe deferred per sourceos-syncd#27.", + ) diff --git a/src/sourceos_syncd/cli.py b/src/sourceos_syncd/cli.py index 9983e79..8ebe0fb 100644 --- a/src/sourceos_syncd/cli.py +++ b/src/sourceos_syncd/cli.py @@ -23,6 +23,12 @@ from .katello_client import KatelloContentClient from .receipt_store import ReceiptStore from .trust import TrustRequest, evaluate_trust, validate_trust_decision +from . import noise_budget as _noise_budget +from . import narrative as _narrative +from . import systema as _systema +from . import coherence as _coherence +from . import authority as _authority +from . import harvest as _harvest def add_compact(parser: argparse.ArgumentParser) -> None: @@ -206,6 +212,63 @@ def add_katello_args(p: argparse.ArgumentParser) -> None: receipts_last.add_argument("--store-root", default=None, help="store root") add_compact(receipts_last) + noise = subcommands.add_parser("noise-budget", help="event coalescing and noise budget commands") + noise_sub = noise.add_subparsers(dest="command", required=True) + noise_status = noise_sub.add_parser("status", help="show current noise budget utilization") + add_compact(noise_status) + noise_flush = noise_sub.add_parser("flush", help="flush the noise budget window") + add_compact(noise_flush) + + narrative_cmd = subcommands.add_parser("narrative", help="operator narrative summary commands") + narrative_sub = narrative_cmd.add_subparsers(dest="command", required=True) + narr_status = narrative_sub.add_parser("status", help="emit a stub operator narrative for this device") + narr_status.add_argument("--device-ref", default="unknown", help="device URN") + add_compact(narr_status) + + systema_cmd = subcommands.add_parser("systema", help="Systema source-confidence and membrane commands") + systema_sub = systema_cmd.add_subparsers(dest="command", required=True) + sys_conf = systema_sub.add_parser("confidence", help="map a confidence score to SourceOS contracts") + sys_conf.add_argument("score", type=float, help="confidence score 0.0–1.0") + add_compact(sys_conf) + sys_mem = systema_sub.add_parser("membrane", help="map a membrane boundary event") + sys_mem.add_argument("kind", choices=sorted(_systema.MEMBRANE_KINDS), help="membrane kind") + sys_mem.add_argument("decision", choices=sorted(_systema.MEMBRANE_DECISIONS), help="membrane decision") + sys_mem.add_argument("--subject", required=True, help="subject URN") + sys_mem.add_argument("--actor", required=True, help="actor URN") + sys_mem.add_argument("--confidence", type=float, default=0.7, help="confidence score") + sys_mem.add_argument("--policy-decision-ref", default=None, help="optional policy decision URN") + add_compact(sys_mem) + + coherence_cmd = subcommands.add_parser("coherence", help="Metadata Coherence Plane commands") + coherence_sub = coherence_cmd.add_subparsers(dest="command", required=True) + coh_status = coherence_sub.add_parser("status", help="show coherence plane snapshot") + coh_status.add_argument("--device-ref", default="unknown", help="device URN") + add_compact(coh_status) + coh_scan = coherence_sub.add_parser("scan", help="scan a single coherence domain") + coh_scan.add_argument("domain", choices=_coherence.COHERENCE_DOMAINS, help="coherence domain") + coh_scan.add_argument("--device-ref", default="unknown", help="device URN") + add_compact(coh_scan) + + authority_cmd = subcommands.add_parser("authority", help="authority dependency commands") + authority_sub = authority_cmd.add_subparsers(dest="command", required=True) + auth_report = authority_sub.add_parser("report", help="show authority dependency report") + auth_report.add_argument("--device-ref", default="unknown", help="device URN") + add_compact(auth_report) + auth_check = authority_sub.add_parser("check", help="probe a single authority dependency") + auth_check.add_argument("kind", choices=sorted(_authority.AUTHORITY_KINDS), help="authority kind") + auth_check.add_argument("--device-ref", default="unknown", help="device URN") + add_compact(auth_check) + + harvest_cmd = subcommands.add_parser("harvest", help="lawful metadata harvest bridge commands") + harvest_sub = harvest_cmd.add_subparsers(dest="command", required=True) + harv_wrap = harvest_sub.add_parser("wrap", help="wrap an import event in a harvest envelope") + harv_wrap.add_argument("--file", "-f", required=True, help="import event JSON file") + harv_wrap.add_argument("--basis", default="legitimate-interest", choices=sorted(_harvest.HARVEST_CONSENT_BASES), help="harvest consent basis") + harv_wrap.add_argument("--scope", default="metadata-only", choices=sorted(_harvest.HARVEST_SCOPES), help="harvest scope") + harv_wrap.add_argument("--collector-ref", default="urn:srcos:collector:sourceos-syncd:default", help="collector URN") + harv_wrap.add_argument("--retention-policy-ref", default="urn:srcos:retention-policy:default", help="retention policy URN") + add_compact(harv_wrap) + return parser @@ -505,6 +568,69 @@ def main(argv: list[str] | None = None) -> int: sys.stdout.write(pretty_json(receipt, pretty=pretty)) return 0 + if args.area == "noise-budget" and args.command == "status": + report = _noise_budget.budget_report() + sys.stdout.write(pretty_json(report.to_dict(), pretty=pretty)) + return 0 + + if args.area == "noise-budget" and args.command == "flush": + _noise_budget._default_engine.flush() + sys.stdout.write(pretty_json({"flushed": True}, pretty=pretty)) + return 0 + + if args.area == "narrative" and args.command == "status": + narr = _narrative.stub_narrative(device_ref=args.device_ref) + sys.stdout.write(pretty_json(narr.to_dict(), pretty=pretty)) + return 0 + + if args.area == "systema" and args.command == "confidence": + mapping = _systema.map_confidence(args.score) + sys.stdout.write(pretty_json(mapping.to_dict(), pretty=pretty)) + return 0 + + if args.area == "systema" and args.command == "membrane": + event = _systema.map_membrane_event( + args.kind, args.decision, + subject_ref=args.subject, actor_ref=args.actor, + confidence=args.confidence, + policy_decision_ref=args.policy_decision_ref, + ) + sys.stdout.write(pretty_json(event.to_dict(), pretty=pretty)) + return 0 + + if args.area == "coherence" and args.command == "status": + snap = _coherence.stub_snapshot(device_ref=args.device_ref) + sys.stdout.write(pretty_json(snap.to_dict(), pretty=pretty)) + return 0 + + if args.area == "coherence" and args.command == "scan": + state = _coherence.scan_domain(args.domain, device_ref=args.device_ref) + sys.stdout.write(pretty_json(state.to_dict(), pretty=pretty)) + return 0 + + if args.area == "authority" and args.command == "report": + rpt = _authority.stub_report(device_ref=args.device_ref) + sys.stdout.write(pretty_json(rpt.to_dict(), pretty=pretty)) + return 0 + + if args.area == "authority" and args.command == "check": + dep = _authority.check_authority(args.kind, device_ref=args.device_ref) + sys.stdout.write(pretty_json(dep.to_dict(), pretty=pretty)) + return 0 + + if args.area == "harvest" and args.command == "wrap": + with open(args.file) as _f: + event_data = json.load(_f) + envelope = _harvest.wrap_import_event( + event_data, + harvest_basis=args.basis, + scope=args.scope, + collector_ref=args.collector_ref, + retention_policy_ref=args.retention_policy_ref, + ) + sys.stdout.write(pretty_json(envelope.to_dict(), pretty=pretty)) + return 0 + except Exception as exc: # noqa: BLE001 - CLI boundary should present clean error JSON. sys.stderr.write(pretty_json({"error": type(exc).__name__, "message": str(exc)}, pretty=pretty)) return 1 diff --git a/src/sourceos_syncd/coherence.py b/src/sourceos_syncd/coherence.py new file mode 100644 index 0000000..6aa76d5 --- /dev/null +++ b/src/sourceos_syncd/coherence.py @@ -0,0 +1,131 @@ +"""SourceOS Metadata Coherence Plane for sourceos-syncd. + +The typed reconciliation layer for filesystem, registry, extension, container, +persona, cache, privacy, process, and index state. Implements sourceos-syncd#23. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +def _now() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +COHERENCE_DOMAINS = [ + "filesystem", + "registry", + "extension", + "container", + "persona", + "cache", + "privacy", + "process", + "index", +] + + +@dataclass(frozen=True) +class CoherenceState: + """Coherence state for a single domain.""" + + domain: str + status: str + last_checked: str + conflict_count: int + repair_eligible: int + human_review_required: int + note: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "domain": self.domain, + "status": self.status, + "lastChecked": self.last_checked, + "conflictCount": self.conflict_count, + "repairEligible": self.repair_eligible, + "humanReviewRequired": self.human_review_required, + "note": self.note, + } + + +@dataclass +class CoherencePlaneSnapshot: + """Full snapshot of the Metadata Coherence Plane.""" + + device_ref: str + domains: list[CoherenceState] = field(default_factory=list) + overall_status: str = "unavailable" + generated_at: str = field(default_factory=_now) + note: str = "Stub — sourceos-syncd daemon not yet running; coherence scan deferred per sourceos-syncd#23." + + def to_dict(self) -> dict[str, Any]: + return { + "deviceRef": self.device_ref, + "overallStatus": self.overall_status, + "domains": [d.to_dict() for d in self.domains], + "generatedAt": self.generated_at, + "note": self.note, + } + + +@dataclass(frozen=True) +class CoherenceConflict: + """A single coherence conflict in a domain.""" + + conflict_ref: str + domain: str + object_ref: str + staleness_class: str + auto_repair_eligible: bool + human_review_required: bool + local_value_redacted: bool + remote_source_ref: str + observed_at: str + + def to_dict(self) -> dict[str, Any]: + return { + "conflictRef": self.conflict_ref, + "domain": self.domain, + "objectRef": self.object_ref, + "stalenessClass": self.staleness_class, + "autoRepairEligible": self.auto_repair_eligible, + "humanReviewRequired": self.human_review_required, + "localValueRedacted": self.local_value_redacted, + "remoteSourceRef": self.remote_source_ref, + "observedAt": self.observed_at, + } + + +def stub_snapshot(device_ref: str = "unknown") -> CoherencePlaneSnapshot: + """Return a stub snapshot with all domains in unavailable state.""" + domains = [ + CoherenceState( + domain=d, + status="unavailable", + last_checked=_now(), + conflict_count=0, + repair_eligible=0, + human_review_required=0, + ) + for d in COHERENCE_DOMAINS + ] + return CoherencePlaneSnapshot(device_ref=device_ref, domains=domains) + + +def scan_domain(domain: str, device_ref: str = "unknown") -> CoherenceState: + """Scan a single coherence domain. Stub until daemon is live.""" + if domain not in COHERENCE_DOMAINS: + raise ValueError(f"Unknown coherence domain: {domain}. Valid: {COHERENCE_DOMAINS}") + return CoherenceState( + domain=domain, + status="unavailable", + last_checked=_now(), + conflict_count=0, + repair_eligible=0, + human_review_required=0, + note="Stub — live scan deferred to sourceos-syncd daemon per sourceos-syncd#23.", + ) diff --git a/src/sourceos_syncd/harvest.py b/src/sourceos_syncd/harvest.py new file mode 100644 index 0000000..7c79a92 --- /dev/null +++ b/src/sourceos_syncd/harvest.py @@ -0,0 +1,144 @@ +"""Lawful metadata harvesting bridge for sourceos-syncd. + +Binds lawful metadata harvesting to SourceOS import bridge events. +Implements the ProCybernetica harvesting doctrine (PR #59) integration +with sourceos-syncd import events. Implements sourceos-syncd#28. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +def _now() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +HARVEST_CONSENT_BASES = frozenset([ + "explicit-consent", + "legitimate-interest", + "legal-obligation", + "vital-interest", + "public-task", + "contract", +]) + +HARVEST_SCOPES = frozenset([ + "metadata-only", + "structural-only", + "content-summary", + "full-record", +]) + + +@dataclass(frozen=True) +class HarvestEnvelope: + """A lawful metadata harvest envelope bound to an import bridge event.""" + + envelope_ref: str + source_event_ref: str + harvest_basis: str + scope: str + subject_ref: str + collector_ref: str + retention_policy_ref: str + suppressed_fields: list[str] + sensitive_data_excluded: bool + harvested_at: str + payload_hash: str + note: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "envelopeRef": self.envelope_ref, + "sourceEventRef": self.source_event_ref, + "harvestBasis": self.harvest_basis, + "scope": self.scope, + "subjectRef": self.subject_ref, + "collectorRef": self.collector_ref, + "retentionPolicyRef": self.retention_policy_ref, + "suppressedFields": self.suppressed_fields, + "sensitiveDataExcluded": self.sensitive_data_excluded, + "harvestedAt": self.harvested_at, + "payloadHash": self.payload_hash, + "note": self.note, + } + + +@dataclass +class HarvestJournal: + """Local journal of harvest envelopes for a device.""" + + device_ref: str + envelopes: list[HarvestEnvelope] = field(default_factory=list) + total: int = 0 + suppressed_count: int = 0 + note: str = "Stub — live harvest bridge deferred to sourceos-syncd daemon per sourceos-syncd#28." + + def to_dict(self) -> dict[str, Any]: + return { + "deviceRef": self.device_ref, + "total": self.total, + "suppressedCount": self.suppressed_count, + "envelopes": [e.to_dict() for e in self.envelopes], + "note": self.note, + } + + +SENSITIVE_FIELDS = frozenset([ + "raw_content", + "credentials", + "private_key", + "session_token", + "password", + "secret", + "raw_prompt", + "kv_cache", +]) + + +def _payload_hash(payload: dict[str, Any]) -> str: + raw = json.dumps(payload, sort_keys=True).encode() + return "sha256:" + hashlib.sha256(raw).hexdigest() + + +def wrap_import_event( + event: dict[str, Any], + harvest_basis: str = "legitimate-interest", + scope: str = "metadata-only", + collector_ref: str = "urn:srcos:collector:sourceos-syncd:default", + retention_policy_ref: str = "urn:srcos:retention-policy:default", +) -> HarvestEnvelope: + """Wrap an import bridge event in a lawful harvest envelope.""" + if harvest_basis not in HARVEST_CONSENT_BASES: + raise ValueError(f"Unknown harvest basis: {harvest_basis}") + if scope not in HARVEST_SCOPES: + raise ValueError(f"Unknown harvest scope: {scope}") + + redacted = {k: v for k, v in event.items() if k not in SENSITIVE_FIELDS} + suppressed = [k for k in event if k in SENSITIVE_FIELDS] + payload_hash = _payload_hash(redacted) + ts = _now() + envelope_ref = f"urn:srcos:harvest-envelope:{payload_hash[:12]}:{ts.replace(':', '').replace('-', '')[:15]}" + + return HarvestEnvelope( + envelope_ref=envelope_ref, + source_event_ref=event.get("id", "unknown"), + harvest_basis=harvest_basis, + scope=scope, + subject_ref=event.get("subjectRef", event.get("subject_ref", "unknown")), + collector_ref=collector_ref, + retention_policy_ref=retention_policy_ref, + suppressed_fields=suppressed, + sensitive_data_excluded=True, + harvested_at=ts, + payload_hash=payload_hash, + ) + + +def stub_journal(device_ref: str = "unknown") -> HarvestJournal: + return HarvestJournal(device_ref=device_ref) diff --git a/src/sourceos_syncd/narrative.py b/src/sourceos_syncd/narrative.py new file mode 100644 index 0000000..318c35a --- /dev/null +++ b/src/sourceos_syncd/narrative.py @@ -0,0 +1,164 @@ +"""Operator narrative and SocioSphere dashboard contract for sourceos-syncd. + +Converts canonical evidence and State Integrity Reports into legible operator +cards for the SocioSphere dashboard. Implements issue sourceos-syncd#9. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +def _now() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +CARD_KIND_MAP = { + "state-integrity-violation": "operator-alert", + "policy-violation": "operator-alert", + "repair-required": "operator-action", + "conflict-detected": "operator-review", + "repair-completed": "operator-info", + "sync-ok": "operator-status", + "identity-confirmed": "operator-status", + "org-policy-loaded": "operator-status", +} + + +@dataclass +class OperatorCard: + """A legible operator-facing card for the SocioSphere dashboard.""" + + card_id: str + kind: str + severity: str + title: str + summary: str + subject_ref: str + evidence_ref: str | None + policy_decision_ref: str | None + action_required: bool + action_label: str | None + generated_at: str + source_event_ref: str | None = None + note: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "cardId": self.card_id, + "kind": self.kind, + "severity": self.severity, + "title": self.title, + "summary": self.summary, + "subjectRef": self.subject_ref, + "evidenceRef": self.evidence_ref, + "policyDecisionRef": self.policy_decision_ref, + "actionRequired": self.action_required, + "actionLabel": self.action_label, + "generatedAt": self.generated_at, + "sourceEventRef": self.source_event_ref, + "note": self.note, + } + + +@dataclass +class NarrativeSummary: + """Operator-facing narrative summary of the current state.""" + + device_ref: str + profile_ref: str | None + workspace_ref: str | None + overall_status: str + cards: list[OperatorCard] = field(default_factory=list) + generated_at: str = field(default_factory=_now) + note: str = "Stub — live sourceos-syncd daemon required for full narrative per sourceos-syncd#9." + + def to_dict(self) -> dict[str, Any]: + return { + "deviceRef": self.device_ref, + "profileRef": self.profile_ref, + "workspaceRef": self.workspace_ref, + "overallStatus": self.overall_status, + "cards": [c.to_dict() for c in self.cards], + "generatedAt": self.generated_at, + "note": self.note, + } + + +def _card_id(prefix: str) -> str: + ts = _now().replace(":", "").replace("-", "").replace("T", "").replace("Z", "") + return f"urn:srcos:operator-card:{prefix}:{ts}" + + +def card_from_event(event: dict[str, Any]) -> OperatorCard | None: + """Convert a state integrity event to an operator card. Returns None if not surfaceable.""" + kind = event.get("kind", "") + card_kind = CARD_KIND_MAP.get(kind) + if card_kind is None: + return None + + severity = "critical" if kind in ("state-integrity-violation", "policy-violation") else "warning" if kind in ("repair-required", "conflict-detected") else "info" + action_required = kind in ("repair-required", "conflict-detected", "policy-violation") + + return OperatorCard( + card_id=_card_id(kind), + kind=card_kind, + severity=severity, + title=kind.replace("-", " ").title(), + summary=event.get("body", event.get("summary", kind)), + subject_ref=event.get("subjectRef", event.get("subject_ref", "unknown")), + evidence_ref=event.get("evidenceRef", event.get("evidence_ref")), + policy_decision_ref=event.get("policyDecisionRef", event.get("policy_decision_ref")), + action_required=action_required, + action_label="Review and approve repair" if action_required else None, + generated_at=_now(), + source_event_ref=event.get("id"), + ) + + +def narrative_from_report(report: dict[str, Any]) -> NarrativeSummary: + """Build a NarrativeSummary from a State Integrity Report.""" + overall = report.get("status", "unknown") + cards = [] + + for item in report.get("violations", []): + card = card_from_event({ + "kind": "state-integrity-violation", + "body": item.get("description", "State integrity violation"), + "subjectRef": item.get("subjectRef", "unknown"), + "evidenceRef": item.get("evidenceRef"), + "policyDecisionRef": item.get("policyDecisionRef"), + }) + if card: + cards.append(card) + + for item in report.get("repairs", []): + card = card_from_event({ + "kind": "repair-required", + "body": item.get("description", "Repair required"), + "subjectRef": item.get("subjectRef", "unknown"), + }) + if card: + cards.append(card) + + return NarrativeSummary( + device_ref=report.get("deviceRef", "unknown"), + profile_ref=report.get("profileRef"), + workspace_ref=report.get("workspaceRef"), + overall_status=overall, + cards=cards, + ) + + +def stub_narrative(device_ref: str = "unknown") -> NarrativeSummary: + """Return a stub narrative when the daemon is not yet running.""" + return NarrativeSummary( + device_ref=device_ref, + profile_ref=None, + workspace_ref=None, + overall_status="unavailable", + cards=[], + ) diff --git a/src/sourceos_syncd/noise_budget.py b/src/sourceos_syncd/noise_budget.py new file mode 100644 index 0000000..6331d97 --- /dev/null +++ b/src/sourceos_syncd/noise_budget.py @@ -0,0 +1,179 @@ +"""Duplicate suppression and noise-budget engine for sourceos-syncd. + +Coalesces repeated low-level events so only operator-meaningful state +changes surface as records. Implements issue sourceos-syncd#8. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +def _now() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _event_key(event: dict[str, Any]) -> str: + key_fields = { + "kind": event.get("kind", ""), + "source": event.get("source", ""), + "subject_ref": event.get("subjectRef", event.get("subject_ref", "")), + } + raw = json.dumps(key_fields, sort_keys=True).encode() + return hashlib.sha256(raw).hexdigest()[:16] + + +@dataclass +class NoiseBudgetConfig: + """Configuration for the noise-budget engine.""" + + max_duplicates_per_window: int = 10 + window_seconds: int = 300 + critical_kinds: frozenset[str] = field(default_factory=lambda: frozenset([ + "state-integrity-violation", + "policy-violation", + "repair-required", + "conflict-detected", + "identity-loss", + ])) + exempt_sources: frozenset[str] = field(default_factory=frozenset) + + +@dataclass +class CoalescedEvent: + """A coalesced event record with suppression metadata.""" + + event: dict[str, Any] + count: int + first_seen: str + last_seen: str + suppressed_count: int + key: str + + def to_dict(self) -> dict[str, Any]: + return { + "event": self.event, + "count": self.count, + "firstSeen": self.first_seen, + "lastSeen": self.last_seen, + "suppressedCount": self.suppressed_count, + "key": self.key, + } + + +@dataclass +class BudgetReport: + """Noise budget utilization report.""" + + window_seconds: int + total_events_received: int + unique_events: int + suppressed_events: int + budget_utilization: float + critical_events_passed: int + note: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "windowSeconds": self.window_seconds, + "totalEventsReceived": self.total_events_received, + "uniqueEvents": self.unique_events, + "suppressedEvents": self.suppressed_events, + "budgetUtilization": round(self.budget_utilization, 3), + "criticalEventsPassed": self.critical_events_passed, + "note": self.note, + } + + +class NoiseBudgetEngine: + """Event coalescer and noise budget enforcer.""" + + def __init__(self, config: NoiseBudgetConfig | None = None) -> None: + self._config = config or NoiseBudgetConfig() + self._seen: dict[str, CoalescedEvent] = {} + self._total_received = 0 + self._total_suppressed = 0 + self._critical_passed = 0 + + def ingest(self, event: dict[str, Any]) -> CoalescedEvent | None: + """Ingest an event. Returns the CoalescedEvent if it should surface, None if suppressed.""" + self._total_received += 1 + kind = event.get("kind", "") + source = event.get("source", "") + + if kind in self._config.critical_kinds or source in self._config.exempt_sources: + self._critical_passed += 1 + key = _event_key(event) + coalesced = CoalescedEvent( + event=event, + count=1, + first_seen=_now(), + last_seen=_now(), + suppressed_count=0, + key=key, + ) + self._seen[key] = coalesced + return coalesced + + key = _event_key(event) + if key in self._seen: + existing = self._seen[key] + if existing.count >= self._config.max_duplicates_per_window: + existing.suppressed_count += 1 + self._total_suppressed += 1 + self._seen[key] = existing + return None + existing = CoalescedEvent( + event=event, + count=existing.count + 1, + first_seen=existing.first_seen, + last_seen=_now(), + suppressed_count=existing.suppressed_count, + key=key, + ) + self._seen[key] = existing + return existing + + coalesced = CoalescedEvent( + event=event, + count=1, + first_seen=_now(), + last_seen=_now(), + suppressed_count=0, + key=key, + ) + self._seen[key] = coalesced + return coalesced + + def budget_report(self) -> BudgetReport: + utilization = self._total_suppressed / max(self._total_received, 1) + return BudgetReport( + window_seconds=self._config.window_seconds, + total_events_received=self._total_received, + unique_events=len(self._seen), + suppressed_events=self._total_suppressed, + budget_utilization=utilization, + critical_events_passed=self._critical_passed, + ) + + def flush(self) -> None: + """Reset the window state.""" + self._seen.clear() + self._total_received = 0 + self._total_suppressed = 0 + self._critical_passed = 0 + + +_default_engine = NoiseBudgetEngine() + + +def ingest(event: dict[str, Any]) -> CoalescedEvent | None: + return _default_engine.ingest(event) + + +def budget_report() -> BudgetReport: + return _default_engine.budget_report() diff --git a/src/sourceos_syncd/service.py b/src/sourceos_syncd/service.py index d5cc954..afc0c25 100644 --- a/src/sourceos_syncd/service.py +++ b/src/sourceos_syncd/service.py @@ -167,6 +167,21 @@ def do_GET(self) -> None: # noqa: N802 self.write_text(HTTPStatus.OK, metrics_text(build_state_report(store_root)), "text/plain; charset=utf-8") elif path == "/repairz": self.write_json(HTTPStatus.OK, planning_preview(store_root)) + elif path == "/noisez": + from sourceos_syncd.noise_budget import budget_report + self.write_json(HTTPStatus.OK, budget_report().to_dict()) + elif path == "/narrativez": + from sourceos_syncd.narrative import stub_narrative + report = build_state_report(store_root) + from sourceos_syncd.narrative import narrative_from_report + narr = narrative_from_report(report) + self.write_json(HTTPStatus.OK, narr.to_dict()) + elif path == "/coherencez": + from sourceos_syncd.coherence import stub_snapshot + self.write_json(HTTPStatus.OK, stub_snapshot().to_dict()) + elif path == "/authorityz": + from sourceos_syncd.authority import stub_report + self.write_json(HTTPStatus.OK, stub_report().to_dict()) else: self.write_json(HTTPStatus.NOT_FOUND, {"error": "not_found", "path": path}) except Exception as exc: # noqa: BLE001 diff --git a/src/sourceos_syncd/systema.py b/src/sourceos_syncd/systema.py new file mode 100644 index 0000000..ab81b85 --- /dev/null +++ b/src/sourceos_syncd/systema.py @@ -0,0 +1,190 @@ +"""Systema integration for sourceos-syncd. + +Implements sourceos-syncd's Systema role: local-first state membranes, +source-confidence event mapping, and projection-aware operator narratives. + +Maps ProCybernetica's source-confidence, projection-loss, membrane-boundary, +and capability-radius profiles to existing SourceOS schemas without introducing +new schemas (uses ProvenanceRecord, ValidatorReceipt, ReasoningAssay, +StaleStateRecord from sourceos-spec). Implements issue sourceos-syncd#20. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + + +def _now() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +# Systema membrane kinds +MEMBRANE_KINDS = frozenset([ + "activation", + "side-effect", + "storage", +]) + +# Systema membrane decisions +MEMBRANE_DECISIONS = frozenset([ + "admitted", + "blocked", + "revoked", + "applied", + "suppressed", + "transformed", + "mounted", + "refused", +]) + + +@dataclass(frozen=True) +class SourceConfidenceMapping: + """Maps Systema source-confidence score to SourceOS staleness/exactness contracts.""" + + confidence: float + + @property + def sourceos_class(self) -> str: + if self.confidence >= 0.85: + return "verifier-required" + if self.confidence >= 0.70: + return "user-confirmed" + if self.confidence >= 0.50: + return "ambient-host" + return "stale-context-injection" + + @property + def admission_policy(self) -> str: + if self.confidence >= 0.85: + return "verifier-required" + if self.confidence >= 0.70: + return "user-confirmed" + return "cryptographic-proof" + + @property + def staleness_risk(self) -> str: + if self.confidence >= 0.85: + return "none" + if self.confidence >= 0.70: + return "low" + if self.confidence >= 0.50: + return "medium" + return "high" + + @property + def suppress_mutation(self) -> bool: + return self.confidence < 0.50 + + def to_dict(self) -> dict[str, Any]: + return { + "confidence": self.confidence, + "sourceosClass": self.sourceos_class, + "admissionPolicy": self.admission_policy, + "stalenessRisk": self.staleness_risk, + "suppressMutation": self.suppress_mutation, + } + + +@dataclass(frozen=True) +class MembraneEvent: + """A Systema membrane boundary event mapped to SourceOS contracts.""" + + membrane_kind: str + decision: str + subject_ref: str + actor_ref: str + confidence: float + policy_decision_ref: str | None + observed_at: str = field(default_factory=_now) + + @property + def confidence_mapping(self) -> SourceConfidenceMapping: + return SourceConfidenceMapping(self.confidence) + + def to_dict(self) -> dict[str, Any]: + return { + "membraneKind": self.membrane_kind, + "decision": self.decision, + "subjectRef": self.subject_ref, + "actorRef": self.actor_ref, + "confidence": self.confidence, + "confidenceMapping": self.confidence_mapping.to_dict(), + "policyDecisionRef": self.policy_decision_ref, + "observedAt": self.observed_at, + } + + +# Capability radius levels (R0–R5) +CAPABILITY_RADIUS_DESCRIPTIONS = { + 0: "observe-only — no side effects, no network, no file writes", + 1: "local-read — file-read from allowed paths, no writes", + 2: "local-write — file-read + controlled file-write to workspace", + 3: "local-execute — process-spawn within policy constraints", + 4: "network-local — TCP to localhost/LAN, no internet", + 5: "internet-connected — full network egress, policy-gated", +} + + +@dataclass(frozen=True) +class CapabilityRadiusProfile: + """Maps a Systema capability radius to Agent Machine contracts.""" + + radius: int + agent_ref: str + + @property + def description(self) -> str: + return CAPABILITY_RADIUS_DESCRIPTIONS.get(self.radius, "unknown") + + @property + def capability_contract_kind(self) -> str: + map_ = {0: "observe-only", 1: "read-local", 2: "write-local", 3: "execute-local", 4: "network-local", 5: "network-internet"} + return map_.get(self.radius, "unknown") + + @property + def fail_closed(self) -> bool: + return True + + def to_dict(self) -> dict[str, Any]: + return { + "radius": self.radius, + "agentRef": self.agent_ref, + "description": self.description, + "capabilityContractKind": self.capability_contract_kind, + "failClosed": self.fail_closed, + } + + +def map_confidence(confidence: float) -> SourceConfidenceMapping: + return SourceConfidenceMapping(confidence) + + +def map_membrane_event( + membrane_kind: str, + decision: str, + subject_ref: str, + actor_ref: str, + confidence: float, + policy_decision_ref: str | None = None, +) -> MembraneEvent: + if membrane_kind not in MEMBRANE_KINDS: + raise ValueError(f"Unknown membrane kind: {membrane_kind}") + if decision not in MEMBRANE_DECISIONS: + raise ValueError(f"Unknown membrane decision: {decision}") + return MembraneEvent( + membrane_kind=membrane_kind, + decision=decision, + subject_ref=subject_ref, + actor_ref=actor_ref, + confidence=confidence, + policy_decision_ref=policy_decision_ref, + ) + + +def capability_radius(radius: int, agent_ref: str) -> CapabilityRadiusProfile: + if not (0 <= radius <= 5): + raise ValueError(f"Capability radius must be 0–5, got {radius}") + return CapabilityRadiusProfile(radius=radius, agent_ref=agent_ref)