#!/usr/bin/env python3 """Tests for the change_context slice's decision detection (FLY-1482, FLY-1505). Run: python3 test_change_context.py Covers the two things that made `decisions` empty on every real issue: * the relay prepends an attribution line — `**@Matt Elsey** (via FlyDocs)` — to every comment it writes on a person's behalf, and `_lead_line` returned *that* line, which matches no marker; * the transition template leads with the canonical status word (`REVIEW — …`, `IMPLEMENTING — …`), and the marker tuple did not carry those words. and the third that made it read a pasted code block as prose: a comment that opens with a fence had the fence's first line for its lead (FLY-1505). The behaviour here has a twin on the remote tier — `DECISION_MARKERS`, `ATTRIBUTION_LINE`, `leadLine` and `decisionMarker` in `convex/lib/mcpChangeContext.ts`, `stripCodeBlocks` and `CODE_OMITTED_MARKER` in `convex/lib/scmPullRequestSummary.ts` — and the two must agree, or the same comment yields a different marker depending on which tier read it. The fence fixtures below are the remote tier's own. Hermetic by construction: `change_context` is a pure module. Nothing here opens a socket, reads a file, or builds a client. """ import json import sys from pathlib import Path SCRIPT_DIR = Path(__file__).parent.resolve() sys.path.insert(0, str(SCRIPT_DIR)) import change_context as cc # noqa: E402 passed = 0 failed = 0 def test(name: str): """Decorator to register and run a test.""" def decorator(fn): global passed, failed try: fn() print(f" PASS: {name}") passed += 1 except AssertionError as e: print(f" FAIL: {name} — {e}") failed += 1 except Exception as e: # noqa: BLE001 — a broken test is a failed test print(f" ERROR: {name} — {type(e).__name__}: {e}") failed += 1 return fn return decorator # The comment the relay writes for `issue_transition` REF REVIEW, verbatim. RELAY_REVIEW = ( "**@Matt Elsey** (via FlyDocs)\n" "\n" "REVIEW — Implementation complete — PR #326 opened; full suite green." ) print("\n--- The attribution line is not the lead line ---") @test("a relay-written transition comment reads the status word, not the author") def _(): assert cc._lead_line(RELAY_REVIEW) == ( "REVIEW — Implementation complete — PR #326 opened; full suite green." ), cc._lead_line(RELAY_REVIEW) assert cc.decision_marker(RELAY_REVIEW) == "review", \ cc.decision_marker(RELAY_REVIEW) @test("a comment that is only an attribution line carries no decision") def _(): body = "**@Matt Elsey** (via FlyDocs)" assert cc._lead_line(body) == "", cc._lead_line(body) assert cc.decision_marker(body) is None, cc.decision_marker(body) @test("the attribution match is not tied to one name or one product") def _(): body = "**@kyle.dev** (via FlyDocs Relay)\n\nBlocked — waiting on the key." assert cc._lead_line(body) == "Blocked — waiting on the key.", \ cc._lead_line(body) assert cc.decision_marker(body) == "blocked", cc.decision_marker(body) @test("a line that merely mentions an author is still the lead line") def _(): # No parenthetical `(via …)`, so it is prose, not attribution. body = "@Matt Elsey asked for this — decision: ship it." assert cc._lead_line(body).startswith("@Matt Elsey asked"), cc._lead_line(body) assert cc.decision_marker(body) is None, cc.decision_marker(body) print("\n--- Canonical status words are decisions ---") @test("IMPLEMENTING — Starting implementation") def _(): body = "IMPLEMENTING — Starting implementation" assert cc.decision_marker(body) == "implementing", cc.decision_marker(body) @test("the other status words the transition template writes") def _(): cases = { "READY — refined and estimated": "ready", "TESTING — deployed to staging": "testing", "COMPLETE — all acceptance criteria verified": "complete", "CANCELED — superseded by FLY-1500": "canceled", "CANCELLED — superseded by FLY-1500": "cancelled", "DUPLICATE — of FLY-1300": "duplicate", } for body, expected in cases.items(): assert cc.decision_marker(body) == expected, \ f"{body!r} → {cc.decision_marker(body)!r}, expected {expected!r}" @test("bookkeeping transitions are still not decisions") def _(): for body in ( "Captured from discussion", "Activated", "Progress: half of the acceptance criteria are ticked", ): assert cc.decision_marker(body) is None, \ f"{body!r} → {cc.decision_marker(body)!r}" @test("the same bookkeeping words behind an attribution line stay out") def _(): body = "**@Matt Elsey** (via FlyDocs)\n\nProgress: still working through it." assert cc.decision_marker(body) is None, cc.decision_marker(body) print("\n--- Marker ordering ---") @test("'ready for review' wins over 'ready' — longest prefix first") def _(): body = "Ready for review — 7/7 criteria complete." assert cc.decision_marker(body) == "ready for review", \ cc.decision_marker(body) assert cc.decision_marker("Ready — refined") == "ready", \ cc.decision_marker("Ready — refined") @test("every marker sharing a prefix with a shorter one is listed first") def _(): # The rule `decision_marker` depends on, asserted over the whole tuple # rather than the one pair we know about. markers = cc.DECISION_MARKERS for i, marker in enumerate(markers): for other in markers[i + 1:]: assert not other.startswith(marker) or other == marker, ( f"{other!r} starts with the earlier {marker!r} and can never " "be matched — list the longer marker first" ) @test("the tuple carries the tier-parity contents") def _(): # Mirrors DECISION_MARKERS in convex/lib/mcpChangeContext.ts, in order. assert cc.DECISION_MARKERS == ( "decision", "decided", "triaged", "refined", "blocked", "unblocked", "ready for review", "code review", "qe approved", "qe issues found", "qe partial", "qe:", "closed", "archived", "root cause", "rejected", "reverted", "trade-off", "tradeoff", "ready", "implementing", "review", "testing", "complete", "canceled", "cancelled", "duplicate", ), cc.DECISION_MARKERS @test("bookkeeping words are absent from the tuple") def _(): for word in ("captured", "activated", "progress"): assert word not in cc.DECISION_MARKERS, word print("\n--- The slice end to end ---") def _issue(comments: list[dict]) -> dict: """The shape `issues.py get --fields full` returns, cut to what is used.""" return { "identifier": "FLY-1300", "id": "e1f2a3b4", "title": "Remote MCP surface", "status": "In Review", "description": "The relay-side surface.", "createdAt": "2026-08-20T09:00:00.000Z", "updatedAt": "2026-08-29T18:00:00.000Z", "comments": comments, "links": [], } @test("a relay-written REVIEW comment becomes a cited decision claim") def _(): issue = _issue([{ "id": "c-100", "body": RELAY_REVIEW, "user": "Matt Elsey", "createdAt": "2026-08-29T17:45:00.000Z", }]) slice_, warnings = cc.build_shipped_slice( issue, target="FLY-1300", observed_at="2026-08-30T10:00:00.000Z") decisions = [c for c in slice_["claims"] if c["kind"] == "decision"] assert len(decisions) == 1, slice_["claims"] decision = decisions[0] assert decision["marker"] == "review", decision assert decision["statement"] == ( "REVIEW — Implementation complete — PR #326 opened; full suite green." ), decision assert decision["derivation"] == cc.RECORDED, decision assert decision["author"] == "Matt Elsey", decision assert "comment:c-100" in decision["source"]["refs"], decision assert "FLY-1300" in decision["source"]["refs"], decision assert decision["source"]["timestamp"] == "2026-08-29T17:45:00.000Z", decision assert decision["source"]["timestampKind"] == "recorded", decision assert warnings == [], warnings @test("a bookkeeping comment produces no decision claim") def _(): issue = _issue([{ "id": "c-101", "body": "**@Matt Elsey** (via FlyDocs)\n\nActivated — picked this up.", "user": "Matt Elsey", "createdAt": "2026-08-29T17:00:00.000Z", }]) slice_, _warnings = cc.build_shipped_slice( issue, target="FLY-1300", observed_at="2026-08-30T10:00:00.000Z") assert [c for c in slice_["claims"] if c["kind"] == "decision"] == [], \ slice_["claims"] @test("decisions come back newest first") def _(): issue = _issue([ { "id": "c-1", "body": "**@Matt Elsey** (via FlyDocs)\n\nIMPLEMENTING — Starting.", "createdAt": "2026-08-28T09:00:00.000Z", }, { "id": "c-2", "body": RELAY_REVIEW, "createdAt": "2026-08-29T17:45:00.000Z", }, ]) slice_, _warnings = cc.build_shipped_slice( issue, target="FLY-1300", observed_at="2026-08-30T10:00:00.000Z") markers = [c["marker"] for c in slice_["claims"] if c["kind"] == "decision"] assert markers == ["review", "implementing"], markers print("\n--- Fenced code blocks ---") # The decision comment the remote tier's mcpChangeContext.test.ts uses as its # `c4` fixture, verbatim. Both tiers must read the same lead out of it. FENCE_FIRST = ( "```\n" "leaked code in a decision\n" "```\n" "**Decision**: keep rosters static per endpoint." ) @test("the placeholder is the remote tier's, verbatim") def _(): # `CODE_OMITTED_MARKER` in convex/lib/scmPullRequestSummary.ts. The two # tiers skip this exact line when reading a lead. assert cc.CODE_OMITTED_MARKER == "[code omitted]", cc.CODE_OMITTED_MARKER @test("a comment that opens with a fence reads the prose under it") def _(): assert cc._lead_line(FENCE_FIRST) == ( "Decision: keep rosters static per endpoint." ), cc._lead_line(FENCE_FIRST) assert cc.decision_marker(FENCE_FIRST) == "decision", \ cc.decision_marker(FENCE_FIRST) assert "leaked" not in cc._strip_code_blocks(FENCE_FIRST), \ cc._strip_code_blocks(FENCE_FIRST) @test("a marker inside a fence is not a decision") def _(): body = "```\nDecision: not a real one\n```\nJust a status note." assert cc._lead_line(body) == "Just a status note.", cc._lead_line(body) assert cc.decision_marker(body) is None, cc.decision_marker(body) # And a comment that is nothing but the fence carries no lead at all. only = "```\nDecided: not a real one\n```" assert cc._lead_line(only) == "", cc._lead_line(only) assert cc.decision_marker(only) is None, cc.decision_marker(only) @test("an unterminated fence swallows everything after it") def _(): body = "prose\n```js\nlet leaked = true;" assert cc._strip_code_blocks(body) == f"prose\n{cc.CODE_OMITTED_MARKER}", \ cc._strip_code_blocks(body) assert cc._lead_line(body) == "prose", cc._lead_line(body) assert cc.decision_marker("```\nDecided: nope") is None, \ cc.decision_marker("```\nDecided: nope") @test("tilde fences are fences") def _(): body = "a\n~~~\nx = 1\n~~~\nb" assert cc._strip_code_blocks(body) == \ f"a\n{cc.CODE_OMITTED_MARKER}\nb", cc._strip_code_blocks(body) @test("the closing fence must be the opening run, not merely a fence") def _(): # A longer closer, or the other character, leaves the fence open — which # is what the remote tier's backreference does, checked against it. for body in ("a\n```\nx = 1\n`````\nb", "a\n```\nx = 1\n~~~\nb"): assert cc._strip_code_blocks(body) == \ f"a\n{cc.CODE_OMITTED_MARKER}", cc._strip_code_blocks(body) @test("a CRLF body keeps the prose after the fence") def _(): body = "prose\r\n```\r\nconst secret = 1;\r\n```\r\nmore prose" assert cc._strip_code_blocks(body) == \ f"prose\n{cc.CODE_OMITTED_MARKER}\nmore prose", \ cc._strip_code_blocks(body) assert "secret" not in cc._strip_code_blocks(body), \ cc._strip_code_blocks(body) assert cc._lead_line(body) == "prose", cc._lead_line(body) @test("an excerpt carries the placeholder, never the code") def _(): text, truncated = cc._excerpt( "Decided: ship it.\n```\nconst secret = 1;\n```") assert text == f"Decided: ship it.\n{cc.CODE_OMITTED_MARKER}", text assert truncated is False, truncated long_lead = "Decided: " + "x" * (cc.EXCERPT_CHARS + 10) cut, truncated = cc._excerpt(long_lead) assert len(cut) == cc.EXCERPT_CHARS, len(cut) assert truncated is True, truncated @test("a decision comment with a fence cites the prose and leaks no code") def _(): issue = _issue([{ "id": "c-102", "body": FENCE_FIRST, "user": "Matt Elsey", "createdAt": "2026-08-29T18:00:00.000Z", }]) slice_, warnings = cc.build_shipped_slice( issue, target="FLY-1300", observed_at="2026-08-30T10:00:00.000Z") decisions = [c for c in slice_["claims"] if c["kind"] == "decision"] assert len(decisions) == 1, slice_["claims"] assert decisions[0]["statement"] == ( "Decision: keep rosters static per endpoint." ), decisions[0] assert decisions[0]["marker"] == "decision", decisions[0] assert decisions[0]["truncatedText"] is False, decisions[0] assert "leaked" not in json.dumps(slice_), slice_ assert warnings == [], warnings # --------------------------------------------------------------------------- # Summary # --------------------------------------------------------------------------- print(f"\n{'='*50}") print(f"Results: {passed} passed, {failed} failed, {passed + failed} total") if failed > 0: sys.exit(1)