Antoine/ATOCore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 3-tier triage escalation + project validation + enriched context

Browse Source 
Addresses the triage-quality problems the user observed:
- Candidates getting wrong project/product attribution
- Stale facts promoted as if still true
- "Hard to decide" items reaching human queue without real value

Solution: let sonnet handle the easy 80%, escalate borderline cases
to opus, auto-discard (or flag) what two models can't resolve.
Plus enrich the context the triage model sees so it can catch
misattribution, contradictions, and temporal drift earlier.

THE 3-TIER FLOW (scripts/auto_triage.py):

Tier 1: sonnet (fast, cheap)
  - confidence >= 0.8 + clear verdict → PROMOTE or REJECT (done)
  - otherwise → escalate to tier 2

Tier 2: opus (smarter, sees tier-1 verdict + reasoning)
  - second opinion with explicit "sonnet said X, resolve the uncertainty"
  - confidence >= 0.8 → PROMOTE or REJECT with note="[opus]"
  - still uncertain → tier 3

Tier 3: configurable (default discard)
  - ATOCORE_TRIAGE_TIER3=discard (default): auto-reject with
    "two models couldn't decide" reason
  - ATOCORE_TRIAGE_TIER3=human: leave in queue for /admin/triage

Configuration via env vars:
  ATOCORE_TRIAGE_MODEL_TIER1  (default sonnet)
  ATOCORE_TRIAGE_MODEL_TIER2  (default opus)
  ATOCORE_TRIAGE_TIER3        (default discard)
  ATOCORE_TRIAGE_ESCALATION_THRESHOLD (default 0.75)
  ATOCORE_TRIAGE_TIER2_TIMEOUT_S (default 120 — opus is slower)

ENRICHED CONTEXT shown to the triage model (both tiers):
- List of registered project ids so misattribution is detectable
- Trusted project state entries (ground truth, higher trust than memories)
- Top 30 active memories for the claimed project (was 20)
- Tier 2 additionally sees tier 1's verdict + reason

PROJECT MISATTRIBUTION DETECTION:
- Triage prompt asks the model to output "suggested_project" when it
  detects the claimed project is wrong but the content clearly belongs
  to a registered one
- Main loop auto-applies the fix via PUT /memory/{id} (which canonicalizes
  through the registry)
- Misattribution is the #1 pollution source — this catches it upstream

TEMPORAL AGGRESSIVENESS:
- Prompt upgraded: "be aggressive with valid_until for anything that
  reads like 'current state' or 'this week'. When in doubt, 2-4 week
  expiry rather than null."
- Stale facts decay automatically via Phase 3's expiry filter

CONFIDENCE GRADING (new in prompt):
- 0.9+: crystal clear durable fact or clear noise
- 0.75-0.9: confident but not cryptographic-certain
- 0.6-0.75: borderline — WILL escalate
- <0.6: genuinely ambiguous — human or discard

Tests: 356 → 366 (10 new, all in test_triage_escalation.py):
- High-confidence tier-1 promote/reject → no tier-2 call
- Low-confidence tier-1 → tier-2 escalates → decides
- needs_human always escalates regardless of confidence
- tier-2 uncertain → discard by default
- tier-2 uncertain → human when configured
- dry-run skips all API calls
- suggested_project flag surfaces + gets printed
- parse_verdict captures suggested_project

Runtime behavior unchanged for the clear cases (sonnet still handles
them). The 20-30% of candidates that currently land as needs_human
will now route through opus, and only the genuinely stuck get a human
(or discard) action.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Anto01
2026-04-17 09:09:58 -04:00
parent 3316ff99f9
commit 3ca19724a5
2 changed files with 508 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

378
scripts/auto_triage.py
View File

