"""The `change_context` slice — issue and PR records, with provenance (FLY-1294). `graph.change_context` answers one question: *what shipped around this, and where is that written down?* The join is issue → the pull requests recorded on it → the issues it links to → the comments that carry a decision, returned as a compact, ordered list of claims that each cite their own source. This module is **pure**. It takes a provider issue body (whatever `issues.py get --fields full` returned) and gives back a slice; it opens no socket, reads no file, imports nothing from the dispatchers, and knows nothing about the bridge envelope. `bridge.py` owns the I/O — it calls the same `cmd_get` the `issue.get` operation calls, so there is one read path, not two — and hands the body here. That split is what makes every rule below testable without a relay, and it is why the provenance guarantee can be a unit test rather than an integration one. ## The provenance contract (remote-mcp-consumption-tier-spec.md §5.5) Every claim carries: * **source refs** — the issue id, and the comment id or PR number the claim was read out of; * **a source timestamp** — the record's own, when the record has one; otherwise `timestampKind: "observed"` and the slice's single `observedAt`, because a record with no timestamp of its own was last known true when it was read; * **a derivation marker** — `recorded` for a fact lifted verbatim out of a record, `derived` for anything this module concluded; * and the slice carries an **explicit truncation flag** whenever it was cut. > A claim that cannot carry a source ref does not appear in the result. That is the hard rule, not a quality target (§5.5, D4): a marketing seat one tier up publishes from these slices, and the failure mode is a capability claim that no record supports. It is enforced twice on purpose — `_claim` refuses to build such a claim, and `enforce_provenance` sweeps the assembled slice before it is returned, dropping anything that got through and *saying so in a warning* rather than quietly shrinking the answer. ## What is deliberately absent Diff summaries and touched modules are in the §4.1 description of this tool and are **not** in this slice, because nothing records them against an issue today: the tracker holds a PR link, not the PR's contents. Inventing them from the link would be exactly the failure §5.5 exists to prevent, so instead the slice names the hole in `gaps`. When a PR ingest lands, the claims appear with real source refs and nothing else here changes. """ import re # --------------------------------------------------------------------------- # Size cap # --------------------------------------------------------------------------- # # The cap is per section rather than one budget for the whole slice. A single # total would let forty PR mentions on a long-running epic push every decision # comment out of the answer, and the caller would have no way to see that it # happened beyond a number it never set. DEFAULT_LIMIT = 5 MIN_LIMIT = 1 MAX_LIMIT = 50 #: The longest excerpt a decision claim quotes. Past this the claim says so. EXCERPT_CHARS = 240 MODES = ("shipped", "impact") #: `FLY-123`, `ENG-4`, `AB1-77` — a provider key, a hyphen, a number. _ISSUE_REF = re.compile(r"^[A-Za-z][A-Za-z0-9_]*-\d+$") def is_issue_ref(target: str) -> bool: """Is `target` an issue reference rather than a module or topic string?""" return bool(_ISSUE_REF.match((target or "").strip())) def normalize_ref(target: str) -> str: """`fly-1294` → `FLY-1294`. Refs are case-insensitive; claims are not.""" return (target or "").strip().upper() def check_limit(value: object) -> str | None: """Validate a caller-supplied `limit`. Returns a message, or None. Returned rather than raised: this module has no error type of its own, and borrowing the bridge's would make a pure function import the boundary it is supposed to be independent of. """ if not isinstance(value, int) or isinstance(value, bool): return "'limit' must be an integer." if value < MIN_LIMIT or value > MAX_LIMIT: return ( f"'limit' must be between {MIN_LIMIT} and {MAX_LIMIT}, got {value}. " "It caps each section of the slice — pull requests, related issues, " "decisions — not the slice as a whole." ) return None # --------------------------------------------------------------------------- # Pull request references # --------------------------------------------------------------------------- # # The host is left open in every pattern: GitHub Enterprise, self-hosted GitLab # and Bitbucket Data Center all serve these paths off a private domain, and a # pattern anchored to github.com would silently find nothing in exactly the # workspaces most likely to have a PR link. `platform` therefore names the URL # *grammar* that matched, which is a property of the string in front of us, # rather than a claim about who is hosting it. _PR_PATTERNS = ( ("github", re.compile( r"https?://[\w.\-]+/([\w.\-]+/[\w.\-]+)/pull/(\d+)")), ("gitlab", re.compile( r"https?://[\w.\-]+/([\w.\-]+(?:/[\w.\-]+)+?)/-/merge_requests/(\d+)")), ("bitbucket", re.compile( r"https?://[\w.\-]+/([\w.\-]+/[\w.\-]+)/pull-requests/(\d+)")), ) #: State words a recorded line may attach to a PR link. Order is not meaningful; #: the first match on the line wins, and the claim says it is derived. _PR_STATES = ("merged", "reverted", "closed", "draft", "open") def _pr_mentions(text: str) -> list[dict]: """Every pull request URL in `text`, with the line it was written on. The line is kept because it is the scope a state word is read from. A comment that says "merged" three paragraphs below an unrelated link is not saying that link was merged, and a whole-body search would report that it was. """ found: list[dict] = [] seen: set[str] = set() for line in (text or "").splitlines(): for platform, pattern in _PR_PATTERNS: for match in pattern.finditer(line): url = match.group(0).rstrip(".,);:'\"") if url in seen: continue seen.add(url) found.append({ "url": url, "platform": platform, "repo": match.group(1), "number": int(match.group(2)), "line": line.strip(), }) return found def _state_word(line: str) -> str | None: """The state a recorded line attributes to the PR it names, if any.""" lowered = (line or "").lower() for state in _PR_STATES: if re.search(rf"\b{state}\b", lowered): return state return None # --------------------------------------------------------------------------- # Decision-bearing comments # --------------------------------------------------------------------------- # # Every marker below is a heading the workflow's own comment templates write # (reference/comment-templates.md), plus the words people use when they record a # decision by hand. Selection is a heuristic and is labelled as one: the claim # quotes the comment verbatim — a recorded fact — and carries the marker that # selected it, so a reader can audit the choice instead of trusting it. # # Deliberately absent: "Progress", "Captured", "Activated". They are workflow # bookkeeping; a slice that returns them returns noise with a citation. # # The tail of the tuple is the canonical status words: the transition template # leads with the target status ("REVIEW — Implementation complete — PR #326 …"), # and that comment is where an issue records what shipped and why. Without them # a workspace that uses the templates has no decisions at all (FLY-1482, found # against the real FLY-1300 record). # # Order matters — `decision_marker` matches with `startswith`, so a longer # marker sharing a prefix with a shorter one has to come first or the shorter # one swallows it. "ready for review" is above "ready" for exactly that reason. # This tuple's contents and order match `DECISION_MARKERS` in the remote tier's # `convex/lib/mcpChangeContext.ts`; the two must agree or the same comment # yields a different marker depending on which tier read it. 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", ) #: Emphasis runs, anywhere on the line. Stripping them is not editing the #: record: `**Blocked** — waiting on Kyle` and `Blocked — waiting on Kyle` are #: the same sentence, and the marker match should not turn on whether whoever #: wrote it used the template's bold. _EMPHASIS = re.compile(r"\*\*|__|`") _MARKUP = re.compile(r"^[\s>#*_`\-]+|[\s*_`]+$") #: The attribution line the relay prepends to every comment it writes on a #: person's behalf: `**@Matt Elsey** (via FlyDocs)`. It is who, not what, so it #: is never the lead line — reading it as one made `decision_marker` return None #: for every relay-written comment, and the slice's decisions always empty #: (FLY-1482). _ATTRIBUTION_LINE = re.compile(r"^@\S.*\(via [^)]+\)$", re.IGNORECASE) #: What a stripped code block leaves in its place. The value is the remote #: tier's `CODE_OMITTED_MARKER` (`convex/lib/scmPullRequestSummary.ts`) and the #: two must stay identical: both tiers skip this exact line when reading a lead #: line, and a slice read on one tier should say what the other one says. CODE_OMITTED_MARKER = "[code omitted]" #: A fenced block: ``` or ~~~ with an optional info string, to the closing #: fence. The backreference is the rule — the closing run is the *same* #: character and the same length as the opening one, which is what the remote #: tier's regex enforces. Anything else (a longer closer, a tilde closing a #: backtick fence) leaves the fence unterminated, and the pattern below takes #: it. _FENCED_BLOCK = re.compile( r"(^|\n)[ \t]*(`{3,}|~{3,})[^\n]*\n[\s\S]*?\n[ \t]*\2[ \t]*(?=\n|\Z)") #: An unterminated fence: everything from the fence to the end is code. `\Z` #: rather than `$` on purpose — Python's `$` also matches before a trailing #: newline, and the remote tier's `$` does not. _OPEN_FENCE = re.compile(r"(^|\n)[ \t]*(`{3,}|~{3,})[^\n]*(\n[\s\S]*)?\Z") #: Line endings, normalized first so a CRLF-authored comment reads the same. _CRLF = re.compile(r"\r\n?") def _strip_code_blocks(text: str) -> str: """Replace every fenced code block in `text` with `CODE_OMITTED_MARKER`. Prose around a block is untouched, so a comment that opens with a fence still yields the person's own first sentence as its lead line, and no excerpt this module quotes can carry code out of the repository. The twin is `stripCodeBlocks` in the remote tier's `convex/lib/scmPullRequestSummary.ts`, and the fence rules here match it line for line. That module also drops unified-diff fragments pasted into prose; this one does not, because the two tiers read different records — the remote tier summarizes GitHub pull requests, where a pasted hunk is a real leak path, while this module only ever reads tracker prose (FLY-1505). """ out = _CRLF.sub("\n", text or "") out = _FENCED_BLOCK.sub(lambda m: m.group(1) + CODE_OMITTED_MARKER, out) return _OPEN_FENCE.sub( lambda m: m.group(1) + CODE_OMITTED_MARKER, out, count=1) def _lead_line(body: str) -> str: """The first line of the comment's own prose, stripped of markdown furniture. Neither a fenced block at the top (the marker that replaced it is skipped) nor the relay's attribution line is the lead: scanning continues past both to what the person actually wrote. """ for line in _strip_code_blocks(body).split("\n"): cleaned = _MARKUP.sub("", _EMPHASIS.sub("", line)).strip() if not cleaned or cleaned == CODE_OMITTED_MARKER: continue if _ATTRIBUTION_LINE.match(cleaned): continue return cleaned return "" def _excerpt(value: str) -> tuple[str, bool]: """The text a claim quotes, and whether it was cut at `EXCERPT_CHARS`. Strips again on the way out even though `_lead_line` already did: the remote tier's `excerpt` does, and a quoted excerpt is the last place a fence could reach a transcript. """ cleaned = _strip_code_blocks(value).strip() if len(cleaned) <= EXCERPT_CHARS: return cleaned, False return cleaned[:EXCERPT_CHARS].rstrip(), True def decision_marker(body: str) -> str | None: """The marker that makes this comment decision-bearing, or None.""" lead = _lead_line(body).lower() if not lead: return None for marker in DECISION_MARKERS: if lead.startswith(marker): return marker return None # --------------------------------------------------------------------------- # Claims # --------------------------------------------------------------------------- RECORDED = "recorded" DERIVED = "derived" def _claim(kind: str, statement: str, refs: list[str], *, derivation: str, timestamp: str | None = None, **fields: object) -> dict | None: """Build one claim, or refuse to. The refusal is the §5.5 rule in code: no source refs, no claim. It returns None rather than raising because a single unciteable fact is not a reason to fail the whole read — the caller drops it, counts it, and warns. """ citations: list[str] = [] for ref in refs: if isinstance(ref, str) and ref.strip() and ref.strip() not in citations: citations.append(ref.strip()) if not citations or not (statement or "").strip(): return None claim = { "kind": kind, "statement": statement.strip(), "derivation": derivation, "source": { "refs": citations, "timestamp": timestamp, # A record with no timestamp of its own was last known true when it # was read. That moment is the slice's `observedAt`, recorded once # rather than copied onto every claim that shares it. "timestampKind": RECORDED if timestamp else "observed", }, } claim.update(fields) return claim def enforce_provenance(claims: list[dict]) -> tuple[list[dict], list[str]]: """Drop any claim that cannot cite a source. The second gate. `_claim` already refuses to build one, so this should never fire — which is exactly why it is here. It is cheap, it runs on the assembled slice rather than on one construction site, and it converts "someone appended a dict to the list" from a customer-visible false claim into a dropped claim and a warning. """ kept: list[dict] = [] dropped = 0 for claim in claims: source = claim.get("source") if isinstance(claim, dict) else None refs = source.get("refs") if isinstance(source, dict) else None if isinstance(refs, list) and any( isinstance(ref, str) and ref.strip() for ref in refs ): kept.append(claim) else: dropped += 1 warnings: list[str] = [] if dropped: warnings.append( f"{dropped} claim(s) were dropped: a claim with no source reference " "does not appear in a change_context slice (§5.5)." ) return kept, warnings def _comment_ref(comment: dict, index: int) -> str: """A citable handle for a comment — its id, or its position as a fallback.""" identifier = comment.get("id") if isinstance(identifier, str) and identifier.strip(): return f"comment:{identifier.strip()}" return f"comment:index-{index}" def _text(value: object) -> str: return value if isinstance(value, str) else "" def _record_timestamp(record: dict, *keys: str) -> str | None: """The record's own timestamp under the first key it actually has. Never `revision`: it is documented as an opaque change token that today happens to be a provider timestamp, and every caller that read a date out of it breaks on the first provider that returns an ETag (adapters/types.ts). """ for key in keys: value = record.get(key) if isinstance(value, str) and value.strip(): return value return None def _cut(items: list, limit: int, section: str, truncation: list[dict]) -> list: """Take the first `limit` items, recording the cut when there was one.""" if len(items) <= limit: return items truncation.append({ "section": section, "returned": limit, "available": len(items), }) return items[:limit] # --------------------------------------------------------------------------- # The slice # --------------------------------------------------------------------------- def build_shipped_slice(issue: object, *, target: str, observed_at: str, limit: int = DEFAULT_LIMIT) -> tuple[dict, list[str]]: """Assemble the shipped-mode slice from one issue body. Returns `(slice, warnings)`. Ordering is the join's own — the issue, then what shipped for it, then what it is connected to, then why — and inside the timestamped sections it is newest first, because the last decision is the one that still holds. """ if not isinstance(issue, dict): issue = {} # The provider's own identifier first, then what the caller asked for. # `id` is last on purpose: Linear's is a UUID, and a slice that cited a UUID # would be citing something nobody can look up. ref = normalize_ref( _text(issue.get("identifier")) or target or _text(issue.get("id")), ) claims: list[dict] = [] truncation: list[dict] = [] gaps: list[str] = [] # 1. The issue itself. title = _text(issue.get("title")) status = _text(issue.get("status")) headline = f'{ref} "{title}"' if title else ref claims.append(_claim( "issue", f"{headline} is {status}." if status else f"{headline} is recorded.", [ref], derivation=RECORDED, timestamp=_record_timestamp(issue, "updatedAt", "createdAt"), ref=ref, **({"title": title} if title else {}), **({"status": status} if status else {}), )) # 2. Pull requests, from every place the issue records one. description = _text(issue.get("description")) raw_comments = issue.get("comments") comments = raw_comments if isinstance(raw_comments, list) else [] if not isinstance(raw_comments, list): # An absent list is not an empty one. Saying which is the difference # between "nothing was decided here" and "the decisions were not read". gaps.append( "comments — the provider returned none, so PR references and " "decisions recorded in comments are not in this slice" ) sources: list[dict] = [{ "ref": f"{ref}#description", "text": description, "timestamp": _record_timestamp(issue, "updatedAt", "createdAt"), }] for index, comment in enumerate(comments): if not isinstance(comment, dict): continue sources.append({ "ref": _comment_ref(comment, index), "text": _text(comment.get("body")), "timestamp": _record_timestamp(comment, "createdAt", "updatedAt"), }) # One PR, however many places name it — grouped rather than deduplicated, # because which record is cited matters. The newest mention is the citation # (it is the one still true), and a state word is read from the newest # mention that carries one, which is how "opened" stops overriding "merged". grouped: dict[str, list[dict]] = {} for source in sources: for mention in _pr_mentions(source["text"]): grouped.setdefault(mention["url"], []).append( {**mention, "source": source}, ) def _when(mention: dict) -> str: return mention["source"]["timestamp"] or "" pull_requests = [ max(occurrences, key=_when) for occurrences in grouped.values() ] # Newest first, and stable for everything undated: a slice read twice # without the record changing has to come back in the same order, or a # caching client pays for the churn. pull_requests.sort(key=lambda m: (_when(m), m["url"]), reverse=True) pull_requests = _cut(pull_requests, limit, "pullRequests", truncation) for mention in pull_requests: source = mention["source"] pr_id = f'{mention["repo"]}#{mention["number"]}' claims.append(_claim( "pullRequest", f"{ref} records pull request {pr_id}.", [ref, source["ref"]], derivation=RECORDED, timestamp=source["timestamp"], url=mention["url"], repo=mention["repo"], number=mention["number"], platform=mention["platform"], )) stated = [ occurrence for occurrence in grouped[mention["url"]] if _state_word(occurrence["line"]) ] if stated: newest = max(stated, key=_when) state = _state_word(newest["line"]) # Read out of prose, so it is a reading, not a record — the marker # says so, and the ref points at the line it was read from. claims.append(_claim( "pullRequestState", f"The record naming {pr_id} describes it as {state}.", [ref, newest["source"]["ref"]], derivation=DERIVED, timestamp=newest["source"]["timestamp"], url=mention["url"], state=state, )) if pull_requests: gaps.append( "diffSummary, modulesTouched — no source records them against the " "issue; the tracker holds the PR link, not the PR's contents" ) # 3. Related issues. raw_links = issue.get("links") if isinstance(issue.get("links"), list) else [] # Filtered before it is cut: a link with no target is not a related issue, # and counting it in `available` would report a truncation that never # happened. links = [ link for link in raw_links if isinstance(link, dict) and normalize_ref(_text(link.get("targetRef"))) ] for link in _cut(links, limit, "relatedIssues", truncation): target_ref = normalize_ref(_text(link.get("targetRef"))) link_type = _text(link.get("type")) or "relates" link_title = _text(link.get("targetTitle")) described = f'{target_ref} "{link_title}"' if link_title else target_ref claims.append(_claim( "relatedIssue", f"{ref} {link_type} {described}.", [ref, target_ref], derivation=RECORDED, ref=target_ref, linkType=link_type, **({"title": link_title} if link_title else {}), )) # 4. Decisions. decisions: list[dict] = [] for index, comment in enumerate(comments): if not isinstance(comment, dict): continue body = _text(comment.get("body")) marker = decision_marker(body) if not marker: continue decisions.append({ "ref": _comment_ref(comment, index), "body": body, "marker": marker, "author": _text(comment.get("user")), "timestamp": _record_timestamp(comment, "createdAt", "updatedAt"), }) decisions.sort(key=lambda d: d["timestamp"] or "", reverse=True) for decision in _cut(decisions, limit, "decisions", truncation): excerpt, truncated_text = _excerpt(_lead_line(decision["body"])) claims.append(_claim( "decision", excerpt, [ref, decision["ref"]], derivation=RECORDED, timestamp=decision["timestamp"], marker=decision["marker"], truncatedText=truncated_text, **({"author": decision["author"]} if decision["author"] else {}), )) kept, warnings = enforce_provenance([c for c in claims if c is not None]) dropped_unciteable = len([c for c in claims if c is None]) if dropped_unciteable: warnings.append( f"{dropped_unciteable} fact(s) had no source reference and were not " "included (§5.5)." ) return { "target": target, "targetRef": ref, "mode": "shipped", "observedAt": observed_at, "limit": limit, "truncated": bool(truncation), "truncation": truncation, "gaps": gaps, "claims": kept, }, warnings