"""Caller-owned synthetic reader policy, not Morgan authority or a natural-language parser. PYTHONPATH=. python examples/reader_policy_zero_model.py NEW_DIRECTORY No provider, private memory or network calls. Raw simulated reader output is preserved. """ from __future__ import annotations import argparse import asyncio import hashlib import json from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, Literal, TypedDict, Unpack from pydantic import BaseModel, ConfigDict from morgan_brain.composition import build_memory_context from morgan_brain.config import Settings from morgan_brain.models import Memory from morgan_brain.providers.openai_compat import OpenAICompatAdapter from morgan_brain.providers.request_budget import ChatRequestOptions from morgan_brain.providers.structured import generate_structured from morgan_brain.providers.wire import ChatMessage, ChatResult, ToolSpec class Event(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) task: str kind: Literal["report", "proof", "permission", "proposal"] value: str class ReaderAnswer(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) verification_status: Literal["verified", "unverified_report", "unknown"] independently_verified: bool shipping_permission: Literal["granted", "revoked", "unknown"] proposed_step: str | None completed_step: str | None action_authority: Literal["none"] class VerificationArtifact(BaseModel): model_config = ConfigDict(extra="forbid", strict=True) user_id: str project: str task: str confirmed: bool completed_step: str | None @dataclass(frozen=True) class ProofDesignation: """Trusted caller input; never accept a path/designation generated by the reader.""" event_id: str artifact_path: Path artifact_sha256: str author_id: str def _active_events( context: dict[str, Any], user_id: str, project: str, task: str, ) -> tuple[dict[str, tuple[dict[str, Any], Event]], bool]: if (context["user_id"], context["project"], context["action_authority"]) != ( user_id, project, "none", ): raise ValueError("scope or authority mismatch") cutoff = datetime.fromisoformat(context["effective_at"]) active: dict[str, tuple[dict[str, Any], Event]] = {} conflict = False for source in context["sources"]: if (source["user_id"], source["project"]) != (user_id, project): raise ValueError("source scope mismatch") # This example accepts episodic typed events, not semantic facts or future records. if source["kind"] != "episodic" or source["status"] != "stored": continue if datetime.fromisoformat(source["created_at"]) > cutoff: continue event = Event.model_validate_json(source["content"]) if event.task != task: continue if event.kind == "permission" and ( source["revision_state"] == "conflicted" or source["revision_truncated"] ): conflict = True if source["revision_state"] == "active" and not source["revision_truncated"]: active[source["id"]] = (source, event) return active, conflict def _reported_state( active: dict[str, tuple[dict[str, Any], Event]], ) -> tuple[set[str], set[str], bool]: permissions = set() proposals = set() reports = False for source, event in active.values(): if event.kind == "report": reports = True elif event.kind == "permission": if source["source"] != "user_stated" or event.value not in {"granted", "revoked"}: raise ValueError("unsupported permission evidence") permissions.add(event.value) elif event.kind == "proposal": proposals.add(event.value) return permissions, proposals, reports def _designated_progress( active: dict[str, tuple[dict[str, Any], Event]], proof: ProofDesignation | None, scope: tuple[str, str, str], ) -> tuple[bool, str | None]: verified = False completed = None if proof is not None: designated = active.get(proof.event_id) if designated is None: raise ValueError("designated proof is unavailable or inactive") source, event = designated content = proof.artifact_path.read_bytes() digest = hashlib.sha256(content).hexdigest() if ( digest != proof.artifact_sha256 or event.kind != "proof" or event.value != digest or source["source"] != "tool_observed" or source["author_id"] != proof.author_id ): raise ValueError("proof designation mismatch") artifact = VerificationArtifact.model_validate_json(content) if artifact.confirmed is not True: raise ValueError("artifact does not confirm progress") if (artifact.user_id, artifact.project, artifact.task) != scope: raise ValueError("verification artifact scope mismatch") verified, completed = True, artifact.completed_step return verified, completed def expected_answer( context: dict[str, Any], *, user_id: str, project: str, task: str, proof: ProofDesignation | None = None, ) -> ReaderAnswer: """Resolve typed source eligibility, report state and designated proof separately. Attribution is not authentication. The public gate envelope is trusted caller input, never a model-supplied replacement. """ active, conflict = _active_events(context, user_id, project, task) permissions, proposals, reports = _reported_state(active) verified, completed = _designated_progress(active, proof, (user_id, project, task)) return ReaderAnswer( verification_status="verified" if verified else "unverified_report" if reports else "unknown", independently_verified=verified, shipping_permission=next(iter(permissions)) if len(permissions) == 1 and not conflict else "unknown", proposed_step=next(iter(proposals)) if len(proposals) == 1 else None, completed_step=completed, action_authority="none", ) def validate_reader(raw: str, context: dict[str, Any], **policy: Any) -> ReaderAnswer: """Keep raw output outside this function; reject rather than repair or execute it.""" answer = ReaderAnswer.model_validate_json(raw) expected = expected_answer(context, **policy) if answer != expected: raise ValueError("reader answer conflicts with caller evidence policy") return answer class ReaderReceiptError(RuntimeError): """Persistence failed; raw_result is recoverable, unvalidated, and never authority.""" def __init__(self, raw_result: ChatResult) -> None: super().__init__("reader receipt persistence failed; raw result available for recovery") self.raw_result = raw_result class RecordedReader: """Record one public adapter result before schema or policy validation.""" def __init__(self, adapter: OpenAICompatAdapter, raw_path: Path) -> None: self.adapter = adapter self.raw_path = raw_path self.calls = 0 @property def request_options(self) -> ChatRequestOptions | None: return self.adapter.request_options async def agenerate( self, messages: list[ChatMessage], *, model: str, tools: list[ToolSpec] | None = None, response_format: dict[str, Any] | None = None, ) -> ChatResult: if self.calls or self.raw_path.exists(): raise ValueError("reader call already consumed or raw receipt exists") # Reserve exclusively before dispatch: invalid paths cannot consume a call. output = self.raw_path.open("x") self.calls += 1 result: ChatResult | None = None try: result = await self.adapter.agenerate( messages, model=model, tools=tools, response_format=response_format ) try: output.write(result.model_dump_json() + "\n") output.flush() except Exception as exc: raise ReaderReceiptError(result) from exc finally: try: output.close() except Exception as exc: if result is not None: raise ReaderReceiptError(result) from exc raise if result.finish_reason != "stop": raise ValueError("incomplete reader response") return result class _RequiredReaderOptions(TypedDict): model: str raw_path: Path context: dict[str, Any] user_id: str project: str task: str class ReaderOptions(_RequiredReaderOptions, total=False): """Closed keyword contract preserving the existing SDK example's call style.""" proof: ProofDesignation | None timeout_s: float request_byte_limit: int async def read_with_policy( adapter: OpenAICompatAdapter, messages: list[ChatMessage], **options: Unpack[ReaderOptions], ) -> ReaderAnswer: """Actual SDK -> strict schema -> caller policy; returns data, never actions. Valid awaited keyword calls and defaults are preserved. The runtime signature is **options; TypedDict exposes the named static contract. Missing/unknown keyword errors occur when awaited, before policy checks or provider dispatch, rather than at Python function-call binding. """ missing = ReaderOptions.__required_keys__ - options.keys() unexpected = options.keys() - ReaderOptions.__annotations__.keys() if missing or unexpected: raise TypeError( f"invalid reader options: missing={sorted(missing)}, unknown={sorted(unexpected)}" ) policy = { "user_id": options["user_id"], "project": options["project"], "task": options["task"], "proof": options.get("proof"), } expected_answer(options["context"], **policy) recorded = RecordedReader(adapter, options["raw_path"]) answer = await asyncio.wait_for( generate_structured( recorded, messages, model=options["model"], schema=ReaderAnswer, json_mode="json_schema", max_reask=0, request_byte_limit=options.get("request_byte_limit", 8192), ), timeout=options.get("timeout_s", 60), ) return validate_reader(answer.model_dump_json(), options["context"], **policy) async def demonstrate(directory: Path) -> dict[str, Any]: owner, project, task = "synthetic-owner", "synthetic-reader-demo", "synthetic-task" writer = build_memory_context( Settings( data_dir=str(directory), owner_user_id=owner, embedding_backend="hash", embedding_dim=16, _env_file=None, ) ) ids: list[str] = [] try: for index, (kind, value, parent) in enumerate( [ ("permission", "granted", None), ("permission", "revoked", 0), ("report", "count reported complete without designated proof", None), ("proposal", "draft-local-summary", None), ] ): identity = await writer.gate.store( Memory( user_id=owner, project=project, source="user_stated", author_id="person:synthetic", created_at=datetime(2026, 1, 1, tzinfo=UTC) + timedelta(seconds=index), content=Event(task=task, kind=kind, value=value).model_dump_json(), revises_event_ids=[ids[parent]] if parent is not None else [], ) ) ids.append(identity) context = await writer.gate.inspect_context( user_id=owner, project=project, evidence_ids=ids, selections=[], effective_at=datetime(2026, 1, 3, tzinfo=UTC), ) finally: writer.conn.close() # Simulated reader result: preserves the exact raw fixture without any model call. raw = expected_answer(context, user_id=owner, project=project, task=task).model_dump_json() (directory / "raw-simulated-reader-response.json").write_text(raw + "\n") result = validate_reader(raw, context, user_id=owner, project=project, task=task) receipt = { "answer": result.model_dump(), "model_calls": 0, "actions_taken": 0, "proof_kind": "no designated proof", "simulation": True, } (directory / "context.json").write_text(json.dumps(context, indent=2) + "\n") (directory / "result.json").write_text(json.dumps(receipt, indent=2) + "\n") return receipt if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("new_directory", type=Path) directory = parser.parse_args().new_directory.expanduser().resolve() directory.mkdir(parents=True, exist_ok=False) print(json.dumps(asyncio.run(demonstrate(directory)), indent=2))