#!/usr/bin/env python3
"""Knowledge Curator 0.2.0: deterministic, strictly read-only dry-run.
This module intentionally has no Neo4j write transaction and no persistence
other than stdout. Credentials are read from the environment only.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import sys
import unicodedata
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parent
CURATOR_VERSION = "knowledge-curator-ro/0.3.0"
NORMALIZATION_VERSION = "kc-norm-v3"
MODALITIES = {"factual_current", "factual_historical", "planned", "proposed",
"recommended", "assumed", "reported", "tested", "confirmed",
"failed", "rolled_back", "withdrawn"}
POLARITIES = {"positive", "negative", "mixed", "unknown"}
DOMAIN_TYPES = {"KnowledgeClaim", "KnowledgeEvent", "KnowledgeRule", "KnowledgeRuleGroup",
"DocumentVersion", "DocumentScope", "DocumentMetadata", "FactStatusModel",
"CandidateGroup"}
def normalize(value: str, rules: dict[str, Any]) -> str:
text = unicodedata.normalize("NFKC", value).casefold().strip()
text = re.sub(r"\bv\s*(\d+(?:\.\d+)*)\b", r"version \1", text)
text = re.sub(r"\b(\d{1,2})[./-](\d{1,2})[./-](\d{4})\b", r"\3-\2-\1", text)
text = re.sub(r"[\u2010-\u2015\-]+", "-", text)
text = re.sub(r"[^\w\s./:-]", " ", text, flags=re.UNICODE)
text = re.sub(r"\s+", " ", text).strip()
for source, target in rules.get("synonyms", {}).items():
text = re.sub(rf"\b{re.escape(source.casefold())}\b", target.casefold(), text)
return text
def digest(*parts: str) -> str:
return hashlib.sha256("|".join(parts).encode("utf-8")).hexdigest()
def evidence(row: dict[str, Any]) -> dict[str, Any]:
text = row["text"] or ""
return {
"source_document_import_key": row["document_import_key"],
"source_content_hash": row["content_hash"],
"source_chunk_key": row["chunk_key"],
"source_text_hash": row.get("text_hash") or hashlib.sha256(text.encode()).hexdigest(),
"evidence_text": text,
"evidence_start_offset": 0,
"evidence_end_offset": len(text),
"evidence_offset_basis": "analyzed_chunk_text",
"evidence_hash": hashlib.sha256(text.encode()).hexdigest(),
"heading": row.get("heading"),
"relative_path": row["path"],
"extraction_rule_version": "kc-rules-v2",
"curator_version": CURATOR_VERSION,
"normalization_version": NORMALIZATION_VERSION,
}
def source_quality(path: str, text: str, rules: dict[str, Any]) -> tuple[str, int, str]:
low = f"{path}\n{text}".casefold()
# Explicitly weak markers win. This is still a conservative heuristic,
# never an assertion of truth.
for marker in ("unbestätigt", "unverified", "user_stated", "vermutung"):
if marker in low:
item = next(x for x in rules["source_classes"] if x["name"] == "unverified_claim")
return item["name"], item["quality"], f"signal:{marker}"
for item in rules["source_classes"]:
if item["name"] == "unverified_claim":
continue
for signal in item["signals"]:
if signal.casefold() in low:
return item["name"], item["quality"], f"signal:{signal}"
return "unknown", 0, "no deterministic source-class signal"
def status_proposal(reason: str, proposed: str = "needs_review") -> dict[str, Any]:
return {
"status": "needs_review" if reason else "candidate",
"proposed_status": proposed,
"proposed_reason": reason or "Keine automatische autoritative Statuszuweisung.",
"requires_human_approval": True,
}
def claim_key(claim: dict[str, Any], rules: dict[str, Any]) -> str:
fields = [NORMALIZATION_VERSION, claim["claim_type"], claim["subject"],
claim["predicate"], claim["object"], claim["object_type"],
claim["modality"], claim["polarity"], claim["authority_scope"],
claim["temporal_scope"]]
return digest(*(normalize(x, rules) for x in fields))
def base_meta(row: dict[str, Any], source_type: str, quality: int, signal: str) -> dict[str, Any]:
return {"source_type": source_type, "source_quality": quality,
"claim_confidence": 0.0, "verification_state": "unverified",
"asserted_at": None, "observed_at": None, "valid_from": None,
"valid_to": None, "recorded_at": None, "verified_at": None,
"superseded_at": None, "authority_scope": "unknown",
"source_signal": signal}
def make_claim(row: dict[str, Any], rule: dict[str, Any], rules: dict[str, Any],
*, modality: str, polarity: str, temporal_scope: str = "unknown",
subject: str | None = None, predicate: str | None = None,
object_value: str | None = None, authority_scope: str = "unknown") -> dict[str, Any]:
st, quality, signal = source_quality(row["path"], row["text"], rules)
c = {"candidate_type": "KnowledgeClaim", "claim_type": "state_or_relation",
"subject": subject or rule["subject"], "predicate": predicate or rule["predicate"],
"object": object_value or rule["object"], "object_type": "text",
"modality": modality, "polarity": polarity, "temporal_scope": temporal_scope,
**base_meta(row, st, quality, signal), "authority_scope": authority_scope,
"decision_reason": "Dry-Run-Kandidat; keine automatische Wahrheitseinstufung.",
**evidence(row), "extraction_rule": rule["rule"]}
c["claim_key"] = claim_key(c, rules)
c.update(status_proposal("Human-Review vor jeder autoritativen Statuszuweisung."))
return c
def make_rule(row: dict[str, Any], rule: dict[str, Any], rules: dict[str, Any]) -> dict[str, Any]:
st, quality, signal = source_quality(row["path"], row["text"], rules)
body = {"candidate_type": "KnowledgeRule", "rule_type": "normative_policy",
"subject": rule["subject"], "predicate": rule["predicate"],
"object": rule["object"], "authority": "unknown",
"authority_scope": "unknown", "valid_from": None, "valid_to": None,
"approval_state": "unknown", "version": "unknown",
"source_type": st, "source_quality": quality, "source_signal": signal,
**evidence(row), "extraction_rule": rule["rule"]}
body["rule_key"] = digest(NORMALIZATION_VERSION, "KnowledgeRule", normalize(rule["subject"], rules),
normalize(rule["predicate"], rules), normalize(rule["object"], rules))
body.update(status_proposal("Normative Regel; Autorität und Freigabe müssen manuell bestätigt werden."))
return body
def make_event(row: dict[str, Any], rule: dict[str, Any], rules: dict[str, Any],
old_key: str, new_key: str) -> dict[str, Any]:
body = {"candidate_type": "KnowledgeEvent", "event_type": "provider_replacement",
"subject": rule["subject"], "from_object": "deepseek/deepseek-chat",
"to_object": "openrouter/auto", "event_at": None,
"old_state_claim_key": old_key, "new_state_claim_key": new_key,
**evidence(row), "extraction_rule": rule["rule"]}
body["event_key"] = digest(NORMALIZATION_VERSION, "KnowledgeEvent", normalize(body["subject"], rules),
body["event_type"], normalize(body["from_object"], rules),
normalize(body["to_object"], rules))
body.update(status_proposal("Änderungsereignis; zeitliche und autoritative Bestätigung erforderlich."))
return body
def make_version(row: dict[str, Any], rule: dict[str, Any], rules: dict[str, Any]) -> dict[str, Any]:
version = rule["object"]
body = {"candidate_type": "DocumentVersion", "document_identity": rule["subject"],
"version": version, "revision_date": None, "content_hash": row["content_hash"],
"previous_version": None, "revision_relation": "unknown", **evidence(row),
"extraction_rule": rule["rule"]}
body["document_version_key"] = digest(NORMALIZATION_VERSION, "DocumentVersion",
normalize(body["document_identity"], rules), normalize(version, rules),
row["content_hash"])
body.update(status_proposal("Dokumentrevision; kein Wissenskonflikt ohne widersprüchliche Fachclaims."))
return body
def candidates_for(row: dict[str, Any], rule: dict[str, Any], rules: dict[str, Any]) -> list[dict[str, Any]]:
kind = rule["rule"]
if kind in {"direct_system_evidence", "human_review_gate"}:
return [make_rule(row, rule, rules)]
if kind == "version_heading":
return [make_version(row, rule, rules)]
if kind == "explicit_replacement":
old = make_claim(row, {**rule, "object": rule["subject"]}, rules,
modality="factual_historical", polarity="positive",
predicate="used_provider", object_value=rule["subject"],
temporal_scope="historical", authority_scope="unknown")
new = make_claim(row, rule, rules, modality="factual_current", polarity="positive",
predicate="uses_provider", object_value=rule["object"],
temporal_scope="unknown", authority_scope="unknown")
return [old, new, make_event(row, rule, rules, old["claim_key"], new["claim_key"])]
if kind == "historical_decision":
return [make_claim(row, rule, rules, modality="factual_historical", polarity="positive",
temporal_scope="2026-05", authority_scope="unknown")]
if kind == "explicit_user_stated_fact":
return [make_claim(row, rule, rules, modality="reported", polarity="positive",
temporal_scope="unknown", authority_scope="unknown")]
return [make_claim(row, rule, rules, modality="reported", polarity="positive")]
def read_inventory(driver, keys: list[str]) -> dict[str, Any]:
with driver.session() as s:
labels = s.run("CALL db.labels() YIELD label RETURN label ORDER BY label").data()
rels = s.run("CALL db.relationshipTypes() YIELD relationshipType RETURN relationshipType ORDER BY relationshipType").data()
docs = s.run("MATCH (n:ObsidianDocument) RETURN count(n) AS n").single()["n"]
chunks = s.run("MATCH (n:ObsidianChunk) RETURN count(n) AS n").single()["n"]
props = {}
for label in ("ObsidianDocument", "ObsidianChunk"):
props[label] = s.run(f"MATCH (n:{label}) UNWIND keys(n) AS k RETURN collect(distinct k) AS keys").single()["keys"]
rows = [dict(r) for r in s.run("""MATCH (d:ObsidianDocument)-[:HAS_CHUNK]->(c:ObsidianChunk)
WHERE c.chunk_key IN $keys
RETURN d.relative_path AS path,d.import_key AS document_import_key,d.content_hash AS content_hash,
c.chunk_key AS chunk_key,c.chunk_index AS chunk_index,c.heading AS heading,c.text AS text,c.text_hash AS text_hash""", keys=keys)]
return {"labels": labels, "relationship_types": rels, "documents": docs, "chunks": chunks,
"properties": props, "rows": rows}
def classify_test(text: str) -> dict[str, Any]:
raw = text.strip()
low = raw.casefold()
heading_only = bool(re.fullmatch(r"#{1,6}\s+[^\n]+", raw))
fragment = (not raw or raw[:1].islower() or raw.startswith(("t —", "dian-"))
or raw.endswith(("(", "—", "...")))
if heading_only or "ohne aussage" in low or fragment or "?" in text or low.startswith(("frage:", "question:")):
return {"candidate_type": None, "modality": None, "polarity": None, "review": False, "conflict": "none"}
if ("policy:" in low or "muss" in low or "darf nicht" in low or
"darf erst" in low or "nur nach" in low or "nie vorher" in low):
typ, modality = "KnowledgeRule", None
elif any(x in low for x in (" wurde durch ", " wurde zurückgerollt", " wurde aktualisiert", " wurde ersetzt")):
typ, modality = "KnowledgeEvent", "rolled_back" if "zurückgerollt" in low else "factual_historical"
elif "behauptet" not in low and "version" in low and "businessplan" in low:
typ, modality = "DocumentVersion", None
else:
typ = "KnowledgeClaim"
if "zitat:" in low or '"' in text: modality = "reported"
elif "unbestätigt" in low or "unverified" in low: modality = "reported"
elif "verworfen" in low or "zurückgezogen" in low: modality = "withdrawn"
elif any(x in low for x in ("plant", "soll ", "sollte", "geplant")): modality = "planned"
elif "könnte" in low or "vielleicht" in low: modality = "proposed"
elif "getestet" in low: modality = "tested"
elif "bestätigt" in low or "produktiv" in low: modality = "confirmed"
elif "war " in low or "historisch" in low: modality = "factual_historical"
else: modality = "factual_current"
polarity = "negative" if re.search(r"\b(nicht|kein|keine|nie)\b", low) else "positive"
review = True
return {"candidate_type": typ, "modality": modality, "polarity": polarity, "review": review, "conflict": "manual"}
def validate_candidate_shape(candidate: dict[str, Any]) -> dict[str, Any]:
"""Return deterministic quality flags; never changes a candidate silently."""
typ = candidate.get("candidate_type")
text = str(candidate.get("evidence_text") or "").strip()
issues: list[str] = []
if typ in {"KnowledgeClaim", "KnowledgeRule", "KnowledgeEvent"} and not str(candidate.get("subject") or "").strip():
issues.append("empty_subject")
if re.fullmatch(r"#{1,6}\s+[^\n]+", text):
issues.append("heading_only")
if text[:1].islower() or text.startswith(("t —", "dian-")):
issues.append("fragment_at_start")
if candidate.get("object") in (None, ""):
issues.append("empty_object")
if isinstance(candidate.get("object"), str) and candidate["object"].rstrip().endswith(("…", "...")):
issues.append("truncated_object")
if typ not in DOMAIN_TYPES:
issues.append("unknown_type")
if candidate.get("modality") == "factual_current" and any(x in text.casefold() for x in ("historisch", "aus sessions rekonstruiert", "nicht live verifiziert")):
issues.append("historical_marked_current")
if candidate.get("verification_state") == "verified" and "nicht live verifiziert" in text.casefold():
issues.append("unverified_marked_verified")
candidate["quality_issues"] = issues
candidate["extraction_state"] = "incomplete" if issues else candidate.get("extraction_state", "complete")
candidate["subject_needs_review"] = "empty_subject" in issues
return candidate
def run_adversarial(path: Path) -> dict[str, Any]:
tests = json.loads(path.read_text())
results = []
for item in tests:
actual = classify_test(item["text"])
expected = item["expected"]
passed = all(actual.get(k) == expected.get(k) for k in ("candidate_type", "modality", "polarity", "review", "conflict"))
results.append({"id": item["id"], "text": item["text"], "expected": expected, "actual": actual,
"passed": passed})
return {"total": len(results), "passed": sum(x["passed"] for x in results),
"failed": sum(not x["passed"] for x in results), "results": results}
def readable_report(out: dict[str, Any]) -> None:
tests = out["adversarial_tests"]
print("\nREAD-ONLY REPORT", file=sys.stderr)
print(f"curator={out['curator_version']} normalization={out['normalization_version']}", file=sys.stderr)
print(f"adversarial={tests['passed']}/{tests['total']} passed; failed={tests['failed']}", file=sys.stderr)
print(f"automatic_current={out['automatic_current_assignments']} automatic_verified={out['automatic_verified_assignments']}", file=sys.stderr)
if "summary" in out:
print("candidates=" + json.dumps(out["summary"], ensure_ascii=False, sort_keys=True), file=sys.stderr)
print("write_path=absent; graphiti=absent; openrouter=absent; llm=absent", file=sys.stderr)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--uri", default=os.getenv("NEO4J_URI"))
ap.add_argument("--user", default=os.getenv("NEO4J_USER"))
ap.add_argument("--rules", default=str(ROOT / "rules.json"))
ap.add_argument("--tests", default=str(ROOT / "adversarial_tests.json"))
ap.add_argument("--offline-tests", action="store_true")
args = ap.parse_args()
rules = json.loads(Path(args.rules).read_text())
test_report = run_adversarial(Path(args.tests))
out: dict[str, Any] = {"run_id": "kc-ro-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
"curator_version": CURATOR_VERSION, "normalization_version": NORMALIZATION_VERSION,
"read_only": True, "automatic_current_assignments": 0,
"automatic_verified_assignments": 0, "adversarial_tests": test_report}
if args.offline_tests:
print(json.dumps(out, ensure_ascii=False, indent=2)); readable_report(out)
return 0 if not test_report["failed"] else 2
if not args.uri or not args.user or not os.getenv("NEO4J_PASSWORD"):
raise SystemExit("Fehlende Neo4j-Credentials: NEO4J_URI, NEO4J_USER und NEO4J_PASSWORD müssen gesetzt sein.")
from neo4j import GraphDatabase
selected = rules["selections"]
driver = GraphDatabase.driver(args.uri, auth=(args.user, os.environ["NEO4J_PASSWORD"]))
try:
inventory = read_inventory(driver, [x["chunk_key"] for x in selected])
bykey = {r["chunk_key"]: r for r in inventory.pop("rows")}
all_candidates = []
for selection in selected:
row = bykey.get(selection["chunk_key"])
rule = rules["manual_claim_rules"].get(selection["chunk_key"])
if row and rule:
all_candidates.extend(candidates_for(row, rule, rules))
claims = [x for x in all_candidates if x["candidate_type"] == "KnowledgeClaim"]
groups = defaultdict(list)
for c in claims:
groups[(c["subject"], c["predicate"], c["authority_scope"], c["temporal_scope"])].append(c)
conflicts = []
identical = []
for group in groups.values():
objects = {normalize(c["object"], rules) for c in group}
if len(group) > 1 and len(objects) == 1: identical.append([c["claim_key"] for c in group])
if len(objects) > 1: conflicts.append([c["claim_key"] for c in group])
out.update({"neo4j_inventory": inventory, "candidates": all_candidates,
"identical_claim_groups": identical, "conflicts": conflicts,
"summary": {"claim_candidates": len(claims),
"event_candidates": sum(x["candidate_type"] == "KnowledgeEvent" for x in all_candidates),
"rule_candidates": sum(x["candidate_type"] == "KnowledgeRule" for x in all_candidates),
"document_version_candidates": sum(x["candidate_type"] == "DocumentVersion" for x in all_candidates),
"needs_review": sum(x["status"] == "needs_review" for x in all_candidates),
"automatic_current": 0, "automatic_verified": 0}})
print(json.dumps(out, ensure_ascii=False, indent=2)); readable_report(out)
finally:
driver.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())