530 lines
18 KiB
Python
530 lines
18 KiB
Python
"""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"^(?:\[(?P<source_id>u\d+|a\d+)\]\s*)?(?P<role>user|assistant):\s*(?P<body>.*)$"
|
|
)
|
|
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
|