examples/reader_policy_zero_model.py
Rama main · réplica de lectura del repositorio original.
"""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))