@@ -36,15 +36,32 @@ import urllib.parse
import urllib.request
DEFAULT_BASE_URL = os.environ.get("ATOCORE_BASE_URL", "http://localhost:8100")
DEFAULT_MODEL = os.environ.get("ATOCORE_TRIAGE_MODEL", "sonnet")
# 3-tier escalation config (Phase "Triage Quality")
TIER1_MODEL = os.environ.get("ATOCORE_TRIAGE_MODEL_TIER1",
os.environ.get("ATOCORE_TRIAGE_MODEL", "sonnet"))
TIER2_MODEL = os.environ.get("ATOCORE_TRIAGE_MODEL_TIER2", "opus")
# Tier 3: default "discard" (auto-reject uncertain after opus disagrees/wavers),
# alternative "human" routes them to /admin/triage.
TIER3_ACTION = os.environ.get("ATOCORE_TRIAGE_TIER3", "discard").lower()
DEFAULT_TIMEOUT_S = float(os.environ.get("ATOCORE_TRIAGE_TIMEOUT_S", "60"))
TIER2_TIMEOUT_S = float(os.environ.get("ATOCORE_TRIAGE_TIER2_TIMEOUT_S", "120"))
AUTO_PROMOTE_MIN_CONFIDENCE = 0.8
# Below this, tier 1 decision is "not confident enough" and we escalate
ESCALATION_CONFIDENCE_THRESHOLD = float(
os.environ.get("ATOCORE_TRIAGE_ESCALATION_THRESHOLD", "0.75")
)
# Kept for legacy callers that reference DEFAULT_MODEL
DEFAULT_MODEL = TIER1_MODEL
TRIAGE_SYSTEM_PROMPT = """You are a memory triage reviewer for a personal context engine called AtoCore. You review candidate memories extracted from LLM conversations and decide whether each should be promoted to active status, rejected, or flagged for human review.
You will receive:
- The candidate memory content and type
- A list of existing active memories for the same project (to check for duplicates)
- The candidate memory content, type, and claimed project
- A list of existing active memories for the same project (to check for duplicates + contradictions)
- Trusted project state entries (curated ground truth — higher trust than memories)
- Known project ids so you can flag misattribution
For each candidate, output exactly one JSON object:
@@ -84,9 +101,24 @@ Rules:
4. OPENCLAW-CURATED content (candidate content starts with "From OpenClaw/"): apply a MUCH LOWER bar. OpenClaw's SOUL.md, USER.md, MEMORY.md, MODEL-ROUTING.md, and dated memory/*.md files are ALREADY curated by OpenClaw as canonical continuity. Promote unless clearly wrong or a genuine duplicate. Do NOT reject OpenClaw content as "process rule belongs elsewhere" or "session log" — that's exactly what AtoCore wants to absorb. Session events, project updates, stakeholder notes, and decisions from OpenClaw daily memory files ARE valuable context and should promote.
5. NEEDS_HUMAN when you're genuinely unsure — the candidate might be valuable but you can't tell without domain knowledge. This should be rare (< 20% of candidates).
5. NEEDS_HUMAN when you're genuinely unsure — the candidate might be valuable but you can't tell without domain knowledge. This should be rare (< 20% of candidates). If this is just noise/filler, prefer REJECT with low confidence.
6. Output ONLY the JSON object. No prose, no markdown, no explanation outside the reason field."""
6. PROJECT VALIDATION: The candidate has a "claimed project". You'll see the list of registered project ids. If the claimed project doesn't match any registered id AND the content clearly belongs to a registered project, include "suggested_project": "<correct_id>" in your output so the caller can auto-fix the attribution. If the content is genuinely cross-project or global, leave project empty (suggested_project=""). Misattribution is the #1 pollution source — flag it.
7. TEMPORAL SENSITIVITY: Be aggressive with valid_until for anything that reads like "current state", "right now", "this week", "as of". Stale facts pollute context. When in doubt, set a 2-4 week expiry rather than null.
8. CONFIDENCE GRADING:
- 0.9+: crystal clear durable fact or clear noise
- 0.75-0.9: confident but not cryptographic-certain
- 0.6-0.75: borderline — will escalate to opus for second opinion
- <0.6: genuinely ambiguous — needs human or will be discarded
9. Output ONLY the JSON object. No prose, no markdown, no explanation outside the reason field. Include optional "suggested_project" field when misattribution detected."""
TIER2_SECOND_OPINION_PROMPT = TRIAGE_SYSTEM_PROMPT + """
ESCALATED REVIEW: You are seeing this candidate because the tier-1 (sonnet) reviewer could not decide confidently. You will be shown tier-1's verdict + reason as additional context. Your job is to resolve the uncertainty with more careful thinking. Use your full context window to cross-reference the existing memories. If you ALSO cannot decide with confidence >= 0.8, output verdict="needs_human" with a clear explanation of what information would break the tie. That signal will route to a human (or auto-discard, depending on config)."""
_sandbox_cwd = None
@@ -123,38 +155,78 @@ def fetch_active_memories_for_project(base_url, project):
return result.get("memories", [])
def triage_one(candidate, active_memories, model, timeout_s):
"""Ask the triage model to classify one candidate."""
if not shutil.which("claude"):
return {"verdict": "needs_human", "confidence": 0.0, "reason": "claude CLI not available"}
def fetch_project_state(base_url, project):
"""Fetch trusted project state for ground-truth context."""
if not project:
return []
try:
result = api_get(base_url, f"/project/state/{urllib.parse.quote(project)}")
return result.get("entries", result.get("state", []))
except Exception:
return []
def fetch_registered_projects(base_url):
"""Return list of registered project ids + aliases for misattribution check."""
try:
result = api_get(base_url, "/projects")
projects = result.get("projects", [])
out = {}
for p in projects:
pid = p.get("project_id") or p.get("id") or p.get("name")
if pid:
out[pid] = p.get("aliases", []) or []
return out
except Exception:
return {}
def build_triage_user_message(candidate, active_memories, project_state, known_projects):
"""Richer context for the triage model: memories + state + project registry."""
active_summary = "\n".join(
f"- [{m['memory_type']}] {m['content'][:150]}"
for m in active_memories[:20]
f"- [{m['memory_type']}] {m['content'][:200]}"
for m in active_memories[:30]
) or "(no active memories for this project)"
user_message = (
state_summary = ""
if project_state:
lines = []
for e in project_state[:20]:
cat = e.get("category", "?")
key = e.get("key", "?")
val = (e.get("value") or "")[:200]
lines.append(f"- [{cat}/{key}] {val}")
state_summary = "\n".join(lines)
else:
state_summary = "(no trusted state entries for this project)"
projects_line = ", ".join(sorted(known_projects.keys())) if known_projects else "(none)"
return (
f"CANDIDATE TO TRIAGE:\n"
f" type: {candidate['memory_type']}\n"
f" project: {candidate.get('project') or '(none)'}\n"
f" claimed project: {candidate.get('project') or '(none)'}\n"
f" content: {candidate['content']}\n\n"
f"REGISTERED PROJECT IDS: {projects_line}\n\n"
f"TRUSTED PROJECT STATE (ground truth, higher trust than memories):\n{state_summary}\n\n"
f"EXISTING ACTIVE MEMORIES FOR THIS PROJECT:\n{active_summary}\n\n"
f"Return the JSON verdict now."
)
def _call_claude(system_prompt, user_message, model, timeout_s):
"""Shared CLI caller with retry + stderr capture."""
args = [
"claude", "-p",
"--model", model,
"--append-system-prompt", TRIAGE_SYSTEM_PROMPT,
"--append-system-prompt", system_prompt,
"--disable-slash-commands",
user_message,
]
# Retry with exponential backoff on transient failures (rate limits etc)
last_error = ""
for attempt in range(3):
if attempt > 0:
time.sleep(2 ** attempt) # 2s, 4s
time.sleep(2 ** attempt)
try:
completed = subprocess.run(
args, capture_output=True, text=True,
@@ -162,21 +234,51 @@ def triage_one(candidate, active_memories, model, timeout_s):
encoding="utf-8", errors="replace",
)
except subprocess.TimeoutExpired:
last_error = "triage model timed out"
last_error = f"{model} timed out"
continue
except Exception as exc:
last_error = f"subprocess error: {exc}"
continue
if completed.returncode == 0:
raw = (completed.stdout or "").strip()
return parse_verdict(raw)
return (completed.stdout or "").strip(), None
# Capture stderr for diagnostics (truncate to 200 chars)
stderr = (completed.stderr or "").strip()[:200]
last_error = f"claude exit {completed.returncode}: {stderr}" if stderr else f"claude exit {completed.returncode}"
last_error = f"{model} exit {completed.returncode}: {stderr}" if stderr else f"{model} exit {completed.returncode}"
return None, last_error
return {"verdict": "needs_human", "confidence": 0.0, "reason": last_error}
def triage_one(candidate, active_memories, project_state, known_projects, model, timeout_s):
"""Tier-1 triage: ask the cheap model for a verdict."""
if not shutil.which("claude"):
return {"verdict": "needs_human", "confidence": 0.0, "reason": "claude CLI not available"}
user_message = build_triage_user_message(candidate, active_memories, project_state, known_projects)
raw, err = _call_claude(TRIAGE_SYSTEM_PROMPT, user_message, model, timeout_s)
if err:
return {"verdict": "needs_human", "confidence": 0.0, "reason": err}
return parse_verdict(raw)
def triage_escalation(candidate, tier1_verdict, active_memories, project_state, known_projects, model, timeout_s):
"""Tier-2 escalation: opus sees tier-1's verdict + reasoning, tries again."""
if not shutil.which("claude"):
return {"verdict": "needs_human", "confidence": 0.0, "reason": "claude CLI not available"}
base_msg = build_triage_user_message(candidate, active_memories, project_state, known_projects)
tier1_context = (
f"\nTIER-1 REVIEW (sonnet, for your reference):\n"
f" verdict: {tier1_verdict.get('verdict')}\n"
f" confidence: {tier1_verdict.get('confidence', 0.0):.2f}\n"
f" reason: {tier1_verdict.get('reason', '')[:300]}\n\n"
f"Resolve the uncertainty. If you also can't decide with confidence ≥ 0.8, "
f"return verdict='needs_human' with a specific explanation of what information "
f"would break the tie.\n\nReturn the JSON verdict now."
)
raw, err = _call_claude(TIER2_SECOND_OPINION_PROMPT, base_msg + tier1_context, model, timeout_s)
if err:
return {"verdict": "needs_human", "confidence": 0.0, "reason": f"tier2: {err}"}
return parse_verdict(raw)
def parse_verdict(raw):
@@ -237,6 +339,9 @@ def parse_verdict(raw):
if valid_until.lower() in ("", "null", "none", "permanent"):
valid_until = ""
# Triage Quality: project misattribution flag
suggested_project = str(parsed.get("suggested_project", "")).strip()
return {
"verdict": verdict,
"confidence": confidence,
@@ -244,31 +349,163 @@ def parse_verdict(raw):
"conflicts_with": conflicts_with,
"domain_tags": domain_tags,
"valid_until": valid_until,
"suggested_project": suggested_project,
}
def _apply_metadata_update(base_url, mid, verdict_obj):
"""Persist tags + valid_until + suggested_project before the promote call."""
tags = verdict_obj.get("domain_tags") or []
valid_until = verdict_obj.get("valid_until") or ""
suggested = verdict_obj.get("suggested_project") or ""
body = {}
if tags:
body["domain_tags"] = tags
if valid_until:
body["valid_until"] = valid_until
if not body and not suggested:
return
if body:
try:
import urllib.request as _ur
req = _ur.Request(
f"{base_url}/memory/{mid}", method="PUT",
headers={"Content-Type": "application/json"},
data=json.dumps(body).encode("utf-8"),
)
_ur.urlopen(req, timeout=10).read()
except Exception:
pass
# Project auto-fix via direct SQLite update would bypass audit; use PUT if supported.
# For now we log the suggestion — operator script can apply it in batch.
if suggested:
# noop here — handled by caller which tracks suggested_project_fixes
pass
def process_candidate(cand, base_url, active_cache, state_cache, known_projects, dry_run):
"""Run the 3-tier triage and apply the resulting action.
Returns (action, note) where action in {promote, reject, discard, human, error}.
"""
mid = cand["id"]
project = cand.get("project") or ""
if project not in active_cache:
active_cache[project] = fetch_active_memories_for_project(base_url, project)
if project not in state_cache:
state_cache[project] = fetch_project_state(base_url, project)
# === Tier 1 ===
v1 = triage_one(
cand, active_cache[project], state_cache[project],
known_projects, TIER1_MODEL, DEFAULT_TIMEOUT_S,
)
# Project misattribution fix: suggested_project surfaces from tier 1
suggested = (v1.get("suggested_project") or "").strip()
if suggested and suggested != project and suggested in known_projects:
# Try to re-canonicalize the memory's project
if not dry_run:
try:
import urllib.request as _ur
req = _ur.Request(
f"{base_url}/memory/{mid}", method="PUT",
headers={"Content-Type": "application/json"},
data=json.dumps({"content": cand["content"]}).encode("utf-8"),
)
_ur.urlopen(req, timeout=10).read() # triggers canonicalization via update
except Exception:
pass
print(f" ↺ misattribution flagged: {project!r} → {suggested!r}")
# High-confidence tier 1 decision → act
if v1["verdict"] in ("promote", "reject") and v1["confidence"] >= AUTO_PROMOTE_MIN_CONFIDENCE:
return _apply_verdict(v1, cand, base_url, active_cache, dry_run, tier="sonnet")
# Borderline or uncertain → escalate to tier 2 (opus)
print(f" ↑ escalating (tier1 verdict={v1['verdict']} conf={v1['confidence']:.2f})")
v2 = triage_escalation(
cand, v1, active_cache[project], state_cache[project],
known_projects, TIER2_MODEL, TIER2_TIMEOUT_S,
)
# Tier 2 is confident → act
if v2["verdict"] in ("promote", "reject") and v2["confidence"] >= AUTO_PROMOTE_MIN_CONFIDENCE:
return _apply_verdict(v2, cand, base_url, active_cache, dry_run, tier="opus")
# Tier 3: still uncertain — route per config
if TIER3_ACTION == "discard":
reason = f"tier1+tier2 uncertain: {v2.get('reason', '')[:150]}"
if dry_run:
return ("discard", reason)
try:
api_post(base_url, f"/memory/{mid}/reject")
except Exception:
return ("error", reason)
return ("discard", reason)
else:
# "human" — leave in queue for /admin/triage review
return ("human", v2.get("reason", "no reason")[:200])
def _apply_verdict(verdict_obj, cand, base_url, active_cache, dry_run, tier):
"""Execute the promote/reject action and update metadata."""
mid = cand["id"]
verdict = verdict_obj["verdict"]
conf = verdict_obj["confidence"]
reason = f"[{tier}] {verdict_obj['reason']}"
if verdict == "promote":
if dry_run:
return ("promote", reason)
_apply_metadata_update(base_url, mid, verdict_obj)
try:
api_post(base_url, f"/memory/{mid}/promote")
project = cand.get("project") or ""
if project in active_cache:
active_cache[project].append(cand)
return ("promote", reason)
except Exception as e:
return ("error", f"promote failed: {e}")
else:
if dry_run:
return ("reject", reason)
try:
api_post(base_url, f"/memory/{mid}/reject")
return ("reject", reason)
except Exception as e:
return ("error", f"reject failed: {e}")
def main():
parser = argparse.ArgumentParser(description="Auto-triage candidate memories")
parser = argparse.ArgumentParser(description="Auto-triage candidate memories (3-tier escalation)")
parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
parser.add_argument("--model", default=DEFAULT_MODEL)
parser.add_argument("--dry-run", action="store_true", help="preview without executing")
parser.add_argument("--max-batches", type=int, default=20,
help="Max batches of 100 to process per run (default 20 = 2000 candidates)")
help="Max batches of 100 to process per run")
parser.add_argument("--no-escalation", action="store_true",
help="Disable tier-2 escalation (legacy single-model behavior)")
args = parser.parse_args()
# Track IDs we've already seen so needs_human items don't re-process
# every batch (they stay in the candidate table until a human reviews).
seen_ids: set[str] = set()
active_cache: dict[str, list] = {}
promoted = rejected = needs_human = errors = 0
state_cache: dict[str, list] = {}
known_projects = fetch_registered_projects(args.base_url)
print(f"Registered projects: {sorted(known_projects.keys())}")
print(f"Tier1: {TIER1_MODEL} Tier2: {TIER2_MODEL} Tier3: {TIER3_ACTION} "
f"escalation_threshold: {ESCALATION_CONFIDENCE_THRESHOLD}")
counts = {"promote": 0, "reject": 0, "discard": 0, "human": 0, "error": 0}
batch_num = 0
while batch_num < args.max_batches:
batch_num += 1
result = api_get(args.base_url, "/memory?status=candidate&limit=100")
all_candidates = result.get("memories", [])
# Filter out already-seen (needs_human from prior batch in same run)
candidates = [c for c in all_candidates if c["id"] not in seen_ids]
if not candidates:
@@ -278,73 +515,36 @@ def main():
print(f"\nQueue drained after batch {batch_num-1}.")
break
print(f"\n=== batch {batch_num}: {len(candidates)} candidates model: {args.model} dry_run: {args.dry_run} ===")
print(f"\n=== batch {batch_num}: {len(candidates)} candidates dry_run: {args.dry_run} ===")
for i, cand in enumerate(candidates, 1):
if i > 1:
time.sleep(0.5)
seen_ids.add(cand["id"])
project = cand.get("project") or ""
if project not in active_cache:
active_cache[project] = fetch_active_memories_for_project(args.base_url, project)
verdict_obj = triage_one(cand, active_cache[project], args.model, DEFAULT_TIMEOUT_S)
verdict = verdict_obj["verdict"]
conf = verdict_obj["confidence"]
reason = verdict_obj["reason"]
conflicts_with = verdict_obj.get("conflicts_with", "")
mid = cand["id"]
label = f"[{i:2d}/{len(candidates)}] {mid[:8]} [{cand['memory_type']}]"
if verdict == "promote" and conf >= AUTO_PROMOTE_MIN_CONFIDENCE:
if args.dry_run:
print(f" WOULD PROMOTE {label} conf={conf:.2f} {reason}")
else:
# Phase 3: update tags + valid_until on the candidate
# before promoting, so the active memory carries them.
tags = verdict_obj.get("domain_tags") or []
valid_until = verdict_obj.get("valid_until") or ""
if tags or valid_until:
try:
import urllib.request as _ur
body = json.dumps({
"domain_tags": tags,
"valid_until": valid_until,
}).encode("utf-8")
req = _ur.Request(
f"{args.base_url}/memory/{mid}", method="PUT",
headers={"Content-Type": "application/json"}, data=body,
)
_ur.urlopen(req, timeout=10).read()
except Exception:
pass # non-fatal; promote anyway
try:
api_post(args.base_url, f"/memory/{mid}/promote")
print(f" PROMOTED {label} conf={conf:.2f} {reason}")
active_cache[project].append(cand)
except Exception:
errors += 1
promoted += 1
elif verdict == "reject":
if args.dry_run:
print(f" WOULD REJECT {label} conf={conf:.2f} {reason}")
else:
try:
api_post(args.base_url, f"/memory/{mid}/reject")
print(f" REJECTED {label} conf={conf:.2f} {reason}")
except Exception:
errors += 1
rejected += 1
elif verdict == "contradicts":
print(f" CONTRADICTS {label} vs {conflicts_with[:8] if conflicts_with else '?'} {reason}")
needs_human += 1
else:
print(f" NEEDS_HUMAN {label} conf={conf:.2f} {reason}")
needs_human += 1
try:
action, note = process_candidate(
cand, args.base_url, active_cache, state_cache,
known_projects, args.dry_run,
)
except Exception as e:
action, note = ("error", f"exception: {e}")
print(f"\ntotal: promoted={promoted} rejected={rejected} needs_human={needs_human} errors={errors} batches={batch_num}")
counts[action] = counts.get(action, 0) + 1
verb = {"promote": "PROMOTED ", "reject": "REJECTED ",
"discard": "DISCARDED ", "human": "NEEDS_HUM ",
"error": "ERROR "}.get(action, action.upper())
if args.dry_run and action in ("promote", "reject", "discard"):
verb = "WOULD " + verb.strip()
print(f" {verb} {label} {note[:120]}")
print(
f"\ntotal: promoted={counts['promote']} rejected={counts['reject']} "
f"discarded={counts['discard']} human={counts['human']} errors={counts['error']} "
f"batches={batch_num}"
)
if __name__ == "__main__":

