diff --git a/README.md b/README.md index 0bbee59..1fa5750 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,9 @@ repeated objects, not just strings) is built. Cross-call diffing is built as an **opt-in** lossless tier (`proxy --diff` / `install-mcp --diff`); its fluency gate (`fluency --diff`) has passed on both the record surface (4-model panel, 100%) and the nested-record surface (`structure`: diff 100% vs full-terse 94%), so it's cleared for -live use. The Tier 1 lossy modes `truncate` and +live use. Long-chain drift is soaked from both sides: mechanically +(`tests/test_diff_soak.py` — exact reconstruction hundreds of chained hops deep) and +behaviorally (`fluency --diff-soak` — model accuracy vs chain depth up to the keyframe +bound). The Tier 1 lossy modes `truncate` and `drop-to-retrieve` are built (opt-in, off by default); `summarize` remains designed but not yet built — see TECHNICAL.md "Known Limitations". diff --git a/TECHNICAL.md b/TECHNICAL.md index b171560..161ae9d 100644 --- a/TECHNICAL.md +++ b/TECHNICAL.md @@ -273,8 +273,14 @@ gitignored because captured tool output may contain real data. default 5; 0 disables). A client *reconnect* (a new MCP `initialize`) is a stronger reset signal: the proxy drops every per-tool diff base on it, so the next result of each tool re-anchors as a full rather than a delta against a base the rebuilt context - no longer holds (#20). The residual gap — a context *compaction* with no reconnect — is - unobservable over stdio, and is the standing reason `--diff` is opt-in. + no longer holds (#20). Both properties are **soaked, not just unit-tested**: the drift + soak (`tests/test_diff_soak.py`) drives the real Interceptor hundreds of chained hops + deep — interleaved tools, error interludes, a mid-soak reconnect, and a 300-diff + unbounded chain — with an independent client-side reconstructor asserting exactness at + every hop; `terse fluency --diff-soak` measures the model-side analogue (accuracy vs + chain depth, up to the keyframe bound). The residual gap — a context *compaction* with + no reconnect — is unobservable over stdio, and is the standing reason `--diff` is + opt-in. - **Text diff (Tier 0.7 text, #25) covers non-JSON results, but only the codec side is measured so far.** File reads, source excerpts, and log tails get `applicable: False` in `measure_payload` — zero Tier-0 compression — and used to get zero cross-call diff --git a/USAGE.md b/USAGE.md index f920fb5..539a39a 100644 --- a/USAGE.md +++ b/USAGE.md @@ -354,6 +354,25 @@ questions and PASS/FAILs on the worst model — run it before enabling `proxy -- your consumer. The diff is always lossless and only sent when smaller; it falls back to the full compressed form whenever no diff applies or the prior result isn't available. +### The depth dimension: `fluency --diff-soak` + +`--diff` tests one hop (full result + one diff). In production a model reads up to +`--diff-keyframe-interval` (default 5) **consecutive** diffs off one full anchor +before the proxy re-anchors — so the question gating a default-flip is whether +comprehension *drifts* with chain depth: + +``` +# real corpus runs, depths 1..5 (6 chain windows per depth, round-robin across tools): +TERSE_FLUENCY_BASE_URL=... TERSE_FLUENCY_API_KEY=... TERSE_FLUENCY_MODELS=... \ + uv run terse fluency --diff-soak --corpus corpus --trials 3 +``` + +The report shows accuracy by depth (chain form vs full-terse control on the same +final-state questions) and PASS/FAILs both overall and at the deepest tested depth, +worst model. The mechanical half of the same soak — hundreds of chained hops +reconstructed exactly, keyframe cadence, reconnect resets — is pinned in +`tests/test_diff_soak.py` and runs in CI. + ### Text diffing and its fluency check (`fluency --text-diff-eval`) The diff above only reasons about JSON. Non-JSON tool output (file reads, source diff --git a/src/terse/cli.py b/src/terse/cli.py index c593ea0..554b9c2 100644 --- a/src/terse/cli.py +++ b/src/terse/cli.py @@ -376,6 +376,7 @@ def _cmd_fluency(args: argparse.Namespace) -> int: from .policy import default_policy, load_policy from .report import ( build_diff_report, + build_diff_soak_report, build_dropeval_report, build_fluency_report, build_text_diff_report, @@ -432,6 +433,23 @@ def _cmd_fluency(args: argparse.Namespace) -> int: print("\n" + build_terminal_diff_report(results)) return 0 + # Diff-chain soak: the DEPTH dimension --diff can't see (#8/#20 follow-up) — does + # comprehension drift as consecutive diffs chain off one full anchor? Depth 5 is + # the production keyframe bound, so a PASS here covers the deployed worst case. + if args.diff_soak: + answerers = _build_answerers(args, fluency.openai_answerer) + if not answerers: + print("`fluency --diff-soak` needs a configured model: set TERSE_FLUENCY_BASE_URL/" + "_API_KEY/_MODELS.") + return 1 + results = fluency.run_diff_soak(envelopes, answerers, trials=args.trials, + max_depth=args.soak_depth, + per_depth_cap=args.soak_windows) + _write_report(build_diff_soak_report(results), args.out) + if args.bars: + print("\n" + build_terminal_diff_report(results, form_label="chain-form")) + return 0 + # Text-diff mode: does a model reconstruct the current TEXT as well from (previous # text + text-diff) as from the full current text? The text-payload analogue of # --diff above (text_diff.py, Tier 0.7 — non-JSON tool output). @@ -811,6 +829,16 @@ def main(argv: list[str] | None = None) -> int: help="behavioral eval: does a model reconstruct the current TEXT as " "accurately from (previous text + text-diff) as from the full " "text? needs same-tool TEXT corpus pairs + a configured model") + f.add_argument("--diff-soak", action="store_true", + help="drift soak: does comprehension degrade as a model reads chains " + "of consecutive diffs off one full anchor? scores depths " + "1..--soak-depth over real corpus runs (needs a configured model)") + f.add_argument("--soak-depth", type=int, default=5, metavar="K", + help="--diff-soak: deepest chain to test (default 5 — the production " + "keyframe interval, i.e. the deployed worst case)") + f.add_argument("--soak-windows", type=int, default=6, metavar="N", + help="--diff-soak: max chain windows per depth, round-robin across " + "tools (default 6)") f.add_argument("--drop-eval", action="store_true", help="behavioral eval: does a real tool-calling model call terse.retrieve " "when a dropped field is needed (recall), and leave it alone when " diff --git a/src/terse/fluency.py b/src/terse/fluency.py index ed57b9f..048650b 100644 --- a/src/terse/fluency.py +++ b/src/terse/fluency.py @@ -41,7 +41,8 @@ from . import text_diff from .capture import LONG_TEXT, OTHER, classify_shape, extract_records from .tokenize import count_cl100k -from .transforms import compress, compress_structure, dict_encode, diff_wire, minify, _uniform_dict_list +from .transforms import (compress, compress_structure, dict_encode, diff_wire, + has_terse_marker, minify, _uniform_dict_list) # Loopback hosts where cleartext http is safe (never leaves the machine), so a Bearer # key over http to one of these is fine — a local LiteLLM/CCR gateway is a common setup. @@ -249,13 +250,54 @@ def _nested_questions(obj: Any) -> list[Question]: return qs +def _flat_record_questions(obj: Any) -> list[Question]: + """Questions for a SINGLE flat record — a dict of mostly scalar fields (a search + hit, a status receipt, one KB row). These payloads tabularize to nothing, but they + are a real diff surface: consecutive single-record results diff via the keys shape, + and before this generator the soak/diff evals were blind to every such chain. + Same contract as the other sources: deterministic, programmatically checkable, + fluency-local (the codec/tabularizer rules are untouched).""" + if not isinstance(obj, dict) or has_terse_marker(obj): + return [] + # int/float lookups are unambiguous; strings must be short, non-empty scalars — + # an empty expected would be indistinguishable from _safe_ask's empty-string + # error return (same exclusion the text-diff questions apply). + lookable = { + k: v for k, v in obj.items() + if (isinstance(v, (int, float)) and not isinstance(v, bool)) + or (isinstance(v, str) and 0 < len(v) <= 60) + } + if len(lookable) < 3: + return [] + qs: list[Question] = [Question( + "keys-count", "count", "flat-record", + "How many top-level keys does the object have?", + "Reply with only the integer count.", + len(obj), + )] + keys = sorted(lookable) + num_key = next((k for k in keys + if isinstance(lookable[k], (int, float))), None) + str_key = next((k for k in keys if isinstance(lookable[k], str)), None) + for i, k in enumerate(x for x in (num_key, str_key) if x is not None): + qs.append(Question( + f"field-{i}", "lookup", "flat-record", + f"What is the value of the top-level field {k!r}?", + "Reply with only the value, with no quotes and no extra words.", + lookable[k], + )) + return qs + + def gen_questions(obj: Any) -> list[Question]: """Generate deterministic, programmatically-checkable questions for a payload. Uniform record-shaped payloads (what terse tabularizes) yield questions directly; payloads whose records the strict extractor skips (structure's dict-map of non-uniform - symbol lists) fall through to `_nested_questions` (#71); everything else returns []. - Selection is fully deterministic so a re-run is reproducible. + symbol lists) fall through to `_nested_questions` (#71); a SINGLE flat record (a + search hit, a receipt) falls through to `_flat_record_questions` — the keys-diff + surface the chain soak needs; everything else returns []. Selection is fully + deterministic so a re-run is reproducible. """ # Structure-shaped payloads (a dict-map of records) need GROUP-SCOPED questions; prefer # them over the uniform extractor, which would otherwise emit an unscoped, ambiguous @@ -265,7 +307,7 @@ def gen_questions(obj: Any) -> list[Question]: return nested records = extract_records(obj) if not records: - return [] + return _flat_record_questions(obj) cols = list(records[0].keys()) n = len(records) aliased = _aliased_strings(obj) @@ -452,8 +494,8 @@ def run_fluency(envelopes: list[dict], answerers: dict[str, Answerer], """Run the eval for each named answerer over every record-shaped payload. Returns {model_name: [scored_row, ...]} where each row carries tool/sha plus the - per-form success counts and trial count. Non-record payloads are skipped (they - generate no questions — terse does not transform them). + per-form success counts and trial count. Payloads that generate no questions + (no record list, no nested dict-map, not a flat record) are skipped. """ results: dict[str, list[dict]] = {} for name, fn in answerers.items(): @@ -552,6 +594,134 @@ def run_diff_fluency(envelopes: list[dict], answerers: dict[str, Answerer], return _aggregate_by_model(pairs, answerers, trials, run_diff_payload) +# --------------------------------------------------------------------------- # +# Diff-chain soak — the DEPTH dimension run_diff_fluency can't see (#8/#20 follow-up). +# run_diff_payload tests one hop (full + 1 diff). In production a model reads up to +# `diff_keyframe_interval` (default 5) consecutive diffs off one full anchor before +# the proxy re-anchors — so the open question before any default-flip is whether +# comprehension DRIFTS with chain depth. These functions build real depth-k windows +# from the corpus (chronological order, #67) and score the same final-state questions +# at every depth, so the report can show accuracy as a function of depth. +# --------------------------------------------------------------------------- # +def build_chain_windows(envelopes: list[dict], max_depth: int = 5, + per_depth_cap: int = 6) -> list[tuple]: + """Depth-k windows (k = 1..max_depth) of consecutive same-tool payloads where + EVERY hop admits a lossless diff and the final payload generates questions — + the exact between-keyframe chains `proxy --diff` emits. Envelope order is + preserved (load_corpus replays chronologically, #67), so windows are real + session sequences, not synthetic evolutions. + + Sampling per depth is round-robin across tools (one window per tool per pass, + starts spread along each tool's maximal diffable run) until `per_depth_cap` — + tool diversity beats stacking one chatty tool's windows. Returns + [(tool, final_sha, depth, [obj_0 .. obj_k])].""" + runs: dict[str, list[list[tuple]]] = {} # tool -> maximal runs of (sha, obj) + by_tool: dict[str, list[dict]] = {} + for env in envelopes: + by_tool.setdefault(env["tool"], []).append(env) + for tool, envs in by_tool.items(): + seq: list[tuple] = [] + for e in envs: + try: + seq.append((e.get("sha", "?"), json.loads(e["raw"]))) + except (json.JSONDecodeError, TypeError): + seq.append((e.get("sha", "?"), None)) + cur: list[tuple] = [] + for (psha, p), (csha, c) in zip(seq, seq[1:]): + ok = (p is not None and c is not None + and diff_wire(p, c, tool) is not None) + if ok: + if not cur: + cur = [(psha, p)] + cur.append((csha, c)) + elif cur: + runs.setdefault(tool, []).append(cur) + cur = [] + if cur: + runs.setdefault(tool, []).append(cur) + + windows: list[tuple] = [] + for depth in range(1, max_depth + 1): + # every eligible start per tool, spread over its runs; then round-robin + candidates: dict[str, list[tuple]] = {} + for tool, tool_runs in sorted(runs.items()): + starts: list[tuple] = [] + for run in tool_runs: + for i in range(0, len(run) - depth, depth): # non-overlapping + window = run[i:i + depth + 1] + if gen_questions(window[-1][1]): + starts.append(window) + if starts: + candidates[tool] = starts + picked = 0 + idx = 0 + while picked < per_depth_cap and candidates: + for tool in sorted(list(candidates)): + starts = candidates[tool] + if idx >= len(starts): + del candidates[tool] + continue + window = starts[idx] + windows.append((tool, window[-1][0], depth, + [obj for _, obj in window])) + picked += 1 + if picked >= per_depth_cap: + break + idx += 1 + return windows + + +def run_chain_payload(objs: list, answerer: Answerer, tool: str = "", + trials: int = 1) -> list[dict]: + """run_diff_payload generalized to a depth-N chain: the same questions about the + FINAL state, control = full-terse of the final result, form = the base full-terse + plus every intermediate diff wire in order — exactly the context a model has + accumulated after N consecutive diffs, with no system primer (production + condition). [] if any hop stops admitting a lossless diff or no questions.""" + questions = gen_questions(objs[-1]) + if not questions: + return [] + wires: list[str] = [] + for prev, curr in zip(objs, objs[1:]): + wire = diff_wire(prev, curr, tool) + if wire is None: + return [] + wires.append(wire) + curr_terse = compress(objs[-1]) + chain_data = f"PREVIOUS RESULT:\n{compress(objs[0])}" + "".join( + f"\n\nUPDATE (diff against the result above, applied in order):\n{w}" + for w in wires) + out: list[dict] = [] + for q in questions: + full_u = _user_prompt(q.prompt, q.instruction, curr_terse) + chain_u = _user_prompt(q.prompt, q.instruction, chain_data) + out.append({ + "qid": q.qid, "qtype": q.qtype, "transform": q.transform, "trials": trials, + "depth": len(wires), + "terse_ok": _ask_n(answerer, "", full_u, q.qtype, q.expected, trials), + "diff_ok": _ask_n(answerer, "", chain_u, q.qtype, q.expected, trials), + }) + return out + + +def run_diff_soak(envelopes: list[dict], answerers: dict[str, Answerer], + trials: int = 1, max_depth: int = 5, + per_depth_cap: int = 6) -> dict: + """Score every answerer over the same depth-1..max_depth chain windows. + Returns {model: [row,...]} where each row also carries `depth` — the report + aggregates by it to show comprehension as a function of chain depth.""" + windows = build_chain_windows(envelopes, max_depth=max_depth, + per_depth_cap=per_depth_cap) + results: dict[str, list[dict]] = {} + for name, fn in answerers.items(): + rows: list[dict] = [] + for tool, sha, _depth, objs in windows: + for row in run_chain_payload(objs, fn, tool, trials=trials): + rows.append({"tool": tool, "sha": sha, **row}) + results[name] = rows + return results + + # --------------------------------------------------------------------------- # # Text-diff fluency — the text-payload analogue of the diff eval above. # diff --git a/src/terse/report.py b/src/terse/report.py index 321eda4..94ace2f 100644 --- a/src/terse/report.py +++ b/src/terse/report.py @@ -673,6 +673,97 @@ def build_text_diff_report(results: dict) -> str: ) +def build_diff_soak_report(results: dict) -> str: + """Render the diff-chain soak: does comprehension DRIFT as a model reads deeper + chains of consecutive diffs off one full anchor (#8/#20 follow-up)? + + `results` is {model: [row,...]} from fluency.run_diff_soak; rows carry the + run_diff_payload shape plus `depth` (how many diffs were chained). The overall + verdict gates on the worst model across ALL depths (principle #24); the by-depth + table is the drift signal itself — production caps chains at the keyframe + interval (default 5), so the deepest tested depth is the deployed worst case.""" + out: list[str] = ["# terse diff-chain soak", ""] + out += [ + "Does a model answer questions about the LATEST state as accurately from", + "(one full result + k consecutive diffs, applied in order) as from the full", + "current result — and does accuracy drift as k grows? Chains are real", + "consecutive same-tool corpus payloads (chronological), exactly what", + "`proxy --diff` emits between keyframes. No system primer (production", + "condition).", "", + ] + if not results or not any(results.values()): + out += [ + "No model answers, or no same-tool diffable RUNS in the corpus. Capture a", + "tool 3+ times with small changes between calls, then re-run", + "`terse fluency --diff-soak`.", "", + ] + return "\n".join(out) + + trials = max((r.get("trials", 1) for rows in results.values() for r in rows), default=1) + depths = sorted({r["depth"] for rows in results.values() for r in rows}) + out += [ + "## Accuracy by chain depth", + "", + f"Trials per question: **{trials}**. `±` is the 95% half-width of a pooled " + "binomial bound. depth = diffs chained after the full anchor.", + "", + "| Model | depth | q | full-terse | chain | gap |", + "|---|---|---|---|---|---|", + ] + for model, rows in results.items(): + for depth in depths: + drows = [r for r in rows if r["depth"] == depth] + if not drows: + continue + facc, fse = _form_stats(drows, "terse_ok") + dacc, dse = _form_stats(drows, "diff_ok") + out.append(f"| `{model}` | {depth} | {len(drows)} " + f"| {facc:.0%} ±{_ci(fse) * 100:.0f} " + f"| {dacc:.0%} ±{_ci(dse) * 100:.0f} | {dacc - facc:+.0%} |") + out.append("") + + gap_rows: dict[str, tuple[float, float, float, float]] = {} + for model, rows in results.items(): + if not rows: + continue + facc, fse = _form_stats(rows, "terse_ok") + dacc, dse = _form_stats(rows, "diff_ok") + gap_rows[model] = (dacc, dse, facc, fse) + + out += ["## Verdict", ""] + worst = _worst_case_gap(gap_rows) + if worst: + out.append(_format_worst_case_line(worst, _GAP_TOLERANCE, "chain-form", + "full-terse")) + # The soak-specific signal on top of the overall gate: the worst model's gap + # at the DEEPEST depth, since a clean average can hide a depth-correlated slide. + deep = depths[-1] if depths else 0 + deep_rows = {m: r for m, r in + ((m, [x for x in rs if x["depth"] == deep]) + for m, rs in results.items()) if r} + deep_gaps = {} + for m, rs in deep_rows.items(): + facc, fse = _form_stats(rs, "terse_ok") + dacc, dse = _form_stats(rs, "diff_ok") + deep_gaps[m] = (dacc, dse, facc, fse) + deepest = _worst_case_gap(deep_gaps) + if deepest: + out.append(f"- At the deepest tested depth ({deep}): worst model " + f"`{deepest.model}` chain {deepest.form_acc:.0%} vs full " + f"{deepest.control_acc:.0%} (gap {deepest.gap:+.0%} " + f"±{deepest.gap_ci * 100:.0f} pts). " + f"**{'PASS' if deepest.passed else 'FAIL'}** at " + f"{_GAP_TOLERANCE:.0%} tolerance.") + if worst.passed and (deepest is None or deepest.passed): + out.append("- No depth-correlated comprehension drift within tolerance — " + "chained diffs up to the tested depth read as well as fulls.") + else: + out.append("- Comprehension drifts beyond tolerance somewhere in the chain — " + "keep the keyframe interval at or below the deepest PASSING depth.") + out.append("") + return "\n".join(out) + + def build_dropeval_report(results: dict) -> str: """Render the drop-to-retrieve behavioral eval: does a real tool-calling model call `terse.retrieve` when a dropped field is needed (recall), and leave it alone when it diff --git a/tests/test_diff_soak.py b/tests/test_diff_soak.py new file mode 100644 index 0000000..bec3d34 --- /dev/null +++ b/tests/test_diff_soak.py @@ -0,0 +1,267 @@ +"""Long-chain drift soak for the cross-call diff tier (#8/#20 follow-up). + +The per-hop guarantee is proven at encode time (a diff is accepted only if it +reconstructs curr exactly) and pinned by the test_proxy/test_diff unit tests. What +none of those cover is DRIFT: a client that reconstructs hop N against its own hop +N-1 state, hundreds of hops deep, across interleaved tools, keyframes, shape flips, +error interludes, and a mid-session reconnect. This soak drives the real Interceptor +over a seeded evolving workload while an independent `ModelView` mirrors exactly +what the model must do with each emitted text — and asserts view == the raw +downstream payload at EVERY hop, so any base-bookkeeping bug (wrong base, missed +eviction, stale keyframe counter) surfaces as a first-divergence step index.""" + +from __future__ import annotations + +import json +import random + +from terse import text_diff, transforms +from terse.policy import Policy, Rule +from terse.proxy import Interceptor + +TIERS = ("minify", "tabularize", "dictionary") + + +def _req(mid, name): + return json.dumps({"jsonrpc": "2.0", "id": mid, "method": "tools/call", + "params": {"name": name}}) + + +def _init_req(mid): + return json.dumps({"jsonrpc": "2.0", "id": mid, "method": "initialize", + "params": {"protocolVersion": "2024-11-05"}}) + + +def _result_msg(mid, text): + return json.dumps({"jsonrpc": "2.0", "id": mid, + "result": {"content": [{"type": "text", "text": text}]}}) + + +class ModelView: + """The client side of the diff protocol: per tool, keep the last reconstructed + value and apply each emitted text per the documented terse semantics (diff → + apply to the previous same-tool result; anything else → a self-contained full). + Deliberately implemented over the public decode functions only — it must never + peek at the Interceptor's internal state, or the soak proves nothing.""" + + def __init__(self): + self.state: dict[str, object] = {} # tool -> last full JSON value + self.text_state: dict[str, str] = {} # tool -> last full raw text + + def reset(self): + """A client reconnect: the context (and every prior result) is gone.""" + self.state.clear() + self.text_state.clear() + + def see(self, tool: str, emitted: str): + """Reconstruct the full raw value this emission denotes, updating state. + Returns the value (JSON) or text (str) the model would now believe in.""" + try: + obj = json.loads(emitted) + except (json.JSONDecodeError, ValueError): + obj = None + if isinstance(obj, dict) and obj.get(transforms.DIFF_MARKER) == 1: + curr = transforms.diff_decode(self.state[tool], obj) + self.state[tool] = curr + return curr + if isinstance(obj, dict) and obj.get(text_diff.DIFF_MARKER) == 1: + curr = text_diff.text_diff_decode(self.text_state[tool], obj) + self.text_state[tool] = curr + return curr + if obj is not None: + curr = transforms.decompress(emitted) + self.state[tool] = curr + return curr + self.text_state[tool] = emitted # non-JSON: the raw text IS the full form + return emitted + + +class Soak: + """Drives one Interceptor + one ModelView and asserts equality at every hop.""" + + def __init__(self, interval: int): + pol = Policy(rules=[Rule("*", TIERS)], diff=True, + diff_keyframe_interval=interval) + self.inter = Interceptor(pol) + self.view = ModelView() + self.mid = 0 + self.consec: dict[str, int] = {} # tool -> consecutive emitted diffs + self.max_consec: dict[str, int] = {} + self.diffs = 0 # total diff emissions observed + + def call(self, tool: str, payload) -> str: + """One tools/call round trip. `payload` is a JSON value or a raw str. + Asserts the ModelView reconstructs it exactly. Returns the emitted text.""" + self.mid += 1 + raw = payload if isinstance(payload, str) else json.dumps(payload, indent=2) + self.inter.note_request(_req(self.mid, tool)) + out = self.inter.transform_response(_result_msg(self.mid, raw)) + emitted = json.loads(out)["result"]["content"][0]["text"] + got = self.view.see(tool, emitted) + want = payload + assert got == want, (f"drift at step {self.mid} tool={tool}: " + f"reconstruction diverged from the raw payload") + is_diff = (transforms.DIFF_MARKER in emitted + or text_diff.DIFF_MARKER in emitted) + self.consec[tool] = self.consec.get(tool, 0) + 1 if is_diff else 0 + self.max_consec[tool] = max(self.max_consec.get(tool, 0), self.consec[tool]) + self.diffs += is_diff + return emitted + + def reconnect(self): + """Client re-handshake (#20): both sides drop everything.""" + self.mid += 1 + self.inter.note_request(_init_req(self.mid)) + self.view.reset() + self.consec.clear() + + +# --------------------------------------------------------------------- workloads +class RecordWorld: + """An evolving uniform record list under {"result": [...]} — the agent-loop + shape the row diff targets. Mutations keep the id column unique and append new + rows at the end (the representable pattern), with occasional reorders that + force the diff to bow out to a full form (which must also reconstruct).""" + + def __init__(self, rng: random.Random, n: int = 30, reorders: bool = True): + self.rng = rng + self.next_id = n + self.reorders = reorders + self.rows = [self._row(i) for i in range(n)] + + def _row(self, i): + return {"id": i, "status": "active", "score": i % 11, + "url": "https://x.example/api/items"} + + def step(self): + r = self.rng.random() + if r < 0.55 and self.rows: # mutate one field + self.rng.choice(self.rows)["status"] = self.rng.choice( + ["active", "closed", "stale", "merged"]) + elif r < 0.75: # append a new row + self.rows.append(self._row(self.next_id)) + self.next_id += 1 + elif r < 0.90 and len(self.rows) > 3: # delete one row + del self.rows[self.rng.randrange(len(self.rows))] + elif r < 0.95 or not self.reorders: # no-op: identical payload + pass + else: # reorder: diff bows out + self.rng.shuffle(self.rows) + return {"result": [dict(row) for row in self.rows]} + + +class MapWorld: + """A structure-like dict-map (path -> {symbols: [...]}) with non-uniform inner + records — the nested shape #71/#72 was about. Exercises the coarse keys diff.""" + + def __init__(self, rng: random.Random): + self.rng = rng + self.n = 0 + self.files = {f"src/m{i}.py": self._syms(i) for i in range(6)} + + def _syms(self, i): + self.n += 1 + return {"symbols": [{"name": f"fn_{i}_{j}", "kind": "function", + "line": 10 * j + 1, "hash": f"h{self.n}{j}"} + for j in range(4)] + + [{"name": "os", "kind": "import"}]} + + def step(self): + r = self.rng.random() + keys = list(self.files) + if r < 0.6: # edit one file's symbols + k = self.rng.choice(keys) + self.files[k] = self._syms(len(keys)) + elif r < 0.8: # new file + self.files[f"src/m{self.n}.py"] = self._syms(self.n) + elif len(keys) > 2: # file removed + del self.files[self.rng.choice(keys)] + return {"files": {k: json.loads(json.dumps(v)) + for k, v in self.files.items()}} + + +class LogWorld: + """A growing, occasionally edited plain-text log — the CDC text-diff domain.""" + + def __init__(self, rng: random.Random): + self.rng = rng + self.lines = [f"[{i:04d}] worker heartbeat ok, queue_depth={i % 7}" + for i in range(120)] + + def step(self): + r = self.rng.random() + if r < 0.6: # append (tail -f shape) + n = len(self.lines) + self.lines.append(f"[{n:04d}] worker heartbeat ok, queue_depth={n % 7}") + elif r < 0.85 and self.lines: # edit one line in place + i = self.rng.randrange(len(self.lines)) + self.lines[i] = f"[{i:04d}] RETRY backoff=2 attempt={self.rng.randrange(9)}" + else: # truncate (log rotation) + self.lines = self.lines[-60:] + return "\n".join(self.lines) + + +def test_soak_400_steps_interleaved_tools_no_drift(): + # The headline soak: 400 hops, three tools interleaved, production keyframe + # interval (5), plus error interludes and a mid-soak client reconnect. Every + # hop asserts the client reconstruction equals the raw payload exactly. + rng = random.Random(42) + soak = Soak(interval=5) + worlds = {"gh.items": RecordWorld(rng), "repo.map": MapWorld(rng), + "fs.log": LogWorld(rng)} + tools = list(worlds) + + for step in range(400): + tool = tools[step % 3] if rng.random() < 0.7 else rng.choice(tools) + soak.call(tool, worlds[tool].step()) + + if step in (150, 290) and tool == "gh.items": + # a non-JSON interlude (upstream error) on a JSON tool: becomes the + # visible "previous result", so the JSON base must evict (#8) … + soak.call(tool, "upstream error: rate limited") + # … and the next JSON result must re-anchor as a full, never a diff + # against the now-invisible pre-error base. + emitted = soak.call(tool, worlds[tool].step()) + assert transforms.DIFF_MARKER not in emitted + + if step == 200: + soak.reconnect() # client context wiped (#20) + for t in tools: + # first post-reconnect emission per tool must be self-contained + emitted = soak.call(t, worlds[t].step()) + assert (transforms.DIFF_MARKER not in emitted + and text_diff.DIFF_MARKER not in emitted) + + # the soak must have actually exercised long diff chains, not degenerated to + # fulls — and the keyframe bound must have held for every tool throughout. + assert soak.diffs > 150, f"only {soak.diffs} diffs emitted — workload too weak" + assert soak.max_consec and max(soak.max_consec.values()) == 5 + assert all(v <= 5 for v in soak.max_consec.values()) + + +def test_soak_unbounded_chain_interval_zero_exact_at_depth_300(): + # keyframe interval 0 = never re-anchor: a single tool's chain runs hundreds of + # diffs deep off ONE full result. Reconstruction must stay exact at every depth + # — this is the pure accumulation-drift case the keyframe normally bounds. + rng = random.Random(7) + soak = Soak(interval=0) + world = RecordWorld(rng, n=40, reorders=False) # every step representable + soak.call("gh.items", {"result": [dict(r) for r in world.rows]}) # the anchor + for _ in range(300): + soak.call("gh.items", world.step()) + # small seeded mutations diff nearly every hop; prove real chain depth was hit + assert soak.diffs >= 250 + assert soak.max_consec["gh.items"] >= 200 + + +def test_soak_text_only_long_chain(): + # The CDC text path keeps its own base/keyframe state (#25); soak it separately + # at the production interval so a text-side bookkeeping bug can't hide behind + # the JSON assertions of the interleaved soak. + rng = random.Random(99) + soak = Soak(interval=5) + world = LogWorld(rng) + for _ in range(200): + soak.call("fs.log", world.step()) + assert soak.diffs > 100 + assert max(soak.max_consec.values()) <= 5 diff --git a/tests/test_fluency.py b/tests/test_fluency.py index d06db2d..9eb41e9 100644 --- a/tests/test_fluency.py +++ b/tests/test_fluency.py @@ -530,3 +530,117 @@ def test_openai_answerer_allows_https_with_key(): def test_openai_answerer_allows_http_without_key(): # no key set -> nothing secret to leak, so plain http is permitted assert callable(fluency.openai_answerer("http://api.example.com/v1", "", "gpt-x")) + + +# --- diff-chain soak (#8/#20 follow-up): depth-k windows + chained-diff form --- + +def _soak_envs(tool="gh.items", n=8, start_id=0): + """n consecutive envelopes for one tool, each a small mutation of the last — + every hop admits a lossless row diff (update-in-place, append at the end).""" + envs = [] + rows = [{"id": i, "status": "active", "score": i % 7} for i in range(start_id, start_id + 20)] + for k in range(n): + if k: + rows[k % len(rows)]["status"] = f"state-{k}" + rows.append({"id": start_id + 20 + k, "status": "new", "score": k}) + envs.append({"tool": tool, "sha": f"s{k:02d}", + "raw": json.dumps({"result": [dict(r) for r in rows]})}) + return envs + + +def test_build_chain_windows_yields_every_depth_and_valid_hops(): + from terse.transforms import diff_wire + + windows = fluency.build_chain_windows(_soak_envs(n=8), max_depth=3, per_depth_cap=4) + depths = {d for _, _, d, _ in windows} + assert depths == {1, 2, 3} + for tool, sha, depth, objs in windows: + assert len(objs) == depth + 1 + for prev, curr in zip(objs, objs[1:]): + assert diff_wire(prev, curr, tool) is not None # every hop truly chains + assert fluency.gen_questions(objs[-1]) # final state is askable + + +def test_build_chain_windows_never_spans_a_diff_break(): + # an unrelated payload mid-run splits it: no window may bridge the break, because + # in production that hop would have re-anchored as a full, ending the chain. + envs = _soak_envs(n=4) + envs.insert(2, {"tool": "gh.items", "sha": "sXX", + "raw": json.dumps({"totally": "different", "shape": [1, 2, 3]})}) + windows = fluency.build_chain_windows(envs, max_depth=3, per_depth_cap=8) + from terse.transforms import diff_wire + for tool, _sha, _depth, objs in windows: + for prev, curr in zip(objs, objs[1:]): + assert diff_wire(prev, curr, tool) is not None + + +def test_run_chain_payload_context_carries_one_full_plus_depth_wires(): + envs = _soak_envs(n=4) + objs = [json.loads(e["raw"]) for e in envs] + seen: list[str] = [] + + def spy(system, user): + seen.append(user) + return "" + + rows = fluency.run_chain_payload(objs, spy, tool="gh.items", trials=1) + assert rows and all(r["depth"] == 3 for r in rows) + chain_prompts = [u for u in seen if "UPDATE (diff against the result above" in u] + assert chain_prompts # the chain form was asked + assert all(u.count("UPDATE (diff against the result above") == 3 + for u in chain_prompts) # exactly depth wires + assert all(u.count("PREVIOUS RESULT:") == 1 for u in chain_prompts) # one anchor + + +def test_run_chain_payload_empty_when_a_hop_stops_diffing(): + objs = [json.loads(e["raw"]) for e in _soak_envs(n=3)] + objs.append({"totally": "different", "shape": [1, 2, 3]}) # last hop can't diff + assert fluency.run_chain_payload(objs, lambda s, u: "", tool="gh.items") == [] + + +def test_run_diff_soak_rows_carry_depth_per_model(): + results = fluency.run_diff_soak(_soak_envs(n=8), {"m1": lambda s, u: ""}, + trials=1, max_depth=3, per_depth_cap=2) + assert set(results) == {"m1"} + assert {r["depth"] for r in results["m1"]} == {1, 2, 3} + assert all({"tool", "sha", "qid", "terse_ok", "diff_ok"} <= set(r) + for r in results["m1"]) + + +def test_build_diff_soak_report_by_depth_table_and_verdicts(): + from terse.report import build_diff_soak_report + + rows = [{"tool": "gh.items", "sha": "s", "qid": "count", "qtype": "count", + "transform": "tabularize", "trials": 1, "depth": d, + "terse_ok": 1, "diff_ok": 1} for d in (1, 2, 3) for _ in range(4)] + report = build_diff_soak_report({"m1": rows}) + assert "## Accuracy by chain depth" in report + assert "At the deepest tested depth (3)" in report + assert "**PASS**" in report and "No depth-correlated comprehension drift" in report + + # a depth-correlated slide beyond tolerance must FAIL the deepest-depth gate + bad = [dict(r, diff_ok=0 if r["depth"] == 3 else 1) for r in rows] + report = build_diff_soak_report({"m1": bad}) + assert "**FAIL**" in report + + # empty results explain how to get soakable data + assert "diffable RUNS" in build_diff_soak_report({}) + + +def test_flat_record_questions_cover_single_record_payloads(): + # a single flat record (search hit, status receipt, one KB row): keys-count plus + # deterministic numeric/string lookups — the diff surface the soak was blind to. + obj = {"type": "note", "id": 42, "title": "diff soak", "snippet": "x" * 200, + "score": 0.87, "nested": {"skip": True}} + qs = fluency.gen_questions(obj) + by_qid = {q.qid: q for q in qs} + assert by_qid["keys-count"].expected == 6 # counts ALL keys + assert by_qid["field-0"].expected == 42 # first numeric by sorted key + assert by_qid["field-1"].expected == "diff soak" # long/nested values not asked + assert all(q.transform == "flat-record" for q in qs) + + +def test_flat_record_questions_stay_silent_when_underqualified(): + assert fluency.gen_questions({"just": "an object", "two": 2}) == [] # <3 lookable + assert fluency.gen_questions({"__terse_dict__": 1, "a": 1, "b": 2, "c": 3}) == [] + assert fluency.gen_questions({"a": "", "b": "y" * 999, "c": None, "d": True}) == []