Fix the Review Lens plugin's wire contract, and 20 defects behind it

backend/agent_dispatch/code_review.py

@@ -98,18 +98,50 @@ async def start_code_review_run(
         the registry is pure task/cancel_event bookkeeping into the
         shared cancel()/cancel_all() sweep)."""
         from backend import agents as agents_module  # deferred: patch-seam + circular-import safety
+        from backend.domain.model import SceneError
+
+        # Defined ABOVE the busy check on purpose: the check-to-claim stretch
+        # below must contain no statement holding an `await`
+        # (tests/test_dispatch_claim_ordering.py's AST gate scans it
+        # recursively), and this is the same placement _dispatch_chat uses
+        # for its own _finalize.
+        claimed: dict[str, str | None] = {"request_id": None}
+
+        async def _finalize() -> None:
+            # ADR-006 stage 6.2's finalize hook, which this surface was
+            # missing. RunRegistry.cancel() pops the handle and schedules
+            # this the moment a cancel lands; _run's own finally does it on
+            # a normal completion, and release() returning True is the
+            # arbiter so it never happens twice.
+            #
+            # Without it, cancelling a review freed the REGISTRY slot but
+            # left node.pending_request_id claimed, because only _run's
+            # finally cleared that - and _run is parked behind an
+            # uninterruptible asyncio.to_thread. So "Cancel" left the node
+            # visibly busy, with every Review Lens action on it refused,
+            # until the model call returned on its own: up to the full
+            # 600-second watchdog.
+            request_id = claimed["request_id"]
+            if request_id is not None and node.pending_request_id == request_id:
+                node.pending_request_id = None
+            await bus.publish("scene")
+
         if node.pending_request_id and node.pending_request_id != agents_module._NODE_RUN_CLAIM_PLACEHOLDER:
             notifications_state.show("Review Lens is already busy for this node.", "info")
             await bus.publish("notification")
             return
 
         cancel_event = threading.Event()
-        handle = self._runs.claim("code_review_run", node_id=node_id, cancel_event=cancel_event)
+        handle = self._runs.claim(
+            "code_review_run", node_id=node_id, cancel_event=cancel_event, finalize=_finalize,
+        )
         request_id = handle.request_id
+        claimed["request_id"] = request_id
         node.pending_request_id = request_id
         await bus.publish("scene")
 
         async def _run():
+            released = False
             try:
                 result = await asyncio.wait_for(
                     asyncio.to_thread(agents_module._call_review_lens_agent, bundle),
@@ -119,29 +151,63 @@ async def _run():
                     notifications_state.show("Review Lens run cancelled.", "info")
                     await bus.publish("notification")
                 else:
-                    on_success(result)
-                    await bus.publish("scene")
+                    try:
+                        on_success(result)
+                    except SceneError:
+                        # The node was deleted while the review ran.
+                        # complete_code_review_run goes through require_node
+                        # and raises; fail_code_review_run's own docstring
+                        # already establishes that a background result
+                        # landing after node deletion is a SILENT no-op, so
+                        # letting this fall through to the generic handler
+                        # below only produced a "Review Lens run failed:
+                        # unknown node: n7" toast for a node the user had
+                        # deliberately removed.
+                        agents_module.logger.info(
+                            "code review result discarded - node deleted mid-run",
+                            extra={"run_id": request_id, "node_id": node_id},
+                        )
+                    else:
+                        await bus.publish("scene")
             except asyncio.TimeoutError:
                 cancel_event.set()
-                notifications_state.show(
+                message = (
                     "Review Lens stopped responding before the request completed. "
-                    "Please try again.",
-                    "error",
+                    "Please try again."
                 )
+                # on_failure as well as the toast. The timeout branch used to
+                # show only a notification, so once that toast was dismissed
+                # (or missed - it fires up to ten minutes after the click)
+                # the node was indistinguishable from one that had never been
+                # run: no error banner, no state change, nothing to say the
+                # review had been attempted and had failed.
+                on_failure(message)
+                notifications_state.show(message, "error")
                 await bus.publish("notification")
+            except asyncio.CancelledError:
+                # Not caught by `except Exception`. Without this the finally
+                # below still runs, but re-raising is required so the event
+                # loop sees the task as cancelled rather than completed.
+                self._runs.release(request_id)
+                released = True
+                if node.pending_request_id == request_id:
+                    node.pending_request_id = None
+                raise
             except Exception as exc:
                 agents_module.logger.exception("code review dispatch failed")
                 on_failure(f"Review Lens run failed: {exc}")
                 notifications_state.show(f"Review Lens run failed: {exc}", "error")
                 await bus.publish("notification")
             finally:
-                self._runs.release(request_id)
-                # Only clear if this task's OWN request_id is still the one
-                # recorded - a stale, already-superseded task finishing
-                # late must never clobber a newer legitimate busy marker.
-                if node.pending_request_id == request_id:
-                    node.pending_request_id = None
-                await bus.publish("scene")
+                if not released:
+                    # release() returns False when cancel() already popped the
+                    # handle - in that case _finalize has already run the end
+                    # transition and this late teardown must not repeat it,
+                    # nor clobber a newer run's busy marker.
+                    if self._runs.release(request_id):
+                        if node.pending_request_id == request_id:
+                            node.pending_request_id = None
+                        await bus.publish("scene")
 
         self._runs.attach_task(handle, asyncio.create_task(_run()))
 
@@ -178,5 +244,15 @@ async def _action():
     def cancel_code_review(self, request_id: str) -> bool:
         """kind="code_review_run": see RunRegistry.cancel's own docstring
         for why kind= is passed now that code_review_run shares self._runs
