"""Journal adapter over the intent-neutral provenance layer. No egress. Journal policy: selected user sources, full canonical rehydration, assistant never a content source, source-local clocks, reorder only with confirmed times. Semantic labels from the model are not verified content. """ from __future__ import annotations import json import re from typing import Any from prompt_budget import ERROR_NO_USER_SOURCES, ERROR_RECONSTRUCTION_INVALID, JournalBudgetError from provenance import ( COVERAGE_ALL_SELECTED_SOURCES, EvidenceRef, ProvenanceError, SourceRecord, SourceRegistry, VerificationPolicy, texts_from_payload, verify, ) JSON_BLOCK = re.compile(r"\{.*\}", re.DOTALL) TIME_RE = re.compile(r"\b\d{1,2}:\d{2}\b|\b\d{1,2}\s*Uhr\b", re.I) CLOCK_STAMP_RE = re.compile(r"^(?:\d{1,2}:\d{2}(?:\s*Uhr)?|\d{1,2}\s*Uhr)$", re.I) DIALOGUE_LINE = re.compile( r"^(?:\[(?Pu\d+|a\d+)\]\s*)?(?Puser|assistant):\s*(?P.*)$" ) USER_SOURCE_ID = re.compile(r"^u\d+$") ASSISTANT_SOURCE_ID = re.compile(r"^a\d+$") REQUIRED_TOP = ("source_order", "chronology") LEGACY_FACT_KEYS = ( "events", "people", "places", "perceptions", "feelings_stated", "evaluations", "quotes", "uncertainties", "plan_changes", "corrections", "evidence", "text", ) SOURCED_LIST_KEYS = ("contradictions", "uncertainties", "plan_changes", "corrections") ALLOWED_CLAIM_KEYS = frozenset({"kind", "evidence"}) ALLOWED_ITEM_KEYS = frozenset({"source_id", "source", "time", "claims"}) ALLOWED_SOURCED_KEYS = frozenset({"source_id", "evidence"}) CONTENT_ROLE = "user" JOURNAL_ALLOWED_ROLES = frozenset({CONTENT_ROLE}) def _fail(reason: str, **extra: Any) -> None: raise JournalBudgetError( ERROR_RECONSTRUCTION_INVALID, diagnostics={"reason": reason, **extra}, ) def parse_dialogue_line(line: str) -> dict[str, str] | None: match = DIALOGUE_LINE.match((line or "").rstrip("\n")) if not match: return None return { "source_id": match.group("source_id") or "", "role": match.group("role"), "body": match.group("body") or "", } def is_dialogue_role_line(line: str) -> bool: return parse_dialogue_line(line) is not None def assign_source_ids(messages: list[dict]) -> list[dict]: """Local, stable IDs in encounter order. User u1…; assistant a1….""" labeled: list[dict] = [] user_n = 0 assistant_n = 0 for message in messages or []: item = dict(message) role = item.get("role") or "user" body = (item.get("body") or "").strip() if role == "user" and body: user_n += 1 item["source_id"] = f"u{user_n}" elif role == "assistant" and body: assistant_n += 1 item["source_id"] = f"a{assistant_n}" else: item["source_id"] = None labeled.append(item) return labeled def expected_user_source_ids(messages: list[dict]) -> list[str]: return [ item["source_id"] for item in assign_source_ids(messages) if item.get("role") == "user" and item.get("source_id") ] def parse_reconstruction(raw: str) -> dict[str, Any]: text = (raw or "").strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.IGNORECASE | re.DOTALL) match = JSON_BLOCK.search(text) if not match: _fail("not_json") try: data = json.loads(match.group(0)) except json.JSONDecodeError as exc: raise JournalBudgetError( ERROR_RECONSTRUCTION_INVALID, diagnostics={"reason": "json_parse"}, ) from exc if not isinstance(data, dict): _fail("not_object") return data def parse_clock(stamp: str) -> tuple[int, int] | None: text = (stamp or "").strip() match = re.search(r"(\d{1,2}):(\d{2})", text) if match: hour, minute = int(match.group(1)), int(match.group(2)) if 0 <= hour <= 23 and 0 <= minute <= 59: return hour, minute return None match = re.search(r"(\d{1,2})\s*Uhr", text, re.I) if match: hour = int(match.group(1)) if 0 <= hour <= 23: return hour, 0 return None def clocks_in_text(text: str) -> set[tuple[int, int]]: found: set[tuple[int, int]] = set() for match in TIME_RE.finditer(text or ""): parsed = parse_clock(match.group(0)) if parsed: found.add(parsed) return found def clock_in_source(stamp: str, source_body: str) -> bool: parsed = parse_clock(stamp) if parsed is None: return False return parsed in clocks_in_text(source_body) def _as_list(value: Any) -> list: if value is None: return [] if isinstance(value, list): return value return [value] def _legacy_payload(raw: dict[str, Any]) -> bool: for key in LEGACY_FACT_KEYS: values = [str(part).strip() for part in _as_list(raw.get(key)) if str(part).strip()] if values: return True return False def _normalize_claim(raw: Any, *, index: int, claim_index: int) -> dict[str, str]: if not isinstance(raw, dict): _fail("claim_shape", index=index, claim_index=claim_index) extra = {key: raw.get(key) for key in raw.keys() if key not in ALLOWED_CLAIM_KEYS} if any(value not in (None, "", [], {}) for value in extra.values()): _fail("unverified_claim", index=index, claim_index=claim_index) evidence = str(raw.get("evidence") or "").strip() if not evidence: _fail("evidence_missing", index=index, claim_index=claim_index) kind = raw.get("kind") return {"kind": "" if kind is None else str(kind), "evidence": evidence} def _normalize_sourced_item(raw: Any, *, key: str, index: int) -> dict[str, str]: if isinstance(raw, str) and raw.strip(): _fail("unverified_claim", key=key, index=index) if not isinstance(raw, dict): _fail("sourced_item_shape", key=key, index=index) extra = {item: raw.get(item) for item in raw.keys() if item not in ALLOWED_SOURCED_KEYS} if any(value not in (None, "", [], {}) for value in extra.values()): _fail("unverified_claim", key=key, index=index) source_id = str(raw.get("source_id") or "").strip() evidence = str(raw.get("evidence") or "").strip() if not source_id: _fail("missing_source_id", key=key, index=index) if not evidence: _fail("evidence_missing", key=key, index=index) return {"source_id": source_id, "evidence": evidence} def _confirmed_minutes(stamp: str | None, cited: str) -> int | None: if not stamp: return None parsed = parse_clock(stamp) if parsed is None or parsed not in clocks_in_text(cited): return None return parsed[0] * 60 + parsed[1] def _reorder_allowed( items: list[dict[str, Any]], expected_ids: list[str], user_by_id: dict[str, str], ) -> bool: chrono_ids = [item["source_id"] for item in items] if chrono_ids == expected_ids: return True minutes: list[int] = [] for item in items: value = _confirmed_minutes(item.get("time"), user_by_id.get(item["source_id"]) or "") if value is None: return False minutes.append(value) for index in range(len(items) - 1): if minutes[index] > minutes[index + 1]: return False if minutes[index] == minutes[index + 1]: left = expected_ids.index(items[index]["source_id"]) right = expected_ids.index(items[index + 1]["source_id"]) if left > right: return False return True def _registry_from_messages(labeled: list[dict]) -> SourceRegistry: registry = SourceRegistry() for item in labeled: source_id = item.get("source_id") role = item.get("role") or "" body = item.get("body") or "" if not source_id or not body.strip(): continue registry.add( SourceRecord( source_id=source_id, role=role, text=body, metadata={"conversation_id": item.get("conversation_id")}, ) ) return registry def _map_provenance(exc: ProvenanceError) -> None: reason = exc.reason details = dict(exc.details or {}) if reason == "unknown_source": _fail("unknown_source_id", **details) if reason == "role_not_allowed": role = details.get("role") source_id = str(details.get("source_id") or "") if role != CONTENT_ROLE or ASSISTANT_SOURCE_ID.match(source_id): _fail("assistant_as_fact", **details) _fail("invalid_source", **details) if reason in {"unverified_excerpt", "canonical_overwrite"}: _fail("unverified_claim", **details) if reason == "wrong_source": _fail("wrong_source", **details) if reason == "duplicate_source_id": _fail("duplicate_source_id", **details) if reason == "missing_source_id": _fail("missing_source_id", **details) _fail(reason, **details) def _journal_policy(expected_ids: list[str], chronology_ids: list[str]) -> VerificationPolicy: return VerificationPolicy( allowed_roles=JOURNAL_ALLOWED_ROLES, coverage=COVERAGE_ALL_SELECTED_SOURCES, selected_ids=tuple(chronology_ids), source_order=tuple(expected_ids), include_unverified_annotations=False, ) def validate_reconstruction( data: dict[str, Any], *, messages: list[dict] | None = None, user_bodies: list[str] | None = None, assistant_bodies: list[str] | None = None, ) -> dict[str, Any]: for key in REQUIRED_TOP: if key not in data: _fail("missing_key", key=key) chronology = data.get("chronology") source_order_raw = data.get("source_order") if not isinstance(chronology, list): _fail("chronology_missing") if not isinstance(source_order_raw, list): _fail("source_order_shape") if messages is not None: labeled = assign_source_ids(messages) else: labeled = [] for index, body in enumerate(user_bodies or [], start=1): if (body or "").strip(): labeled.append({"role": "user", "body": body, "source_id": f"u{index}"}) for index, body in enumerate(assistant_bodies or [], start=1): if (body or "").strip(): labeled.append({"role": "assistant", "body": body, "source_id": f"a{index}"}) expected_ids = [ item["source_id"] for item in labeled if item.get("role") == CONTENT_ROLE and item.get("source_id") ] user_by_id = { item["source_id"]: item.get("body") or "" for item in labeled if item.get("role") == CONTENT_ROLE and item.get("source_id") } source_order = [str(part).strip() for part in source_order_raw] if source_order != expected_ids: _fail("source_order_mismatch", expected=expected_ids, got=source_order) items: list[dict[str, Any]] = [] evidence_refs: list[EvidenceRef] = [] seen: list[str] = [] for index, raw in enumerate(chronology): if not isinstance(raw, dict): _fail("chronology_item_shape", index=index) if _legacy_payload(raw): _fail("unverified_claim", index=index) extra_keys = [ key for key in raw.keys() if key not in ALLOWED_ITEM_KEYS and raw.get(key) not in (None, "", [], {}) ] if extra_keys: _fail("unverified_claim", index=index, keys=extra_keys) if "source" not in raw or raw.get("source") is None: source = None elif not isinstance(raw.get("source"), str): _fail("invalid_source", source=raw.get("source"), index=index) elif raw.get("source") == "": source = None else: source = raw.get("source") claims_raw = raw.get("claims") if claims_raw is None: claims_raw = [] if not isinstance(claims_raw, list): _fail("claims_shape", index=index) source_id = str(raw.get("source_id") or "").strip() if not source_id: _fail("missing_source_id", index=index) if ASSISTANT_SOURCE_ID.match(source_id): _fail("assistant_as_fact", source_id=source_id, index=index) if source_id not in user_by_id: _fail("unknown_source_id", source_id=source_id, index=index) if source_id in seen: _fail("duplicate_source_id", source_id=source_id, index=index) seen.append(source_id) if source is None: _fail("missing_source", source_id=source_id, index=index) if source != CONTENT_ROLE: _fail("invalid_source", source=source, source_id=source_id, index=index) claims = [ _normalize_claim(claim, index=index, claim_index=claim_index) for claim_index, claim in enumerate(claims_raw) ] if not claims: _fail("empty_event", source_id=source_id, index=index) stamp = raw.get("time") if isinstance(stamp, str): stamp = stamp.strip() or None elif stamp is None: stamp = None else: stamp = str(stamp).strip() or None cited = user_by_id[source_id] if stamp: if ( not CLOCK_STAMP_RE.match(stamp) or parse_clock(stamp) is None or not clock_in_source(stamp, cited) ): _fail("time_source_mismatch", source_id=source_id, time=stamp, index=index) items.append({"source_id": source_id, "time": stamp, "claims": claims}) for claim in claims: evidence_refs.append(EvidenceRef(source_id, claim["evidence"])) missing = [sid for sid in expected_ids if sid not in seen] if missing: _fail("incomplete_coverage", missing=missing) if expected_ids and not items: _fail("empty_chronology") if not _reorder_allowed(items, expected_ids, user_by_id): _fail( "reorder_unjustified", expected=expected_ids, chronology=[item["source_id"] for item in items], ) for key in SOURCED_LIST_KEYS: raw_list = data.get(key) if raw_list is None: continue if not isinstance(raw_list, list): _fail("sourced_list_shape", key=key) for index, raw in enumerate(raw_list): row = _normalize_sourced_item(raw, key=key, index=index) if ASSISTANT_SOURCE_ID.match(row["source_id"]): _fail("assistant_as_fact", source_id=row["source_id"], key=key) evidence_refs.append(EvidenceRef(row["source_id"], row["evidence"])) registry = _registry_from_messages(labeled) policy = _journal_policy(expected_ids, [item["source_id"] for item in items]) try: artifact = verify(registry, policy=policy, evidence=evidence_refs) except ProvenanceError as exc: _map_provenance(exc) return artifact.to_payload() def local_verified_artifact(messages: list[dict]) -> dict[str, Any]: """Authoritative coverage from local user sources. No model JSON required.""" labeled = assign_source_ids(messages) users = [ item for item in labeled if item.get("role") == CONTENT_ROLE and item.get("source_id") and (item.get("body") or "").strip() ] if not users: raise JournalBudgetError( ERROR_NO_USER_SOURCES, diagnostics={"reason": "no_user_sources"}, ) expected_ids = [item["source_id"] for item in users] registry = _registry_from_messages(labeled) evidence = [EvidenceRef(item["source_id"], (item.get("body") or "").strip()) for item in users] policy = _journal_policy(expected_ids, expected_ids) try: artifact = verify(registry, policy=policy, evidence=evidence) except ProvenanceError as exc: _map_provenance(exc) return artifact.to_payload() def reconstruction_from_model(raw_content: str, messages: list[dict]) -> tuple[dict[str, Any], dict[str, Any]]: """Stage 1 model is untrusted. Invalid output never drops local user sources.""" try: parsed = parse_reconstruction(raw_content) artifact = validate_reconstruction(parsed, messages=messages) return artifact, {"stage1": "model_accepted"} except JournalBudgetError as exc: if exc.code == ERROR_NO_USER_SOURCES: raise artifact = local_verified_artifact(messages) return artifact, { "stage1": "local_fallback", "model_rejected": exc.code, "reason": (exc.diagnostics or {}).get("reason") or exc.code, } def reconstruction_text(data: dict[str, Any]) -> str: return json.dumps(data, ensure_ascii=False, indent=2) def _claim_for_body(body: str) -> list[dict[str, str]]: claims = [{"kind": "event", "evidence": body}] if TIME_RE.search(body): stamp = TIME_RE.search(body).group(0) claims.append({"kind": "time", "evidence": stamp}) return claims def fake_reconstruction(rendered: str) -> str: """Deterministic stage-1 stand-in. Claims remain untrusted; the adapter rehydrates.""" chronology = [] source_order: list[str] = [] user_n = 0 for line in (rendered or "").splitlines(): parsed = parse_dialogue_line(line) if parsed is None and line.startswith("user:"): parsed = {"role": "user", "body": line[5:].strip(), "source_id": ""} if not parsed or parsed["role"] != "user": continue body = (parsed["body"] or "").strip() if not body: continue user_n += 1 source_id = parsed["source_id"] if USER_SOURCE_ID.match(parsed["source_id"] or "") else f"u{user_n}" source_order.append(source_id) times = TIME_RE.findall(body) chronology.append( { "source_id": source_id, "source": "user", "time": times[0] if times else None, "claims": _claim_for_body(body), } ) payload = { "source_order": source_order, "chronology": chronology, "contradictions": [], "uncertainties": [], "plan_changes": [], "corrections": [], } return json.dumps(payload, ensure_ascii=False) def claim_texts(data: dict[str, Any]) -> list[str]: parts = texts_from_payload(data) if parts: return parts collected: list[str] = [] for item in data.get("chronology") or []: if not isinstance(item, dict): continue for claim in item.get("claims") or []: if isinstance(claim, dict) and claim.get("evidence"): collected.append(str(claim["evidence"])) return collected