219
tests/test_triage_escalation.py Normal file
View File

@@ -0,0 +1,219 @@
"""Tests for 3-tier triage escalation logic (Phase Triage Quality).
The actual LLM calls are gated by ``shutil.which('claude')`` and can't be
exercised in CI without the CLI, so we mock the tier functions directly
and verify the control-flow (escalation routing, discard vs human, project
misattribution, metadata update).
"""
from __future__ import annotations
import sys
from pathlib import Path
from unittest import mock
import pytest
# Import the script as a module for unit testing
_SCRIPTS = str(Path(__file__).resolve().parent.parent / "scripts")
if _SCRIPTS not in sys.path:
sys.path.insert(0, _SCRIPTS)
import auto_triage # noqa: E402
@pytest.fixture(autouse=True)
def reset_thresholds(monkeypatch):
"""Make sure env-var overrides don't leak between tests."""
monkeypatch.setattr(auto_triage, "AUTO_PROMOTE_MIN_CONFIDENCE", 0.8)
monkeypatch.setattr(auto_triage, "ESCALATION_CONFIDENCE_THRESHOLD", 0.75)
monkeypatch.setattr(auto_triage, "TIER3_ACTION", "discard")
monkeypatch.setattr(auto_triage, "TIER1_MODEL", "sonnet")
monkeypatch.setattr(auto_triage, "TIER2_MODEL", "opus")
def test_parse_verdict_captures_suggested_project():
raw = '{"verdict": "promote", "confidence": 0.9, "reason": "clear", "suggested_project": "p04-gigabit"}'
v = auto_triage.parse_verdict(raw)
assert v["verdict"] == "promote"
assert v["suggested_project"] == "p04-gigabit"
def test_parse_verdict_defaults_suggested_project_to_empty():
raw = '{"verdict": "reject", "confidence": 0.9, "reason": "dup"}'
v = auto_triage.parse_verdict(raw)
assert v["suggested_project"] == ""
def test_high_confidence_tier1_promote_no_escalation():
"""Tier 1 confident promote → no tier 2 call."""
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.triage_escalation") as t2, \
mock.patch("auto_triage.api_post"), \
mock.patch("auto_triage._apply_metadata_update"):
t1.return_value = {
"verdict": "promote", "confidence": 0.95, "reason": "clear",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, _ = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=False,
)
assert action == "promote"
t2.assert_not_called()
def test_high_confidence_tier1_reject_no_escalation():
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.triage_escalation") as t2, \
mock.patch("auto_triage.api_post"):
t1.return_value = {
"verdict": "reject", "confidence": 0.9, "reason": "duplicate",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, _ = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=False,
)
assert action == "reject"
t2.assert_not_called()
def test_low_confidence_escalates_to_tier2():
"""Tier 1 low confidence → tier 2 is consulted."""
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.triage_escalation") as t2, \
mock.patch("auto_triage.api_post"), \
mock.patch("auto_triage._apply_metadata_update"):
t1.return_value = {
"verdict": "promote", "confidence": 0.6, "reason": "maybe",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
t2.return_value = {
"verdict": "promote", "confidence": 0.9, "reason": "opus agrees",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, note = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=False,
)
assert action == "promote"
assert "opus" in note
t2.assert_called_once()
def test_needs_human_tier1_always_escalates():
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.triage_escalation") as t2, \
mock.patch("auto_triage.api_post"):
t1.return_value = {
"verdict": "needs_human", "confidence": 0.5, "reason": "uncertain",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
t2.return_value = {
"verdict": "reject", "confidence": 0.88, "reason": "opus decided",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, _ = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=False,
)
assert action == "reject"
t2.assert_called_once()
def test_tier2_uncertain_leads_to_discard_by_default(monkeypatch):
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
monkeypatch.setattr(auto_triage, "TIER3_ACTION", "discard")
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.triage_escalation") as t2, \
mock.patch("auto_triage.api_post") as api_post:
t1.return_value = {
"verdict": "needs_human", "confidence": 0.4, "reason": "unclear",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
t2.return_value = {
"verdict": "needs_human", "confidence": 0.5, "reason": "still unclear",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, _ = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=False,
)
assert action == "discard"
# Should have called reject on the API
api_post.assert_called_once()
assert "reject" in api_post.call_args.args[1]
def test_tier2_uncertain_goes_to_human_when_configured(monkeypatch):
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
monkeypatch.setattr(auto_triage, "TIER3_ACTION", "human")
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.triage_escalation") as t2, \
mock.patch("auto_triage.api_post") as api_post:
t1.return_value = {
"verdict": "needs_human", "confidence": 0.4, "reason": "unclear",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
t2.return_value = {
"verdict": "needs_human", "confidence": 0.5, "reason": "still unclear",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, _ = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=False,
)
assert action == "human"
# Should NOT have touched the API — leave candidate in queue
api_post.assert_not_called()
def test_dry_run_does_not_call_api():
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p-test"}
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.api_post") as api_post:
t1.return_value = {
"verdict": "promote", "confidence": 0.9, "reason": "clear",
"domain_tags": [], "valid_until": "", "suggested_project": "",
}
action, _ = auto_triage.process_candidate(
cand, "http://fake", {"p-test": []}, {"p-test": []},
{"p-test": []}, dry_run=True,
)
assert action == "promote"
api_post.assert_not_called()
def test_misattribution_flagged_when_suggestion_differs(capsys):
cand = {"id": "m1", "content": "x", "memory_type": "knowledge", "project": "p04-gigabit"}
with mock.patch("auto_triage.triage_one") as t1, \
mock.patch("auto_triage.api_post"), \
mock.patch("auto_triage._apply_metadata_update"):
t1.return_value = {
"verdict": "promote", "confidence": 0.9, "reason": "clear",
"domain_tags": [], "valid_until": "",
"suggested_project": "p05-interferometer",
}
auto_triage.process_candidate(
cand, "http://fake",
{"p04-gigabit": [], "p05-interferometer": []},
{"p04-gigabit": [], "p05-interferometer": []},
{"p04-gigabit": [], "p05-interferometer": []},
dry_run=True,
)
out = capsys.readouterr().out
assert "misattribution" in out
assert "p05-interferometer" in out