-        with other cancel_event-bearing kinds."""
+        with other cancel_event-bearing kinds.
+
+        Returns False for a request_id this registry does not hold, which
+        for Review Lens specifically means "the busy marker belongs to a
+        fetch or an Ask, not a review run". Those two claim
+        node.pending_request_id through _run_node_blocking_action, which
+        mints a bare uuid4 and never registers it here - they are awaited
+        directly by their intent and own no cancellation primitive. The
+        caller MUST surface that False rather than dropping it: the node's
+        Cancel button is offered for any pending request, so a silently
+        ignored return made it a dead control that looked live."""
         return self._runs.cancel(request_id, kind="code_review_run")

backend/agents.py

@@ -177,11 +177,19 @@
 # finish well under this.
 GITLINK_CONTEXT_TIMEOUT_SECONDS = 300
 # Review Lens: one PR-metadata GET + up to two pages of file-listing GETs +
-# one diff download (network-timeout-capped at 60s by diff_fetch itself).
-CODE_REVIEW_DIFF_TIMEOUT_SECONDS = 120
-# Review Lens: one LLM completion over up to 45,000 chars of diff
-# (review_engine's MAX_DIFF_MODEL_CHARS) - same call-count shape as a
-# Gitlink run, with a comparable input size, hence the same watchdog.
+# one diff download. Those are 25s, 25s, 25s and 60s of per-request network
+# timeout respectively (GitHubRestClient.request's default, and diff_fetch's
+# own _DIFF_TIMEOUT_SECONDS), so the work this watchdog bounds can legally
+# take 135 seconds - and the watchdog was 120. A fetch that was merely slow,
+# not stuck, was reported to the user as "stopped responding" while the
+# request it had given up on was still running and about to succeed. The
+# outer bound has to be larger than the sum of the inner ones, with headroom
+# for the JSON parsing and normalization between them.
+CODE_REVIEW_DIFF_TIMEOUT_SECONDS = 180
+# Review Lens: one LLM completion over up to MAX_DIFF_CHARS of diff
+# (review_engine's MAX_DIFF_MODEL_CHARS, which is that same constant) - same
+# call-count shape as a Gitlink run, with a comparable input size, hence the
+# same watchdog.
 CODE_REVIEW_RUN_TIMEOUT_SECONDS = 600
 # Review Lens: one follow-up Q&A completion over the same capped diff.
 CODE_REVIEW_ASK_TIMEOUT_SECONDS = 300

backend/api/intents_code_review.py

@@ -21,6 +21,8 @@
 from backend.agents import _NODE_RUN_CLAIM_PLACEHOLDER, AgentDispatcher
 from backend.api._shared import claim_busy_node_or_notify, make_publish_scene
 from backend.domain.graph import SceneDocument
+from backend.domain.node_access import is_node_of
+from backend.domain.node_states import CodeReviewState
 from backend.events import SessionBus
 from backend.notifications import NotificationState
 
@@ -43,7 +45,7 @@ async def set_code_review_pr_url(node_id, pr_url):
 
     async def fetch_code_review_diff(node_id, pr_url=None):
         node = document.nodes.get(node_id)
-        if node is None or node.kind != "code_review":
+        if not is_node_of(node, "code_review", CodeReviewState):
             notifications.show("This node no longer exists.", "warning")
             await bus.publish("notification")
             return None
