From 5f3c8f92f924bb957f790d40bb89c19842765889 Mon Sep 17 00:00:00 2001 From: David Hyrule Date: Sat, 11 Jul 2026 19:29:03 +0200 Subject: [PATCH 1/4] integrate Knowledge Loop with LHP v2 --- README.md | 26 ++ knowledge-policy.yml | 30 ++ knowledge.config.yml | 12 + ledger/README.md | 5 + pyproject.toml | 7 +- schema/learning-event.schema.json | 4 +- src/hyrule_knowledge/context_pack.py | 25 +- src/hyrule_knowledge/coordination.py | 415 ++++++++++++++++++++++++ src/hyrule_knowledge/learning_ledger.py | 10 +- src/hyrule_knowledge/policy.py | 36 +- tests/test_coordination_v2.py | 162 +++++++++ uv.lock | 28 +- 12 files changed, 746 insertions(+), 14 deletions(-) create mode 100644 src/hyrule_knowledge/coordination.py create mode 100644 tests/test_coordination_v2.py diff --git a/README.md b/README.md index f9eda7e..b7f7823 100644 --- a/README.md +++ b/README.md @@ -198,6 +198,32 @@ uv run hyrule-knowledge loop --once --create-pr \ Synced decisions/labels land under `ledger/insights/` and ride the reviewed nightly PR — `ledger/` is excluded from the auto-merge allowlist precisely so this stream always crosses human eyes. Env equivalents: `HYRULE_KNOWLEDGE_LOOP_INSIGHT_SYNC`, `HYRULE_KNOWLEDGE_LOOP_INSIGHT_COLLECTOR_URL`, `HYRULE_KNOWLEDGE_LOOP_MAX_INSIGHT_RECORDS_PER_DAY`. +### Agentic coordinator intake + +`hyrule-knowledge-coordinator` is the Knowledge adapter for Loop Handoff +Protocol v2. It serves policy-gated context to SOC, NOC, and Engineering with a +role derived from the authenticated source loop, so a caller cannot request a +more privileged Knowledge role. It also accepts approved, cited learning +proposals from those loops. Coordinator intake can only stage sanitized A4 +events; it cannot write A1/A2 knowledge or bypass the Knowledge Loop PR. + +```bash +HYRULE_COORDINATOR_URL=http://127.0.0.1:8771 \ +HYRULE_COORDINATOR_SECRET=... \ +uv run hyrule-knowledge-coordinator --once \ + --proposal-dir /var/lib/hyrule-knowledge/coordinator-intake + +uv run hyrule-knowledge loop --once --create-pr \ + --learning-event /var/lib/hyrule-knowledge/coordinator-intake +``` + +SOC context uses the dedicated `soc_shadow` policy role. It permits cited +intended-state and sanitized observation lookup while denying production +actions, live diagnostics, secrets, raw logs, and packet captures. Learning +intake independently validates the coordinator-bound scope, producer/event +type, citations, forbidden data classes/keys, A4 authority, and retained human +review gate before writing the spool file. + ## Governed agent consumption Humans can browse `okf/index.md`. Agents should first read `okf/index.md`, then directory indexes, then individual concepts. Machine consumers should prefer `exports/knowledge.sqlite` and the JSONL exports: diff --git a/knowledge-policy.yml b/knowledge-policy.yml index cd0b307..186e296 100644 --- a/knowledge-policy.yml +++ b/knowledge-policy.yml @@ -69,6 +69,30 @@ roles: - production.* - live_diagnostic.* - secret.* + soc_shadow: + allow_actions: + - knowledge.resolve + - knowledge.search + - knowledge.claims + - knowledge.neighborhood + - knowledge.firewall_rules + - knowledge.server_registry + - knowledge.flows_into + - knowledge.flows_from + - knowledge.context_pack.generate + - knowledge.intended_state.read + - knowledge.observed_state.read + - knowledge.diff.read + - knowledge.intended_state + - knowledge.observed_state.fixture + - knowledge.diff_intended_observed + - knowledge.related_runbooks + - knowledge.find_conflicts + - knowledge.find_stale + deny_actions: + - production.* + - live_diagnostic.* + - secret.* general: allow_actions: - knowledge.resolve @@ -100,3 +124,9 @@ rules: actors_any: [noc_shadow] action_prefixes_any: [production., live_diagnostic., secret.] reason: NOC shadow evals may not invoke live operational tools. + - id: soc_shadow_no_live_tools + effect: deny + when: + actors_any: [soc_shadow] + action_prefixes_any: [production., live_diagnostic., secret.] + reason: SOC context retrieval may not invoke live operational tools or disclose protected data. diff --git a/knowledge.config.yml b/knowledge.config.yml index c7a1bc9..7ff0098 100644 --- a/knowledge.config.yml +++ b/knowledge.config.yml @@ -45,6 +45,18 @@ sources: project_type: Service title: Hyrule Engineering Loop tags: [hyrule, engineering-loop, automation, langgraph] + - repo: AS215932/soc-agent + project_type: Service + title: Hyrule SOC Agent + tags: [hyrule, soc, security, posture, incident-response] + - repo: AS215932/agent-core + project_type: Service + title: Hyrule Agent Core + tags: [hyrule, agents, contracts, coordination, observability] + - repo: AS215932/agentic-observatory + project_type: Service + title: Hyrule Agentic Observatory + tags: [hyrule, agents, observability, approvals, control-plane] - repo: AS215932/as215932.net project_type: Service title: as215932.net diff --git a/ledger/README.md b/ledger/README.md index 6af2593..2adeb3c 100644 --- a/ledger/README.md +++ b/ledger/README.md @@ -21,3 +21,8 @@ Events are A4 proposals/fixtures unless promoted by human review into A1 curated lessons or A2 reviewed summaries. Promotion writes an audit record under `ledger/reviews/` and an OKF concept under `okf/curated/lessons/` or `okf/curated/summaries/`. + +The LHP-v2 Knowledge coordinator adapter follows the same boundary. It stages +validated files in its configured coordinator-intake directory; the Knowledge +Loop imports those files and opens the usual review PR. A coordinator approval +authorizes staging only and is never authority to promote a claim or lesson. diff --git a/pyproject.toml b/pyproject.toml index 53acf3a..fa8c378 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ description = "OKF knowledge catalog, enrichment, quality, and observation tooli requires-python = ">=3.12" dependencies = [ "PyYAML>=6.0.2", - "agent-core", + "agent-core[coordination-client]", ] [project.optional-dependencies] @@ -20,6 +20,7 @@ api = ["starlette>=0.37", "uvicorn>=0.30"] hyrule-knowledge = "hyrule_knowledge.cli:main" hyrule-knowledge-mcp = "hyrule_knowledge.mcp_server:main" hyrule-knowledge-api = "hyrule_knowledge.api:main" +hyrule-knowledge-coordinator = "hyrule_knowledge.coordination:main" [tool.hatch.build.targets.wheel] packages = ["src/hyrule_knowledge"] @@ -29,6 +30,7 @@ dev = [ "mcp>=1.27.0", "mypy>=1.11", "pytest>=8", + "pytest-asyncio>=0.23", "ruff>=0.6", "types-PyYAML>=6.0.12", ] @@ -51,6 +53,7 @@ ignore_missing_imports = true [tool.pytest.ini_options] testpaths = ["tests"] +asyncio_mode = "auto" [tool.uv.sources] -agent-core = { git = "https://github.com/AS215932/agent-core", tag = "v0.8.0" } +agent-core = { git = "https://github.com/AS215932/agent-core", rev = "f7916bb6375ae898cb08bfbdbea5d8d55ba463e7" } diff --git a/schema/learning-event.schema.json b/schema/learning-event.schema.json index c541bca..ee34b08 100644 --- a/schema/learning-event.schema.json +++ b/schema/learning-event.schema.json @@ -6,9 +6,9 @@ "properties": { "id": {"type": "string", "pattern": "^learn_[a-f0-9]{32}$"}, "ledger_version": {"type": "string", "const": "learning_ledger_v1"}, - "event_type": {"type": "string", "enum": ["context_pack_used", "engineering_loop_run_summary", "noc_shadow_eval_summary", "eval_run_summary", "policy_review_summary", "lesson_candidate"]}, + "event_type": {"type": "string", "enum": ["context_pack_used", "engineering_loop_run_summary", "noc_shadow_eval_summary", "soc_case_outcome", "eval_run_summary", "policy_review_summary", "lesson_candidate"]}, "event_time": {"type": "string"}, - "producer": {"type": "string", "enum": ["knowledge", "engineering_loop", "noc_shadow", "human_review", "fixture"]}, + "producer": {"type": "string", "enum": ["knowledge", "engineering_loop", "noc_shadow", "soc_shadow", "human_review", "fixture"]}, "subject": {"type": "string"}, "summary": {"type": "string", "minLength": 1}, "status": {"type": "string", "enum": ["fixture", "proposed", "reviewed", "rejected", "superseded"]}, diff --git a/src/hyrule_knowledge/context_pack.py b/src/hyrule_knowledge/context_pack.py index edbf9ed..8fe4600 100644 --- a/src/hyrule_knowledge/context_pack.py +++ b/src/hyrule_knowledge/context_pack.py @@ -118,6 +118,8 @@ def build_context_pack( if role == "noc_shadow": sections = _noc_shadow_sections(task, included_refs, claims, conflicts, stale, unresolved) + elif role == "soc_shadow": + sections = _soc_shadow_sections(task, included_refs, claims, conflicts, stale, unresolved) else: sections = _engineering_sections(task, included_refs, claims, conflicts, stale, unresolved) sections = _truncate_sections(sections, max_chars=max_chars) @@ -235,7 +237,6 @@ def _relevant_enrichment_ids(task: str, *, parsed: ParsedTask, store: KnowledgeS out.append(concept_id) return out - def _explicit_enrichment_ids(task: str, *, parsed: ParsedTask) -> list[str]: ids = [concept_id for concept_id in parsed.concept_ids if concept_id.startswith("generated/enriched/")] for token in task.replace("`", " ").split(): @@ -626,6 +627,28 @@ def _noc_shadow_sections(task: str, included_refs: list[dict[str, Any]], claims: ] +def _soc_shadow_sections( + task: str, + included_refs: list[dict[str, Any]], + claims: list[dict[str, Any]], + conflicts: list[dict[str, Any]], + stale: list[dict[str, Any]], + unresolved: list[str], +) -> list[ContextPackSection]: + intended = [claim for claim in claims if claim.get("authority_tier") in {"A0", "A1"}] + observed = [claim for claim in claims if claim.get("authority_tier") == "A3"] + return [ + ContextPackSection("task_summary", f"SOC governed-context task: {task}", []), + ContextPackSection("security_intended_state", _format_claims(intended, set()), _claim_refs(intended)), + ContextPackSection("sanitized_observed_state", _format_claims(observed, set()) or "No sanitized observed-state claims included.", _claim_refs(observed)), + ContextPackSection("drift_conflicts_and_unknowns", _format_conflicts_stale(conflicts, stale) + "\n" + _format_list(unresolved, "No unresolved security-context questions detected."), _claim_refs(stale)), + ContextPackSection("related_security_runbooks", _format_refs([ref for ref in included_refs if ref.get("type") in {"Runbook", "Policy", "Architecture", "Lesson"}], None), _ref_ids(included_refs, None)), + ContextPackSection("safe_security_boundaries", "Use sanitized, cited context only. No raw logs, packet captures, credentials, broad scans, production mutation, or unapproved live diagnostics.", []), + ContextPackSection("forbidden_actions", "Knowledge retrieval does not authorize remediation or active probing. Any RT-2 probe requires an immutable senior approval bound to its exact scope.", []), + ContextPackSection("unresolved_questions", _format_list(unresolved, "No unresolved questions detected."), []), + ] + + def _advisory_enrichment_refs(refs: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ref for ref in refs if str(ref.get("concept_id") or "").startswith("generated/enriched/")] diff --git a/src/hyrule_knowledge/coordination.py b/src/hyrule_knowledge/coordination.py new file mode 100644 index 0000000..bafaeb2 --- /dev/null +++ b/src/hyrule_knowledge/coordination.py @@ -0,0 +1,415 @@ +"""Knowledge Loop adapter for the shared LHP-v2 coordinator. + +The adapter serves governed context packs and stages sanitized A4 learning +proposals. It never promotes a proposal into accepted knowledge directly; +the existing Knowledge Loop imports the spool and opens its normal reviewed PR. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +from pathlib import Path +from typing import Any +from uuid import uuid4 + +from agent_core.contracts import HandoffRecord, HandoffResult, LoopHeartbeat, SourceRef +from agent_core.coordination import CoordinatorClient, CoordinatorError + +from .context_pack import build_context_pack +from .learning_ledger import ( + LearningLedgerError, + build_local_learning_event, + validate_learning_event, +) +from .store import KnowledgeStore + +LOG = logging.getLogger(__name__) + +_CONTEXT_ROLES = { + "soc": "soc_shadow", + "noc": "noc_shadow", + "engineering": "engineering_loop", +} +_LEARNING_PROFILES = { + "soc": ("soc_shadow", {"soc_case_outcome"}), + "noc": ("noc_shadow", {"noc_shadow_eval_summary", "lesson_candidate"}), + "engineering": ( + "engineering_loop", + {"engineering_loop_run_summary", "lesson_candidate"}, + ), +} +_SUPPORTED_CAPABILITIES = frozenset( + { + "knowledge.context.resolve", + "knowledge.gap.analyze", + "knowledge.learning.proposal", + } +) + + +def _text(value: Any, *, limit: int) -> str: + return " ".join(str(value or "").split())[:limit] + + +def _bounded(value: Any, *, depth: int = 5) -> Any: + if depth <= 0: + return _text(value, limit=1000) + if isinstance(value, dict): + return { + _text(key, limit=100): _bounded(item, depth=depth - 1) + for key, item in list(value.items())[:100] + } + if isinstance(value, (list, tuple)): + return [_bounded(item, depth=depth - 1) for item in list(value)[:100]] + if isinstance(value, str): + return value[:4000] + if isinstance(value, (bool, int, float)) or value is None: + return value + return _text(value, limit=1000) + + +def _citations(record: HandoffRecord) -> list[dict[str, str]]: + citations: list[dict[str, str]] = [] + for ref in [*record.envelope.context_refs, *record.envelope.evidence_refs]: + value = _text(ref.ref, limit=500) + if not value: + continue + if value.startswith("ctx_"): + citation = {"context_pack_id": value} + else: + citation = {"source_uri": value} + if citation not in citations: + citations.append(citation) + return citations[:40] + + +def _pack_citations(included_refs: list[dict[str, Any]]) -> list[dict[str, str]]: + citations: list[dict[str, str]] = [] + for ref in included_refs[:40]: + concept_id = _text(ref.get("concept_id"), limit=300) + if not concept_id: + continue + citation = {"concept_id": concept_id} + source_refs = ref.get("source_refs") + if isinstance(source_refs, list) and source_refs and isinstance(source_refs[0], dict): + source = source_refs[0] + uri = _text(source.get("url"), limit=500) + if not uri: + repo = _text(source.get("repo"), limit=200) + path = _text(source.get("path"), limit=300) + uri = f"{repo}:{path}" if repo and path else "" + if uri: + citation["source_uri"] = uri + citations.append(citation) + return citations + + +def _pack_evidence(included_refs: list[dict[str, Any]]) -> list[SourceRef]: + evidence: list[SourceRef] = [] + for ref in included_refs[:40]: + concept_id = _text(ref.get("concept_id"), limit=300) + if not concept_id: + continue + authority = str(ref.get("authority_tier") or "A4") + evidence.append( + SourceRef( + ref=concept_id, + kind="knowledge_concept", + authority=authority, # type: ignore[arg-type] + ) + ) + return evidence + + +class KnowledgeCoordinatorWorker: + """Consumes coordinator work without bypassing Knowledge policy or review.""" + + def __init__( + self, + client: CoordinatorClient, + *, + store_path: Path = Path("exports/knowledge.sqlite"), + policy_path: Path = Path("knowledge-policy.yml"), + proposal_dir: Path = Path(".cache/hyrule-knowledge/coordinator-intake"), + ) -> None: + self.client = client + self.store_path = store_path + self.policy_path = policy_path + self.proposal_dir = proposal_dir + + async def run_once(self) -> dict[str, int]: + await self.client.heartbeat( + LoopHeartbeat( + loop_id="knowledge", + status="active", + summary="Knowledge coordinator intake active", + metadata={"direct_promotion": False, "learning_authority": "A4"}, + ) + ) + report = {"processed": 0, "failed": 0, "ignored": 0} + for record in await self.client.inbox(status="queued"): + if record.envelope.capability not in _SUPPORTED_CAPABILITIES: + report["ignored"] += 1 + continue + try: + claimed = await self.client.claim(record.envelope.handoff_id, lease_seconds=300) + await self.client.progress( + record.envelope.handoff_id, + "Knowledge accepted bounded coordinator work", + ) + # Context assembly is bounded and uses a short-lived read-only + # SQLite connection. Keep it in-cycle so claim/result ordering + # remains deterministic across service runtimes. + result = self._handle(claimed) + await self.client.submit_result(result) + report["processed"] += 1 + except (CoordinatorError, LearningLedgerError, OSError, ValueError) as exc: + report["failed"] += 1 + LOG.warning( + "knowledge coordinator work failed: handoff=%s error=%s", + record.envelope.handoff_id, + exc, + ) + try: + await self.client.submit_result( + HandoffResult( + handoff_id=record.envelope.handoff_id, + outcome="failed", + summary=f"Knowledge policy or validation rejected the request: {exc}", + payload={"direct_promotion": False}, + ) + ) + except CoordinatorError: + LOG.exception( + "could not publish failed Knowledge result for %s", + record.envelope.handoff_id, + ) + return report + + def _handle(self, record: HandoffRecord) -> HandoffResult: + if record.envelope.capability == "knowledge.learning.proposal": + return self._stage_learning_proposal(record) + return self._resolve_context(record) + + def _resolve_context(self, record: HandoffRecord) -> HandoffResult: + envelope = record.envelope + role = _CONTEXT_ROLES.get(envelope.source_loop) + if role is None: + raise ValueError( + f"source loop may not request Knowledge context: {envelope.source_loop}" + ) + task = _text( + envelope.payload.get("query") + or envelope.payload.get("task") + or envelope.intent + or envelope.summary, + limit=4000, + ) + if not task: + raise ValueError("Knowledge context request has no task or query") + with KnowledgeStore(self.store_path) as store: + pack = build_context_pack( + task=task, + role=role, + store=store, + risk_level=envelope.risk_level, + token_budget=6000, + max_chars=20_000, + task_id=envelope.work_item_id or envelope.handoff_id, + policy_path=self.policy_path, + ) + policy_result = str(pack.policy_decision.get("result") or "deny") + if policy_result not in {"allow", "allow_readonly_substitute"}: + return HandoffResult( + handoff_id=envelope.handoff_id, + outcome="rejected", + summary="Knowledge policy denied the requested context pack", + payload={ + "context_pack_id": pack.id, + "policy_decision_id": pack.policy_decision.get("id"), + "policy_result": policy_result, + }, + ) + sections = [ + { + "name": section.name, + "body": section.body[:4000], + "refs": section.refs[:40], + } + for section in pack.sections + ] + citations = _pack_citations(pack.included_refs) + payload = { + "context_pack_id": pack.id, + "policy_decision_id": pack.policy_decision.get("id"), + "policy_result": policy_result, + "role": role, + "knowledge_snapshot": pack.knowledge_snapshot, + "sections": sections, + "citations": citations, + "unresolved_questions": pack.unresolved_questions[:20], + "gap_detected": bool(pack.unresolved_questions), + "read_only": True, + } + return HandoffResult( + handoff_id=envelope.handoff_id, + outcome="succeeded", + summary=( + "Governed Knowledge gap analysis completed" + if envelope.capability == "knowledge.gap.analyze" + else "Governed Knowledge context pack resolved" + ), + evidence_refs=_pack_evidence(pack.included_refs), + payload=_bounded(payload), + ) + + def _stage_learning_proposal(self, record: HandoffRecord) -> HandoffResult: + envelope = record.envelope + approval = record.approval + if approval is None or approval.decision != "approved": + raise ValueError("learning proposal has no immutable coordinator approval") + if approval.scope_hash != envelope.scope_hash: + raise ValueError("learning proposal approval is stale") + if not envelope.constraints.get("human_review_required"): + raise ValueError("learning proposal must require human review") + if envelope.constraints.get("direct_a1_a2_write") is not False: + raise ValueError("learning proposal must explicitly forbid direct A1/A2 writes") + profile = _LEARNING_PROFILES.get(envelope.source_loop) + if profile is None: + raise ValueError(f"source loop may not contribute learning: {envelope.source_loop}") + producer, allowed_event_types = profile + event_type = _text(envelope.payload.get("event_type"), limit=100) + if event_type not in allowed_event_types: + raise ValueError( + f"{envelope.source_loop} may not propose learning event type {event_type!r}" + ) + citations = _citations(record) + if not citations: + raise ValueError("learning proposal requires at least one cited source or context pack") + context_pack_ids = [ + citation["context_pack_id"] for citation in citations if "context_pack_id" in citation + ] + raw_lessons = envelope.payload.get("lessons") + lessons = ( + [_text(item, limit=1000) for item in raw_lessons[:20]] + if isinstance(raw_lessons, list) + else [] + ) + raw_metrics = envelope.payload.get("metrics") + metrics = _bounded(raw_metrics) if isinstance(raw_metrics, dict) else {} + event = build_local_learning_event( + producer=producer, + event_type=event_type, + subject=_text( + envelope.payload.get("subject") or envelope.case_id or envelope.work_item_id, + limit=500, + ), + summary=_text( + envelope.payload.get("summary") or envelope.summary or envelope.intent, + limit=4000, + ), + citations=citations, + status="proposed", + authority_tier="A4", + context_pack_ids=context_pack_ids, + metrics=metrics, + lessons=lessons, + metadata={ + "coordination_handoff_id": envelope.handoff_id, + "coordination_scope_hash": envelope.scope_hash, + "source_loop": envelope.source_loop, + "adversarial_validation": "passed", + "direct_promotion": False, + }, + ) + findings = validate_learning_event(event) + if findings: + raise LearningLedgerError("adversarial validation failed: " + "; ".join(findings)) + if event["authority_tier"] != "A4" or event["status"] != "proposed": + raise LearningLedgerError("coordinator intake may only stage A4 proposals") + if event.get("promotion", {}).get("review_required") is not True: + raise LearningLedgerError("learning proposal must retain Knowledge PR review") + target = self._write_proposal(event) + return HandoffResult( + handoff_id=envelope.handoff_id, + outcome="succeeded", + summary="Sanitized A4 proposal validated and staged for Knowledge Loop PR review", + artifact_refs=[ + SourceRef( + ref=f"knowledge-learning-proposal:{event['id']}", + kind="learning_proposal", + authority="A4", + review_status="validated_pending_pr", + ) + ], + payload={ + "proposal_id": event["id"], + "review_status": "validated_pending_pr", + "authority_tier": "A4", + "adversarial_validation": "passed", + "direct_promotion": False, + "spool_filename": target.name, + }, + ) + + def _write_proposal(self, event: dict[str, Any]) -> Path: + self.proposal_dir.mkdir(parents=True, exist_ok=True, mode=0o750) + target = self.proposal_dir / f"{event['id']}.json" + encoded = json.dumps(event, sort_keys=True, indent=2) + "\n" + if target.exists(): + if target.read_text(encoding="utf-8") != encoded: + raise LearningLedgerError(f"proposal id collision: {event['id']}") + return target + temporary = target.with_name(f".{target.name}.{uuid4().hex}.tmp") + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, target) + finally: + if temporary.exists(): + temporary.unlink() + return target + + +async def _run(args: argparse.Namespace) -> None: + worker = KnowledgeCoordinatorWorker( + CoordinatorClient.from_env("knowledge"), + store_path=args.store, + policy_path=args.policy, + proposal_dir=args.proposal_dir, + ) + if args.once: + print(json.dumps(await worker.run_once(), sort_keys=True)) + return + while True: + try: + await worker.run_once() + except CoordinatorError: + LOG.exception("Knowledge coordinator cycle failed") + await asyncio.sleep(max(1.0, args.interval)) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run the Knowledge LHP-v2 coordinator adapter") + parser.add_argument("--once", action="store_true") + parser.add_argument("--interval", type=float, default=10.0) + parser.add_argument("--store", type=Path, default=Path("exports/knowledge.sqlite")) + parser.add_argument("--policy", type=Path, default=Path("knowledge-policy.yml")) + parser.add_argument( + "--proposal-dir", + type=Path, + default=Path(".cache/hyrule-knowledge/coordinator-intake"), + ) + logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO")) + asyncio.run(_run(parser.parse_args())) + + +if __name__ == "__main__": + main() diff --git a/src/hyrule_knowledge/learning_ledger.py b/src/hyrule_knowledge/learning_ledger.py index c11f8d9..d7d9711 100644 --- a/src/hyrule_knowledge/learning_ledger.py +++ b/src/hyrule_knowledge/learning_ledger.py @@ -27,12 +27,20 @@ "context_pack_used", "engineering_loop_run_summary", "noc_shadow_eval_summary", + "soc_case_outcome", "eval_run_summary", "policy_review_summary", "lesson_candidate", } STATUSES = {"fixture", "proposed", "reviewed", "rejected", "superseded"} -PRODUCERS = {"knowledge", "engineering_loop", "noc_shadow", "human_review", "fixture"} +PRODUCERS = { + "knowledge", + "engineering_loop", + "noc_shadow", + "soc_shadow", + "human_review", + "fixture", +} FORBIDDEN_DATA_CLASSES = {"secret", "credential", "raw_log", "packet_capture", "command_output", "live_trace"} FORBIDDEN_KEY_PARTS = { "raw_log", diff --git a/src/hyrule_knowledge/policy.py b/src/hyrule_knowledge/policy.py index ab7687c..bc8b87d 100644 --- a/src/hyrule_knowledge/policy.py +++ b/src/hyrule_knowledge/policy.py @@ -90,6 +90,26 @@ def default_policy() -> dict[str, Any]: ], "deny_actions": ["production.*", "live_diagnostic.*", "secret.*"], }, + "soc_shadow": { + "allow_actions": [ + "knowledge.resolve", + "knowledge.search", + "knowledge.claims", + "knowledge.neighborhood", + "knowledge.firewall_rules", + "knowledge.server_registry", + "knowledge.flows_into", + "knowledge.flows_from", + "knowledge.context_pack.generate", + "knowledge.intended_state", + "knowledge.observed_state.fixture", + "knowledge.diff_intended_observed", + "knowledge.related_runbooks", + "knowledge.find_conflicts", + "knowledge.find_stale", + ], + "deny_actions": ["production.*", "live_diagnostic.*", "secret.*"], + }, "general": { "allow_actions": [ "knowledge.resolve", @@ -135,7 +155,9 @@ def evaluate_policy(request: PolicyRequest, policy: dict[str, Any] | None = None roles = _mapping(active.get("roles"), label="roles") role = _mapping(roles.get(request.actor), label=f"roles.{request.actor}") reasons: list[str] = [] - forbid_data_classes = _list(defaults.get("deny_data_classes"), label="defaults.deny_data_classes") + forbid_data_classes = _list( + defaults.get("deny_data_classes"), label="defaults.deny_data_classes" + ) constraints = { "max_context_pack_tokens": int(defaults.get("max_context_pack_tokens", 8000)), "max_result_refs": int(defaults.get("max_result_refs", 40)), @@ -163,9 +185,13 @@ def evaluate_policy(request: PolicyRequest, policy: dict[str, Any] | None = None constraints=constraints, ) - require_human = _list(role.get("require_human_actions"), label=f"roles.{request.actor}.require_human_actions") + require_human = _list( + role.get("require_human_actions"), label=f"roles.{request.actor}.require_human_actions" + ) production_default = str(defaults.get("production_mutation_result", "require_human")) - if _matches(request.action, require_human) or _matches(request.action, sorted(PRODUCTION_MUTATION_ACTIONS)): + if _matches(request.action, require_human) or _matches( + request.action, sorted(PRODUCTION_MUTATION_ACTIONS) + ): result = "require_human" if production_default == "require_human" else "deny" return PolicyDecision.build( request=request, @@ -197,7 +223,9 @@ def evaluate_policy(request: PolicyRequest, policy: dict[str, Any] | None = None fallback = str(defaults.get("result", "deny")) return PolicyDecision.build( request=request, - result="deny" if fallback not in {"allow", "require_human", "allow_readonly_substitute"} else fallback, # type: ignore[arg-type] + result="deny" + if fallback not in {"allow", "require_human", "allow_readonly_substitute"} + else fallback, # type: ignore[arg-type] policy_version=version, reasons=[f"no allow rule matched for actor={request.actor} action={request.action}"], constraints=constraints, diff --git a/tests/test_coordination_v2.py b/tests/test_coordination_v2.py new file mode 100644 index 0000000..6253277 --- /dev/null +++ b/tests/test_coordination_v2.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from agent_core.contracts import ( + ApprovalRecord, + HandoffEnvelope, + HandoffRecord, + HandoffResult, + SourceRef, +) + +from hyrule_knowledge.coordination import KnowledgeCoordinatorWorker +from hyrule_knowledge.learning_ledger import LearningLedgerError + +ROOT = Path(__file__).resolve().parents[1] + + +def _context_record() -> HandoffRecord: + envelope = HandoffEnvelope( + source_loop="soc", + target_loop="knowledge", + capability="knowledge.context.resolve", + work_item_id="finding-1", + intent="Resolve security context", + payload={ + "query": "SOC Agent posture, network operations, and security policy", + "role": "engineering_loop", + }, + idempotency_key="knowledge-context-test", + ) + return HandoffRecord(envelope=envelope, status="claimed", claim_owner="knowledge") + + +def _learning_record(*, unsafe: bool = False) -> HandoffRecord: + metrics = {"stdout": "raw command data"} if unsafe else {"consecutive_passes": 3} + envelope = HandoffEnvelope( + source_loop="soc", + target_loop="knowledge", + capability="knowledge.learning.proposal", + case_id="soc-case-1", + intent="Stage a reviewed security lesson", + summary="Verified intended-state drift was resolved", + risk_level="medium", + approval_tier="operator", + payload={ + "event_type": "soc_case_outcome", + "producer": "soc_shadow", + "subject": "service:soc-agent", + "summary": "Three cited verification cycles confirmed the drift was resolved.", + "metrics": metrics, + }, + context_refs=[SourceRef(ref="soc-case-1", kind="soc_case", authority="A4")], + constraints={ + "no_raw_logs": True, + "no_secrets": True, + "human_review_required": True, + "direct_a1_a2_write": False, + }, + idempotency_key=f"knowledge-learning-test-{unsafe}", + ) + approval = ApprovalRecord( + handoff_id=envelope.handoff_id, + scope_hash=envelope.scope_hash, + decision="approved", + approver_id="observatory:user-1", + approver_login="operator", + approver_role="operator", + rationale="Reviewed bounded sanitized proposal", + ) + return HandoffRecord( + envelope=envelope, + status="claimed", + claim_owner="knowledge", + approval=approval, + ) + + +def test_soc_context_role_is_derived_from_loop_identity(tmp_path: Path) -> None: + worker = KnowledgeCoordinatorWorker( + object(), # type: ignore[arg-type] + store_path=ROOT / "exports/knowledge.sqlite", + policy_path=ROOT / "knowledge-policy.yml", + proposal_dir=tmp_path, + ) + + result = worker._resolve_context(_context_record()) + + assert result.outcome == "succeeded" + assert result.payload["role"] == "soc_shadow" + assert result.payload["policy_result"] == "allow" + assert result.payload["read_only"] is True + section_names = {section["name"] for section in result.payload["sections"]} + assert "safe_security_boundaries" in section_names + + +def test_soc_learning_is_staged_as_a4_pending_review(tmp_path: Path) -> None: + worker = KnowledgeCoordinatorWorker( + object(), # type: ignore[arg-type] + proposal_dir=tmp_path, + ) + + result = worker._stage_learning_proposal(_learning_record()) + + assert result.outcome == "succeeded" + assert result.payload["authority_tier"] == "A4" + assert result.payload["review_status"] == "validated_pending_pr" + assert result.payload["direct_promotion"] is False + event = json.loads((tmp_path / result.payload["spool_filename"]).read_text()) + assert event["producer"] == "soc_shadow" + assert event["event_type"] == "soc_case_outcome" + assert event["promotion"]["review_required"] is True + assert event["metadata"]["direct_promotion"] is False + + +def test_soc_learning_adversarial_validation_rejects_raw_output(tmp_path: Path) -> None: + worker = KnowledgeCoordinatorWorker( + object(), # type: ignore[arg-type] + proposal_dir=tmp_path, + ) + + with pytest.raises(LearningLedgerError, match="forbidden key"): + worker._stage_learning_proposal(_learning_record(unsafe=True)) + + +class _FakeClient: + def __init__(self, record: HandoffRecord) -> None: + self.record = record + self.results: list[HandoffResult] = [] + + async def heartbeat(self, _heartbeat: object) -> dict[str, object]: + return {} + + async def inbox(self, **_kwargs: object) -> list[HandoffRecord]: + return [self.record] + + async def claim(self, _handoff_id: str, **_kwargs: object) -> HandoffRecord: + return self.record + + async def progress(self, _handoff_id: str, _summary: str) -> HandoffRecord: + return self.record + + async def submit_result(self, result: HandoffResult) -> HandoffRecord: + self.results.append(result) + return self.record + + +@pytest.mark.asyncio +async def test_worker_consumes_standard_queue_and_publishes_result(tmp_path: Path) -> None: + client = _FakeClient(_learning_record()) + worker = KnowledgeCoordinatorWorker( + client, # type: ignore[arg-type] + proposal_dir=tmp_path, + ) + + report = await worker.run_once() + + assert report == {"processed": 1, "failed": 0, "ignored": 0} + assert len(client.results) == 1 + assert client.results[0].payload["direct_promotion"] is False diff --git a/uv.lock b/uv.lock index 5a74aca..fe49068 100644 --- a/uv.lock +++ b/uv.lock @@ -11,12 +11,17 @@ resolution-markers = [ [[package]] name = "agent-core" -version = "0.7.0" -source = { git = "https://github.com/AS215932/agent-core?tag=v0.8.0#4329ec96d689d0dc347b3e420f6c4ac4dcdec945" } +version = "0.9.0" +source = { git = "https://github.com/AS215932/agent-core?rev=f7916bb6375ae898cb08bfbdbea5d8d55ba463e7#f7916bb6375ae898cb08bfbdbea5d8d55ba463e7" } dependencies = [ { name = "pydantic" }, ] +[package.optional-dependencies] +coordination-client = [ + { name = "httpx" }, +] + [[package]] name = "annotated-types" version = "0.7.0" @@ -276,7 +281,7 @@ name = "hyrule-knowledge" version = "0.1.0" source = { editable = "." } dependencies = [ - { name = "agent-core" }, + { name = "agent-core", extra = ["coordination-client"] }, { name = "pyyaml" }, ] @@ -294,13 +299,14 @@ dev = [ { name = "mcp" }, { name = "mypy" }, { name = "pytest" }, + { name = "pytest-asyncio" }, { name = "ruff" }, { name = "types-pyyaml" }, ] [package.metadata] requires-dist = [ - { name = "agent-core", git = "https://github.com/AS215932/agent-core?tag=v0.8.0" }, + { name = "agent-core", extras = ["coordination-client"], git = "https://github.com/AS215932/agent-core?rev=f7916bb6375ae898cb08bfbdbea5d8d55ba463e7" }, { name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.27.0" }, { name = "pyyaml", specifier = ">=6.0.2" }, { name = "starlette", marker = "extra == 'api'", specifier = ">=0.37" }, @@ -313,6 +319,7 @@ dev = [ { name = "mcp", specifier = ">=1.27.0" }, { name = "mypy", specifier = ">=1.11" }, { name = "pytest", specifier = ">=8" }, + { name = "pytest-asyncio", specifier = ">=0.23" }, { name = "ruff", specifier = ">=0.6" }, { name = "types-pyyaml", specifier = ">=6.0.12" }, ] @@ -679,6 +686,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8b/5a/ba30a81239b909821b3153e303e7def45178bf353da4f72380e6c5e8793b/pytest-9.1.0-py3-none-any.whl", hash = "sha256:8ebb0e7888bdf2bdfc602ec51f8f62d50200af37356c74e503c79a94f5c81f32", size = 386453, upload-time = "2026-06-13T18:52:44.045Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/7c/d36d04db312ecf4298932ef77e6e4a9e8ad017906e24e34f0b0c361a2473/pytest_asyncio-1.4.0.tar.gz", hash = "sha256:c6c0d2259945122819f171a32ecea2c349ead889ee28176caaf492143424be42", size = 58514, upload-time = "2026-05-26T09:56:04.083Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/e2/08a497ef684b88559c9cc5f4ad53a37e7b99e727094a86d6ea32536d5d3c/pytest_asyncio-1.4.0-py3-none-any.whl", hash = "sha256:933ca923a23075a87fb7070c0ec272a6848489824d887c85c812670932835aa1", size = 16930, upload-time = "2026-05-26T09:56:02.576Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.2" From 0c890abbe0e376621b0d9f751bee32510b6998e1 Mon Sep 17 00:00:00 2001 From: David Hyrule Date: Sat, 11 Jul 2026 19:50:55 +0200 Subject: [PATCH 2/4] pin final agent-core coordination client --- pyproject.toml | 2 +- uv.lock | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index fa8c378..6322675 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,4 +56,4 @@ testpaths = ["tests"] asyncio_mode = "auto" [tool.uv.sources] -agent-core = { git = "https://github.com/AS215932/agent-core", rev = "f7916bb6375ae898cb08bfbdbea5d8d55ba463e7" } +agent-core = { git = "https://github.com/AS215932/agent-core", rev = "46bea255d60501276825fb9be7a75746f6062841" } diff --git a/uv.lock b/uv.lock index fe49068..c303a16 100644 --- a/uv.lock +++ b/uv.lock @@ -12,7 +12,7 @@ resolution-markers = [ [[package]] name = "agent-core" version = "0.9.0" -source = { git = "https://github.com/AS215932/agent-core?rev=f7916bb6375ae898cb08bfbdbea5d8d55ba463e7#f7916bb6375ae898cb08bfbdbea5d8d55ba463e7" } +source = { git = "https://github.com/AS215932/agent-core?rev=46bea255d60501276825fb9be7a75746f6062841#46bea255d60501276825fb9be7a75746f6062841" } dependencies = [ { name = "pydantic" }, ] @@ -306,7 +306,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "agent-core", extras = ["coordination-client"], git = "https://github.com/AS215932/agent-core?rev=f7916bb6375ae898cb08bfbdbea5d8d55ba463e7" }, + { name = "agent-core", extras = ["coordination-client"], git = "https://github.com/AS215932/agent-core?rev=46bea255d60501276825fb9be7a75746f6062841" }, { name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.27.0" }, { name = "pyyaml", specifier = ">=6.0.2" }, { name = "starlette", marker = "extra == 'api'", specifier = ">=0.37" }, From e8efd3cfb9e608699b36c427ddab36a8cd77f231 Mon Sep 17 00:00:00 2001 From: David Hyrule Date: Sat, 11 Jul 2026 20:17:57 +0200 Subject: [PATCH 3/4] ci: request post-merge app promotion --- .github/workflows/request-promotion.yml | 69 +++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 .github/workflows/request-promotion.yml diff --git a/.github/workflows/request-promotion.yml b/.github/workflows/request-promotion.yml new file mode 100644 index 0000000..849ff1a --- /dev/null +++ b/.github/workflows/request-promotion.yml @@ -0,0 +1,69 @@ +name: request-promotion + +on: + workflow_dispatch: + inputs: + sha: + description: Commit SHA to promote; defaults to the current ref + required: false + type: string + workflow_run: + workflows: [validate] + types: [completed] + +permissions: + contents: read + +concurrency: + group: request-promotion-${{ github.event.workflow_run.head_sha || inputs.sha || github.sha }} + cancel-in-progress: false + +jobs: + request: + if: > + github.event_name == 'workflow_dispatch' || + (github.event.workflow_run.conclusion == 'success' && + github.event.workflow_run.head_branch == 'main' && + github.event.workflow_run.event == 'push') + runs-on: [self-hosted, linux, x64, hyrule-public-pr] + timeout-minutes: 5 + steps: + - name: Generate promotion app token + id: app-token + uses: actions/create-github-app-token@v2 + with: + app-id: ${{ secrets.PROMOTION_APP_ID }} + private-key: ${{ secrets.PROMOTION_APP_PRIVATE_KEY }} + owner: AS215932 + repositories: network-operations + + - name: Request network-operations promotion PR + env: + GH_TOKEN: ${{ github.token }} + PROMOTION_TOKEN: ${{ steps.app-token.outputs.token }} + REPOSITORY: ${{ github.repository }} + INPUT_SHA: ${{ inputs.sha }} + WORKFLOW_RUN_SHA: ${{ github.event.workflow_run.head_sha }} + run: | + set -euo pipefail + app="${REPOSITORY#AS215932/}" + SHA="${INPUT_SHA:-${WORKFLOW_RUN_SHA:-${GITHUB_SHA}}}" + SHA="${SHA,,}" + + if ! [[ "$SHA" =~ ^[0-9a-f]{40}$ ]]; then + echo "::error::promotion SHA must be a 40-character hex commit SHA" + exit 1 + fi + + canonical="$(gh api "repos/${REPOSITORY}/commits/${SHA}" --jq .sha)" + if [ "$canonical" != "$SHA" ]; then + echo "::error::commit ${SHA} was not found in ${REPOSITORY}" + exit 1 + fi + + GH_TOKEN="$PROMOTION_TOKEN" gh api --method POST repos/AS215932/network-operations/dispatches \ + -f event_type=app-promote \ + -f "client_payload[repository]=$REPOSITORY" \ + -f "client_payload[sha]=$SHA" \ + -f "client_payload[title]=Promote ${app} ${SHA:0:7}" \ + -f "client_payload[impact]=Automated promotion request from ${REPOSITORY}@${SHA}." From b334c793f893677afaf41ab92d8bc81013b796cc Mon Sep 17 00:00:00 2001 From: David Hyrule Date: Sat, 11 Jul 2026 20:38:36 +0200 Subject: [PATCH 4/4] fix: harden coordinator review boundaries --- src/hyrule_knowledge/coordination.py | 25 +++++++++++++++++- tests/test_coordination_v2.py | 38 ++++++++++++++++++++++++++-- 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/src/hyrule_knowledge/coordination.py b/src/hyrule_knowledge/coordination.py index bafaeb2..72d1c34 100644 --- a/src/hyrule_knowledge/coordination.py +++ b/src/hyrule_knowledge/coordination.py @@ -202,6 +202,14 @@ def _resolve_context(self, record: HandoffRecord) -> HandoffResult: raise ValueError( f"source loop may not request Knowledge context: {envelope.source_loop}" ) + if envelope.source_loop == "soc" and ( + envelope.constraints.get("untrusted_loop_text") is not True + or envelope.constraints.get("model_execution") != "forbidden" + or envelope.payload.get("input_trust") != "untrusted_telemetry" + ): + raise ValueError( + "SOC context must preserve the untrusted-telemetry/no-model marker" + ) task = _text( envelope.payload.get("query") or envelope.payload.get("task") @@ -254,6 +262,9 @@ def _resolve_context(self, record: HandoffRecord) -> HandoffResult: "unresolved_questions": pack.unresolved_questions[:20], "gap_detected": bool(pack.unresolved_questions), "read_only": True, + "input_trust": "untrusted_telemetry" + if envelope.source_loop == "soc" + else "loop_request", } return HandoffResult( handoff_id=envelope.handoff_id, @@ -282,6 +293,11 @@ def _stage_learning_proposal(self, record: HandoffRecord) -> HandoffResult: if profile is None: raise ValueError(f"source loop may not contribute learning: {envelope.source_loop}") producer, allowed_event_types = profile + requested_producer = _text(envelope.payload.get("producer"), limit=100) + if requested_producer and requested_producer != producer: + raise ValueError( + f"producer {requested_producer!r} does not match signed source loop" + ) event_type = _text(envelope.payload.get("event_type"), limit=100) if event_type not in allowed_event_types: raise ValueError( @@ -371,7 +387,14 @@ def _write_proposal(self, event: dict[str, Any]) -> Path: handle.write(encoded) handle.flush() os.fsync(handle.fileno()) - os.replace(temporary, target) + try: + # An atomic hard-link publish cannot overwrite a proposal that + # another worker won the race to create. The temporary file is + # on the same spool filesystem by construction. + os.link(temporary, target) + except FileExistsError: + if target.read_text(encoding="utf-8") != encoded: + raise LearningLedgerError(f"proposal id collision: {event['id']}") finally: if temporary.exists(): temporary.unlink() diff --git a/tests/test_coordination_v2.py b/tests/test_coordination_v2.py index 6253277..ca02b43 100644 --- a/tests/test_coordination_v2.py +++ b/tests/test_coordination_v2.py @@ -28,13 +28,21 @@ def _context_record() -> HandoffRecord: payload={ "query": "SOC Agent posture, network operations, and security policy", "role": "engineering_loop", + "input_trust": "untrusted_telemetry", + }, + constraints={ + "untrusted_loop_text": True, + "model_execution": "forbidden", + "read_only": True, }, idempotency_key="knowledge-context-test", ) return HandoffRecord(envelope=envelope, status="claimed", claim_owner="knowledge") -def _learning_record(*, unsafe: bool = False) -> HandoffRecord: +def _learning_record( + *, unsafe: bool = False, producer: str = "soc_shadow" +) -> HandoffRecord: metrics = {"stdout": "raw command data"} if unsafe else {"consecutive_passes": 3} envelope = HandoffEnvelope( source_loop="soc", @@ -47,7 +55,7 @@ def _learning_record(*, unsafe: bool = False) -> HandoffRecord: approval_tier="operator", payload={ "event_type": "soc_case_outcome", - "producer": "soc_shadow", + "producer": producer, "subject": "service:soc-agent", "summary": "Three cited verification cycles confirmed the drift was resolved.", "metrics": metrics, @@ -92,6 +100,7 @@ def test_soc_context_role_is_derived_from_loop_identity(tmp_path: Path) -> None: assert result.payload["role"] == "soc_shadow" assert result.payload["policy_result"] == "allow" assert result.payload["read_only"] is True + assert result.payload["input_trust"] == "untrusted_telemetry" section_names = {section["name"] for section in result.payload["sections"]} assert "safe_security_boundaries" in section_names @@ -125,6 +134,31 @@ def test_soc_learning_adversarial_validation_rejects_raw_output(tmp_path: Path) worker._stage_learning_proposal(_learning_record(unsafe=True)) +def test_soc_context_requires_untrusted_no_model_marker(tmp_path: Path) -> None: + worker = KnowledgeCoordinatorWorker( + object(), # type: ignore[arg-type] + store_path=ROOT / "exports/knowledge.sqlite", + policy_path=ROOT / "knowledge-policy.yml", + proposal_dir=tmp_path, + ) + record = _context_record() + envelope = record.envelope.model_copy(update={"constraints": {}}) + record = record.model_copy(update={"envelope": envelope}) + with pytest.raises(ValueError, match="untrusted-telemetry/no-model"): + worker._resolve_context(record) + + +def test_learning_producer_must_match_signed_source_loop(tmp_path: Path) -> None: + worker = KnowledgeCoordinatorWorker( + object(), # type: ignore[arg-type] + proposal_dir=tmp_path, + ) + with pytest.raises(ValueError, match="does not match signed source loop"): + worker._stage_learning_proposal( + _learning_record(producer="engineering_loop") + ) + + class _FakeClient: def __init__(self, record: HandoffRecord) -> None: self.record = record