From 70bc8c03881058b1670e96ffa0e750561945dd2f Mon Sep 17 00:00:00 2001 From: agent-54de0bcc6a17828b Date: Fri, 24 Apr 2026 20:58:26 +0200 Subject: [PATCH] build(agent): semicolon#54de0b iteration --- README.md | 9 + idea70_narrativeweave_real_time/__init__.py | 3 + idea70_narrativeweave_real_time/adapters.py | 48 +--- idea70_narrativeweave_real_time/api.py | 30 ++- idea70_narrativeweave_real_time/models.py | 38 ++- idea70_narrativeweave_real_time/narrative.py | 231 +++++++++++++++++++ tests/test_audit.py | 42 ++++ tests/test_replay.py | 9 +- 8 files changed, 353 insertions(+), 57 deletions(-) create mode 100644 idea70_narrativeweave_real_time/narrative.py create mode 100644 tests/test_audit.py diff --git a/README.md b/README.md index 4d8b2ab..72d7d99 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ This repository contains a focused, production-minded foundation for the broader - Deterministic replay function that regenerates NarrativeBlocks from the event log - A simple hash-chain ledger that anchors block versions - A FastAPI app to ingest events and query blocks +- Deterministic claim graph construction, redaction-aware Markdown rendering, and block-to-block audit diffs - Tests and a build-ready pyproject/setup.cfg Getting started @@ -26,6 +27,14 @@ Run the API locally for development: uvicorn idea70_narrativeweave_real_time.api:app --reload +Useful endpoints: + +- `POST /ingest` +- `POST /build_block` +- `GET /block/{block_id}` +- `GET /block/{block_id}/render?redacted=true` +- `GET /block/{left_block_id}/diff/{right_block_id}` + Project layout - idea70_narrativeweave_real_time/: python package diff --git a/idea70_narrativeweave_real_time/__init__.py b/idea70_narrativeweave_real_time/__init__.py index fd86279..c7a5763 100644 --- a/idea70_narrativeweave_real_time/__init__.py +++ b/idea70_narrativeweave_real_time/__init__.py @@ -1,2 +1,5 @@ """idea70_narrativeweave_real_time package""" + +from .models import Claim, ClaimGraph, EvidenceToken, NarrativeBlock, NarrativeDiff, Provenance, Signal, Source + __version__ = "0.1.0" diff --git a/idea70_narrativeweave_real_time/adapters.py b/idea70_narrativeweave_real_time/adapters.py index b7537be..471388a 100644 --- a/idea70_narrativeweave_real_time/adapters.py +++ b/idea70_narrativeweave_real_time/adapters.py @@ -1,7 +1,6 @@ -from typing import Dict, Any, List -from .models import NarrativeBlock, Source, Signal, Provenance -from datetime import datetime -import uuid +from typing import Any, Dict + +from .narrative import build_block_from_events def news_adapter_event(headline: str, uri: str, confidence: float = 0.9) -> Dict[str, Any]: @@ -20,44 +19,3 @@ def news_adapter_event(headline: str, uri: str, confidence: float = 0.9) -> Dict def transcript_adapter_event(speaker: str, text: str, ts: str) -> Dict[str, Any]: return {"type": "transcript", "speaker": speaker, "text": text, "ts": ts} - - -def build_block_from_events(events: List[Dict[str, Any]]) -> NarrativeBlock: - """Deterministically build a NarrativeBlock from a list of events. - - This function demonstrates normalization logic: aggregating sources and signals. - """ - # derive an id deterministically from the concatenated event payloads - concat = "".join([str(e) for e in events]) - block_id = uuid.uuid5(uuid.NAMESPACE_URL, concat).hex - topic = " | ".join(set([e.get("type") for e in events if e.get("type")])) - ts = datetime.utcnow() - sources = [] - signals = [] - sentiment_vals = [] - for e in events: - if e.get("type") == "news": - sources.append(Source(type="news", uri=e.get("uri"), confidence=e.get("confidence"))) - # naive deterministic sentiment from headline length - sentiment_vals.append(len(e.get("headline", "")) % 5 - 2) - if e.get("type") == "transcript": - sources.append(Source(type="transcript", uri=f"transcript://{e.get('speaker')}", confidence=1.0)) - # a toy signal: word count - signals.append(Signal(name="word_count", value=float(len(e.get("text", "").split())), provenance={"speaker": e.get("speaker")})) - - sentiment = None - if sentiment_vals: - sentiment = sum(sentiment_vals) / len(sentiment_vals) - - prov = Provenance(trace_id=block_id) - - block = NarrativeBlock( - id=block_id, - topic=topic, - timestamp=ts, - sources=sources, - signals=signals, - sentiment=sentiment, - provenance=prov, - ) - return block diff --git a/idea70_narrativeweave_real_time/api.py b/idea70_narrativeweave_real_time/api.py index 40283b2..b760805 100644 --- a/idea70_narrativeweave_real_time/api.py +++ b/idea70_narrativeweave_real_time/api.py @@ -1,9 +1,13 @@ +import json +from typing import Any, Dict + from fastapi import FastAPI, HTTPException from pydantic import BaseModel -from typing import Any, Dict + from .storage import EventLog from .adapters import build_block_from_events from .ledger import SimpleLedger +from .narrative import diff_narrative_blocks, render_narrative_block app = FastAPI(title="NarrativeWeave API") @@ -28,12 +32,8 @@ def build_block(): events = _store.read_events() if not events: raise HTTPException(status_code=404, detail="no events") - # events are read as dicts with ts,type,payload - payloads = [e["payload"] for e in events] - block = build_block_from_events(payloads) + block = build_block_from_events(events) # produce deterministic JSON for ledger anchoring - import json - block_json = json.dumps(block.model_dump(), sort_keys=True, default=str) anchor = _ledger.anchor(block_json) _store.save_block(block, ledger_anchor=anchor) @@ -46,3 +46,21 @@ def get_block(block_id: str): if not b: raise HTTPException(status_code=404, detail="block not found") return b.dict() + + +@app.get("/block/{block_id}/render") +def render_block(block_id: str, redacted: bool = False): + b = _store.get_block(block_id) + if not b: + raise HTTPException(status_code=404, detail="block not found") + return {"block_id": block_id, "redacted": redacted, "markdown": render_narrative_block(b, redacted=redacted)} + + +@app.get("/block/{left_block_id}/diff/{right_block_id}") +def diff_blocks(left_block_id: str, right_block_id: str): + left = _store.get_block(left_block_id) + right = _store.get_block(right_block_id) + if not left or not right: + raise HTTPException(status_code=404, detail="block not found") + diff = diff_narrative_blocks(left, right) + return diff.dict() diff --git a/idea70_narrativeweave_real_time/models.py b/idea70_narrativeweave_real_time/models.py index 7e065ad..5bff9a3 100644 --- a/idea70_narrativeweave_real_time/models.py +++ b/idea70_narrativeweave_real_time/models.py @@ -1,5 +1,5 @@ -from typing import List, Optional -from pydantic import BaseModel, Field +from typing import Any, Dict, List, Optional +from pydantic import BaseModel, ConfigDict, Field from datetime import datetime @@ -22,6 +22,8 @@ class Provenance(BaseModel): class NarrativeBlock(BaseModel): + model_config = ConfigDict() + id: str topic: str timestamp: datetime @@ -32,6 +34,32 @@ class NarrativeBlock(BaseModel): scenario_flags: List[str] = Field(default_factory=list) provenance: Optional[Provenance] = None - class Config: - # make JSON serialization deterministic for replay - json_sort_keys = True +class EvidenceToken(BaseModel): + kind: str + uri: str + confidence: Optional[float] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + + +class Claim(BaseModel): + id: str + statement: str + evidence: List[EvidenceToken] = Field(default_factory=list) + confidence: Optional[float] = None + signer: Optional[str] = None + signature: Optional[str] = None + + +class ClaimGraph(BaseModel): + block_id: str + claims: List[Claim] = Field(default_factory=list) + provenance: Optional[Provenance] = None + + +class NarrativeDiff(BaseModel): + left_block_id: str + right_block_id: str + changed_fields: List[Dict[str, Any]] = Field(default_factory=list) + added_claims: List[Claim] = Field(default_factory=list) + removed_claims: List[Claim] = Field(default_factory=list) + modified_claims: List[Dict[str, Any]] = Field(default_factory=list) diff --git a/idea70_narrativeweave_real_time/narrative.py b/idea70_narrativeweave_real_time/narrative.py new file mode 100644 index 0000000..67f42e8 --- /dev/null +++ b/idea70_narrativeweave_real_time/narrative.py @@ -0,0 +1,231 @@ +import hashlib +import json +from datetime import datetime, timezone +from typing import Any, Dict, List, Mapping, Optional, Sequence + +from .models import Claim, ClaimGraph, EvidenceToken, NarrativeBlock, NarrativeDiff, Provenance, Signal, Source + + +def _model_dump(model: Any) -> Dict[str, Any]: + if hasattr(model, "model_dump"): + return model.model_dump() + return model.dict() + + +def _jsonable(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump() + if hasattr(value, "dict"): + return value.dict() + if isinstance(value, dict): + return {key: _jsonable(item) for key, item in value.items()} + if isinstance(value, list): + return [_jsonable(item) for item in value] + if isinstance(value, tuple): + return [_jsonable(item) for item in value] + if isinstance(value, datetime): + return value.isoformat() + return value + + +def _utc_naive(value: Optional[Any]) -> datetime: + if value is None: + return datetime.fromtimestamp(0) + if isinstance(value, datetime): + ts = value + else: + text = str(value).replace("Z", "+00:00") + ts = datetime.fromisoformat(text) + if ts.tzinfo is not None: + ts = ts.astimezone(timezone.utc).replace(tzinfo=None) + return ts + + +def canonical_json(data: Any) -> str: + return json.dumps(data, sort_keys=True, separators=(",", ":"), default=str) + + +def _normalize_events(events: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]: + normalized: List[Dict[str, Any]] = [] + for item in events: + if "payload" in item and "type" in item: + payload = dict(item["payload"]) + event_type = str(item["type"]) + ts = _utc_naive(item.get("ts") or payload.get("ts")) + else: + payload = dict(item) + event_type = str(payload.get("type", "unknown")) + ts = _utc_naive(payload.get("ts")) + normalized.append({"type": event_type, "payload": payload, "ts": ts}) + normalized.sort(key=lambda entry: (entry["ts"], entry["type"], canonical_json(entry["payload"]))) + return normalized + + +def build_claim_graph(block: NarrativeBlock) -> ClaimGraph: + claims: List[Claim] = [] + sources = [_model_dump(source) for source in block.sources] + provenance = _model_dump(block.provenance) if block.provenance else None + + if block.signals: + for signal in block.signals: + signal_data = _model_dump(signal) + evidence: List[EvidenceToken] = [] + for source in block.sources: + source_data = _model_dump(source) + evidence.append( + EvidenceToken( + kind=source_data["type"], + uri=source_data["uri"], + confidence=source_data.get("confidence"), + metadata={}, + ) + ) + if signal_data.get("provenance"): + evidence.append( + EvidenceToken( + kind="signal_provenance", + uri=f"signal://{signal_data['name']}", + confidence=1.0, + metadata=signal_data["provenance"], + ) + ) + + claim_id = hashlib.sha256(canonical_json({"signal": signal_data, "sources": sources}).encode("utf-8")).hexdigest() + statement = f"{block.topic}: {signal_data['name']}={signal_data['value']}" + claims.append( + Claim( + id=claim_id, + statement=statement, + evidence=evidence, + confidence=signal_data.get("value"), + ) + ) + else: + claim_id = hashlib.sha256(canonical_json({"topic": block.topic, "sources": sources}).encode("utf-8")).hexdigest() + claims.append( + Claim( + id=claim_id, + statement=f"{block.topic}: no structured signals captured", + evidence=[ + EvidenceToken(kind=source.type, uri=source.uri, confidence=source.confidence, metadata={}) + for source in block.sources + ], + confidence=block.sentiment, + ) + ) + + return ClaimGraph(block_id=block.id, claims=claims, provenance=block.provenance) + + +def render_narrative_block(block: NarrativeBlock, redacted: bool = False) -> str: + graph = build_claim_graph(block) + lines: List[str] = [] + lines.append(f"# {block.topic}") + lines.append(f"Block ID: `{block.id}`") + lines.append(f"Timestamp: `{block.timestamp.isoformat()}`") + if block.sentiment is not None: + lines.append(f"Sentiment: `{block.sentiment}`") + if block.risk_factors: + lines.append("Risk factors:") + for risk in sorted(block.risk_factors): + lines.append(f"- {risk}") + if block.scenario_flags: + lines.append("Scenario flags:") + for flag in sorted(block.scenario_flags): + lines.append(f"- {flag}") + + lines.append("\n## Claims") + for claim in sorted(graph.claims, key=lambda item: item.id): + lines.append(f"- `{claim.id}` {claim.statement}") + for evidence in sorted(claim.evidence, key=lambda item: (item.kind, item.uri)): + uri = evidence.uri if not redacted else "[redacted]" + lines.append(f" - {evidence.kind}: {uri}") + + if graph.provenance: + lines.append("\n## Provenance") + lines.append(f"- trace_id: `{graph.provenance.trace_id}`") + if graph.provenance.signer: + lines.append(f"- signer: `{graph.provenance.signer}`") + if graph.provenance.ledger_anchor: + lines.append(f"- ledger_anchor: `{graph.provenance.ledger_anchor}`") + + return "\n".join(lines).strip() + "\n" + + +def diff_narrative_blocks(left: NarrativeBlock, right: NarrativeBlock) -> NarrativeDiff: + left_claims = {claim.id: claim for claim in build_claim_graph(left).claims} + right_claims = {claim.id: claim for claim in build_claim_graph(right).claims} + + changed_fields: List[Dict[str, Any]] = [] + for field_name in ["topic", "timestamp", "sentiment", "risk_factors", "scenario_flags", "sources", "signals", "provenance"]: + left_value = getattr(left, field_name) + right_value = getattr(right, field_name) + if left_value != right_value: + changed_fields.append( + { + "field": field_name, + "before": _jsonable(left_value), + "after": _jsonable(right_value), + } + ) + + added_claims = [right_claims[claim_id] for claim_id in sorted(set(right_claims) - set(left_claims))] + removed_claims = [left_claims[claim_id] for claim_id in sorted(set(left_claims) - set(right_claims))] + modified_claims: List[Dict[str, Any]] = [] + for claim_id in sorted(set(left_claims) & set(right_claims)): + if _model_dump(left_claims[claim_id]) != _model_dump(right_claims[claim_id]): + modified_claims.append( + { + "claim_id": claim_id, + "before": _model_dump(left_claims[claim_id]), + "after": _model_dump(right_claims[claim_id]), + } + ) + + return NarrativeDiff( + left_block_id=left.id, + right_block_id=right.id, + changed_fields=changed_fields, + added_claims=added_claims, + removed_claims=removed_claims, + modified_claims=modified_claims, + ) + + +def build_block_from_events(events: Sequence[Mapping[str, Any]]) -> NarrativeBlock: + normalized = _normalize_events(events) + concat = canonical_json(normalized) + block_id = hashlib.sha256(concat.encode("utf-8")).hexdigest() + topic = " | ".join(sorted({entry["type"] for entry in normalized if entry["type"]})) + timestamp = max((entry["ts"] for entry in normalized), default=datetime.fromtimestamp(0)) + + sources: List[Source] = [] + signals: List[Signal] = [] + sentiment_vals = [] + for entry in normalized: + payload = entry["payload"] + if entry["type"] == "news": + sources.append(Source(type="news", uri=str(payload.get("uri", "")), confidence=float(payload.get("confidence", 0.0)))) + sentiment_vals.append(len(payload.get("headline", "")) % 5 - 2) + elif entry["type"] == "transcript": + sources.append(Source(type="transcript", uri=f"transcript://{payload.get('speaker')}", confidence=1.0)) + signals.append( + Signal( + name="word_count", + value=float(len(payload.get("text", "").split())), + provenance={"speaker": payload.get("speaker"), "ts": payload.get("ts")}, + ) + ) + + sentiment = sum(sentiment_vals) / len(sentiment_vals) if sentiment_vals else None + provenance = Provenance(trace_id=block_id) + + return NarrativeBlock( + id=block_id, + topic=topic, + timestamp=timestamp, + sources=sources, + signals=signals, + sentiment=sentiment, + provenance=provenance, + ) diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..62bc152 --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,42 @@ +import os +import sys + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from idea70_narrativeweave_real_time.adapters import build_block_from_events, news_adapter_event, transcript_adapter_event +from idea70_narrativeweave_real_time.narrative import diff_narrative_blocks, render_narrative_block + + +def test_redacted_rendering_keeps_structure(): + events = [ + {"ts": "2024-01-01T00:00:00Z", "type": "news", "payload": news_adapter_event("Shares rise", "https://example.com/story", 0.7)}, + {"ts": "2024-01-01T00:00:01Z", "type": "transcript", "payload": transcript_adapter_event("CFO", "Revenue improved year over year.", "2024-01-01T00:00:01Z")}, + ] + block = build_block_from_events(events) + + clear = render_narrative_block(block) + redacted = render_narrative_block(block, redacted=True) + + assert block.id in clear + assert "[redacted]" in redacted + assert "https://example.com/story" not in redacted + assert "## Claims" in redacted + + +def test_diff_is_deterministic_and_claim_aware(): + left_events = [ + {"ts": "2024-01-01T00:00:00Z", "type": "news", "payload": news_adapter_event("Shares rise", "https://example.com/story", 0.7)}, + ] + right_events = [ + {"ts": "2024-01-01T00:00:00Z", "type": "news", "payload": news_adapter_event("Shares fall", "https://example.com/story", 0.7)}, + {"ts": "2024-01-01T00:00:02Z", "type": "transcript", "payload": transcript_adapter_event("CFO", "Margins compressed.", "2024-01-01T00:00:02Z")}, + ] + + left = build_block_from_events(left_events) + right = build_block_from_events(right_events) + diff = diff_narrative_blocks(left, right) + + assert diff.left_block_id == left.id + assert diff.right_block_id == right.id + assert diff.changed_fields + assert diff.added_claims or diff.modified_claims diff --git a/tests/test_replay.py b/tests/test_replay.py index 1afc1fc..1aef9dd 100644 --- a/tests/test_replay.py +++ b/tests/test_replay.py @@ -8,6 +8,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..") from idea70_narrativeweave_real_time.storage import EventLog from idea70_narrativeweave_real_time.adapters import news_adapter_event, transcript_adapter_event, build_block_from_events from idea70_narrativeweave_real_time.ledger import SimpleLedger +from idea70_narrativeweave_real_time.narrative import render_narrative_block def test_deterministic_replay_and_ledger(): @@ -19,12 +20,14 @@ def test_deterministic_replay_and_ledger(): el.append_event("transcript", transcript_adapter_event("CEO", "We delivered strong growth this quarter.", "2023-01-01T00:00:00Z")) events = el.read_events() - payloads = [e["payload"] for e in events] + payloads = events # build block twice and ensure identical id and deterministic JSON b1 = build_block_from_events(payloads) b2 = build_block_from_events(payloads) assert b1.id == b2.id + assert b1.timestamp == b2.timestamp + assert b1.topic == "news | transcript" ledger = SimpleLedger() import json @@ -34,3 +37,7 @@ def test_deterministic_replay_and_ledger(): # Anchors should be chained and deterministic assert a1 != "" assert a2 != a1 + + rendered = render_narrative_block(b1) + assert "# news | transcript" in rendered + assert "word_count" in rendered