@@ -69,7 +71,7 @@ async def fetch_code_review_diff_text(node_id):
 
     async def run_code_review(node_id):
         node = document.nodes.get(node_id)
-        if node is None or node.kind != "code_review":
+        if not is_node_of(node, "code_review", CodeReviewState):
             notifications.show("This node no longer exists.", "warning")
             await bus.publish("notification")
             return None
@@ -132,11 +134,25 @@ def _on_failure(message):
         return node_id
 
     async def cancel_code_review_request(request_id):
-        agent_dispatcher.cancel_code_review(request_id)
+        # The return value is load-bearing, not decoration. Only a review RUN
+        # is registered as a cancellable run; a diff fetch and an Ask claim
+        # the same node.pending_request_id busy marker through
+        # _run_node_blocking_action, which owns no cancellation primitive.
+        # The node offers Cancel for any pending request, so dropping this
+        # False left the user clicking a button that did nothing and said
+        # nothing. It cannot be hidden at the UI layer either - the wire
+        # carries one opaque pendingRequestId with no kind attached.
+        if not agent_dispatcher.cancel_code_review(request_id):
+            notifications.show(
+                "Only a running review can be cancelled. The pull-request "
+                "fetch will finish on its own.",
+                "info",
+            )
+            await bus.publish("notification")
 
     async def ask_code_review_question(node_id, question):
         node = document.nodes.get(node_id)
-        if node is None or node.kind != "code_review":
+        if not is_node_of(node, "code_review", CodeReviewState):
             notifications.show("This node no longer exists.", "warning")
             await bus.publish("notification")
             return None

backend/domain/graph.py

@@ -100,6 +100,106 @@ def _estimate_tokens(text: str) -> int:
     return TokenEstimator().count_tokens(text)
 
 
+# -- Review Lens nested wire rows ------------------------------------------
+#
+# These five builders exist because the Review Lens node stores its nested
+# rows as the review engine's own snake_case dicts (graphlink_plugins/
+# review_lens/review_engine.py) while the wire contract - and the generated
+# client validator built from it - is camelCase, like every other nested
+# row on SceneNodeRow. scene_payload used to forward them with a bare
+# `dict(row)`, which shipped `patch_truncated`/`previous_path`/
+# `group_title` where CodeReviewFileRow and CodeReviewWalkthroughGroupRow
+# declare `patchTruncated`/`previousPath`/`groupTitle`.
+#
+# That was not a cosmetic mismatch. validateSceneState treats a missing
+# required field as a hard error, and web_ui/src/lib/api-contract/
+# bindTopic.ts DROPS a snapshot that fails validation, so from the first
+# successful PR fetch onward every scene snapshot for the whole session
+# was rejected client-side - the canvas froze for every node, not just
+# this one. Building each row explicitly (the `toolCalls` precedent
+# below) fixes the casing AND guarantees each row carries exactly the
+# contract's fields with the contract's types, so a row that reached the
+# state from an old save file cannot put an unexpected key on the wire.
+#
+# The domain keeps snake_case on purpose: it is what the engine emits and
+# what session_save.py has already written to every existing save file.
+# The conversion belongs at the wire builder, the same place
+# codeReviewScores is coerced to dict[str, str].
+
+
+def _code_review_file_wire(row: dict[str, Any]) -> dict[str, Any]:
+    wire: dict[str, Any] = {
+        "path": str(row.get("path", "")),
+        "status": str(row.get("status", "modified")),
+        "additions": _non_negative_wire_int(row.get("additions")),
+        "deletions": _non_negative_wire_int(row.get("deletions")),
+        # `patch` is DELIBERATELY not forwarded - see the caller's comment.
+        "patch": "",
+        "patchTruncated": bool(row.get("patch_truncated", False)),
+    }
+    # Genuinely absent (not "") for every non-rename, which is why
+    # CodeReviewFileRow.previousPath is the one Optional field on the row.
+    previous = str(row.get("previous_path", "") or "")
+    if previous:
+        wire["previousPath"] = previous
+    return wire
+
+
+def _code_review_walkthrough_wire(row: dict[str, Any]) -> dict[str, Any]:
+    raw_paths = row.get("paths")
+    return {
+        "groupTitle": str(row.get("group_title", "")),
+        "paths": [str(path) for path in raw_paths] if isinstance(raw_paths, list) else [],
+        "explanation": str(row.get("explanation", "")),
+    }
+
+
+def _code_review_finding_wire(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": str(row.get("id", "")),
+        "severity": str(row.get("severity", "")),
+        "tier": str(row.get("tier", "")),
+        "category": str(row.get("category", "")),
+        "path": str(row.get("path", "")),
+        "line": _non_negative_wire_int(row.get("line")),
+        "title": str(row.get("title", "")),
+        "evidence": str(row.get("evidence", "")),
+        "impact": str(row.get("impact", "")),
+        "recommendation": str(row.get("recommendation", "")),
+    }
+
+
+def _code_review_error_wire(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": str(row.get("id", "")),
+        "severity": str(row.get("severity", "")),
+        "tier": str(row.get("tier", "")),
+        "kind": str(row.get("kind", "")),
+        "path": str(row.get("path", "")),
+        "line": _non_negative_wire_int(row.get("line")),
+        "title": str(row.get("title", "")),
+        "evidence": str(row.get("evidence", "")),
+        "fix": str(row.get("fix", "")),
+    }
+
+
+def _code_review_qa_wire(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "question": str(row.get("question", "")),
+        "answer": str(row.get("answer", "")),
+    }
+
+
+def _non_negative_wire_int(value: Any) -> int:
+    """int for the wire, never raising. A row can reach the wire from a
+    hand-edited save file as well as from the engine, and a ValueError
+    here would fail the whole scene republish, not just one row."""
+    try:
+        return max(0, int(value))  # type: ignore[call-overload]
+    except (TypeError, ValueError, OverflowError):
+        return 0
+
+
 
 
 @dataclass
@@ -1084,8 +1184,24 @@ def _node_wire(self, n: SceneNode) -> dict[str, Any]:
             "codeReviewChangedFiles": (
                 n.state.code_review_changed_files if isinstance(n.state, CodeReviewState) else 0
             ),
+            # Rebuilt row by row rather than forwarded as `dict(f)` - see
+            # _code_review_file_wire's own comment for the casing bug that
+            # cost every scene snapshot after a PR fetch.
+            #
+            # `patch` rides as "" on purpose. The per-file patches are
+            # capped at MAX_FILE_PATCH_CHARS (6000) each and MAX_PR_FILES
+            # (100) of them, so forwarding them put up to ~600KB of diff
+            # text on EVERY scene republish - roughly ten times the
+            # 60KB codeReviewDiffText that was excluded from this payload
+            # for exactly that reason (see CodeReviewState's own comment).
+            # No frontend code reads it: CodeReviewNodeView renders the
+            # unified diff it lazily fetches via fetchCodeReviewDiffText,
+            # never these per-file patches. The field stays on the row
+            # because the contract declares it required; the engine still
+            # reads the real patches from node.state, which is where the
+            # fallback pre-screen scans them.
             "codeReviewFiles": (
-                [dict(f) for f in n.state.code_review_files]
+                [_code_review_file_wire(f) for f in n.state.code_review_files]
                 if isinstance(n.state, CodeReviewState)
                 else []
             ),
@@ -1108,17 +1224,17 @@ def _node_wire(self, n: SceneNode) -> dict[str, Any]:
                 n.state.code_review_diff_version if isinstance(n.state, CodeReviewState) else 0
             ),
             "codeReviewWalkthrough": (
-                [dict(g) for g in n.state.code_review_walkthrough]
+                [_code_review_walkthrough_wire(g) for g in n.state.code_review_walkthrough]
                 if isinstance(n.state, CodeReviewState)
                 else []
             ),
             "codeReviewFindings": (
-                [dict(f) for f in n.state.code_review_findings]
+                [_code_review_finding_wire(f) for f in n.state.code_review_findings]
                 if isinstance(n.state, CodeReviewState)
                 else []
             ),
             "codeReviewErrors": (
-                [dict(e) for e in n.state.code_review_errors]
+                [_code_review_error_wire(e) for e in n.state.code_review_errors]
                 if isinstance(n.state, CodeReviewState)
                 else []
             ),
@@ -1156,7 +1272,7 @@ def _node_wire(self, n: SceneNode) -> dict[str, Any]:
                 n.state.code_review_quality_summary if isinstance(n.state, CodeReviewState) else ""
             ),
             "codeReviewQa": (
-                [dict(entry) for entry in n.state.code_review_qa]
+                [_code_review_qa_wire(entry) for entry in n.state.code_review_qa]
                 if isinstance(n.state, CodeReviewState)
                 else []
             ),