From 6e27d93684ee2325874ccc0ec13bc39d46080294 Mon Sep 17 00:00:00 2001 From: Lintel Date: Tue, 7 Apr 2026 19:23:02 +0000 Subject: [PATCH 1/3] Add post-implement verification that planned work was actually completed --- .../src/lintel/agent_definitions_api/store.py | 18 +- .../src/lintel/domain/seed_workflows.py | 2 + .../workflows/src/lintel/workflows/events.py | 19 + .../src/lintel/workflows/feature_to_pr.py | 16 +- .../lintel/workflows/nodes/_stage_tracking.py | 1 + .../workflows/nodes/verify_implementation.py | 341 ++++++++++++++++++ .../workflows/src/lintel/workflows/state.py | 13 +- .../workflows/src/lintel/workflows/types.py | 11 + 8 files changed, 418 insertions(+), 3 deletions(-) create mode 100644 packages/workflows/src/lintel/workflows/nodes/verify_implementation.py diff --git a/packages/agent-definitions-api/src/lintel/agent_definitions_api/store.py b/packages/agent-definitions-api/src/lintel/agent_definitions_api/store.py index f10a5251..779a10fd 100644 --- a/packages/agent-definitions-api/src/lintel/agent_definitions_api/store.py +++ b/packages/agent-definitions-api/src/lintel/agent_definitions_api/store.py @@ -4,7 +4,23 @@ from typing import Any -from lintel.persistence.data_models import AgentDefinitionData +from pydantic import BaseModel, ConfigDict, Field + + +class AgentDefinitionData(BaseModel): + """User-defined agent definition.""" + + agent_id: str + name: str = "" + description: str = "" + role: str = "" + system_prompt: str = "" + model_id: str = "" + temperature: float = 0.7 + tools: list[str] = Field(default_factory=list) + skills: list[str] = Field(default_factory=list) + + model_config = ConfigDict(extra="allow") class AgentDefinitionStore: diff --git a/packages/domain/src/lintel/domain/seed_workflows.py b/packages/domain/src/lintel/domain/seed_workflows.py index 199247fc..f8cac4d9 100644 --- a/packages/domain/src/lintel/domain/seed_workflows.py +++ b/packages/domain/src/lintel/domain/seed_workflows.py @@ -21,6 +21,7 @@ "plan", "approve_spec", "implement", + "verify_implementation", "review", "approved_for_pr", "raise_pr", @@ -34,6 +35,7 @@ "plan", "approval_gate_spec", "implement", + "verify_implementation", "review", "approval_gate_pr", "close", diff --git a/packages/workflows/src/lintel/workflows/events.py b/packages/workflows/src/lintel/workflows/events.py index 9215a117..f978663f 100644 --- a/packages/workflows/src/lintel/workflows/events.py +++ b/packages/workflows/src/lintel/workflows/events.py @@ -151,6 +151,23 @@ class HumanInterruptTimedOut(EventEnvelope): event_type: str = "HumanInterruptTimedOut" +# --- Implementation Verification Events --- + + +@dataclass(frozen=True) +class ImplementationVerified(EventEnvelope): + """Emitted when plan-vs-implementation verification passes.""" + + event_type: str = "ImplementationVerified" + + +@dataclass(frozen=True) +class ImplementationVerificationFailed(EventEnvelope): + """Emitted when plan-vs-implementation verification finds unaddressed tasks.""" + + event_type: str = "ImplementationVerificationFailed" + + # --- Guardrail Approval Events (GRD-6) --- @@ -192,4 +209,6 @@ class GuardrailApprovalRequested(EventEnvelope): GuardrailApprovalRequested, ChangeRequestTriggered, WorkflowQueued, + ImplementationVerified, + ImplementationVerificationFailed, ) diff --git a/packages/workflows/src/lintel/workflows/feature_to_pr.py b/packages/workflows/src/lintel/workflows/feature_to_pr.py index 35d7d9d3..0f24febc 100644 --- a/packages/workflows/src/lintel/workflows/feature_to_pr.py +++ b/packages/workflows/src/lintel/workflows/feature_to_pr.py @@ -24,6 +24,10 @@ from lintel.workflows.nodes.review import review_output from lintel.workflows.nodes.route import route_intent from lintel.workflows.nodes.setup_workspace import setup_workspace +from lintel.workflows.nodes.verify_implementation import ( + VerifyImplementationNode, + route_after_verification, +) from lintel.workflows.state import ThreadWorkflowState MAX_REVIEW_CYCLES = 3 @@ -40,6 +44,10 @@ def _resolve_max_review_cycles(state: ThreadWorkflowState) -> int: return int(state.get("max_review_cycles", DEFAULT_MAX_REVIEW_CYCLES)) +# Singleton node instance — safe because WorkflowNode is stateless between calls. +_verify_implementation_node = VerifyImplementationNode() + + def build_feature_to_pr_graph() -> StateGraph[Any]: """Build the feature-to-PR workflow graph.""" graph: StateGraph[Any] = StateGraph(ThreadWorkflowState) @@ -58,6 +66,7 @@ def build_feature_to_pr_graph() -> StateGraph[Any]: partial(approval_gate, gate_type="spec_approval"), ) graph.add_node("implement", spawn_implementation) + graph.add_node("verify_implementation", _verify_implementation_node) graph.add_node("review", review_output) graph.add_node( "approval_gate_pr", @@ -83,7 +92,12 @@ def build_feature_to_pr_graph() -> StateGraph[Any]: graph.add_conditional_edges( "implement", _check_phase, - {"continue": "review", "close": "close"}, + {"continue": "verify_implementation", "close": "close"}, + ) + graph.add_conditional_edges( + "verify_implementation", + route_after_verification, + {"review": "review", "implement": "implement", "close": "close"}, ) graph.add_conditional_edges( "review", diff --git a/packages/workflows/src/lintel/workflows/nodes/_stage_tracking.py b/packages/workflows/src/lintel/workflows/nodes/_stage_tracking.py index c14a723a..ec7b65a7 100644 --- a/packages/workflows/src/lintel/workflows/nodes/_stage_tracking.py +++ b/packages/workflows/src/lintel/workflows/nodes/_stage_tracking.py @@ -24,6 +24,7 @@ "plan": "plan", "approval_gate_spec": "approve_spec", "implement": "implement", + "verify_implementation": "verify_implementation", "test": "test", "review": "review", "approval_gate_pr": "approved_for_pr", diff --git a/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py b/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py new file mode 100644 index 00000000..1f522e5d --- /dev/null +++ b/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py @@ -0,0 +1,341 @@ +"""Verify implementation completeness against the plan. + +Compares plan tasks against actual file modifications in the sandbox +and computes a completeness score. Routes back to implement if too +many tasks are unaddressed (below COMPLETENESS_THRESHOLD) and retries +remain, otherwise proceeds to review. +""" + +from __future__ import annotations + +from pathlib import PurePosixPath +from typing import TYPE_CHECKING, Any + +import structlog + +from lintel.workflows.base import WorkflowNode + +if TYPE_CHECKING: + from langchain_core.runnables import RunnableConfig + + from lintel.workflows.state import ThreadWorkflowState + +logger = structlog.get_logger() + +# A plan task is considered addressed if this fraction or more of tasks +# have corresponding file modifications. +COMPLETENESS_THRESHOLD: float = 0.8 + +# Maximum number of implement → verify loops before giving up. +MAX_IMPLEMENT_ATTEMPTS: int = 3 + + +async def _get_modified_files( + config: dict[str, Any], + state: dict[str, Any], +) -> list[str]: + """Get the list of files modified in the sandbox via ``git diff``.""" + from lintel.sandbox.types import SandboxJob + + configurable = config.get("configurable", {}) + sandbox_manager = configurable.get("sandbox_manager") + if sandbox_manager is None: + run_id = configurable.get("run_id") or state.get("run_id", "") + if run_id: + from lintel.workflows.nodes._runtime_registry import get_sandbox_manager + + sandbox_manager = get_sandbox_manager(str(run_id)) + + sandbox_id = state.get("sandbox_id") + if sandbox_manager is None or not sandbox_id: + logger.warning("verify_impl_no_sandbox", sandbox_id=sandbox_id) + return [] + + workspace_path = state.get("workspace_path", "/workspace") + + # Use git diff against the base branch to find all modified files. + # --name-only --diff-filter=ACMR lists Added, Copied, Modified, Renamed files. + result = await sandbox_manager.execute( + sandbox_id, + SandboxJob( + command="git diff --name-only --diff-filter=ACMR HEAD~100 HEAD 2>/dev/null" + " || git diff --name-only --diff-filter=ACMR --cached 2>/dev/null" + " || git diff --name-only 2>/dev/null" + " || true", + workdir=workspace_path, + timeout_seconds=30, + ), + ) + output = getattr(result, "output", "") or "" + files = [line.strip() for line in output.splitlines() if line.strip()] + return files + + +def _task_addressed( + task: dict[str, Any], + modified_files: list[str], +) -> bool: + """Check whether a plan task is addressed by the modified files. + + Matches on: + - Exact file_path match + - Same filename (stem) in any directory + - Same parent directory + """ + file_path = task.get("file_path") or task.get("file") or "" + if not file_path: + # Tasks without a file_path are considered addressed by default + # (e.g. "update documentation", "run tests"). + return True + + task_path = PurePosixPath(file_path) + task_stem = task_path.stem + task_parent = str(task_path.parent) + + for mod_file in modified_files: + mod_path = PurePosixPath(mod_file) + # Exact match (relative paths) + if mod_file == file_path or str(mod_path) == str(task_path): + return True + # Same filename + if mod_path.stem == task_stem and mod_path.suffix == task_path.suffix: + return True + # Same directory — any file changed in the target directory counts + if str(mod_path.parent) == task_parent: + return True + return False + + +class VerifyImplementationNode(WorkflowNode): + """Cross-references plan tasks against sandbox-verified modified files.""" + + name: str = "verify_implementation" + + async def execute( + self, + state: ThreadWorkflowState, + config: RunnableConfig, + ) -> dict[str, Any]: + """Run verification and return state updates.""" + from lintel.workflows.types import VerificationResult + + await self.tracker.append_log( + self.name, "Verifying implementation completeness against plan..." + ) + + plan = state.get("plan", {}) + tasks: list[dict[str, Any]] = plan.get("tasks", []) + + if not tasks: + # No tasks in plan — nothing to verify, pass through. + result = VerificationResult( + completeness_score=1.0, + addressed_tasks=(), + unaddressed_tasks=(), + modified_files=(), + attempt_count=state.get("implement_attempt_count", 0), + ) + await self.tracker.append_log(self.name, "No plan tasks to verify — passing through.") + await self._emit_verified(config, state, result) + await self.complete(outputs={"completeness_score": 1.0}) + return {"verification_result": result, "verification_feedback": None} + + # Get ground-truth modified files from sandbox. + try: + modified_files = await _get_modified_files(dict(config), dict(state)) + except Exception as exc: + logger.exception("verify_impl_sandbox_error", error=str(exc)) + await self.tracker.append_log(self.name, f"Failed to list modified files: {exc}") + await self.complete(error=f"Sandbox error: {exc}") + return { + "error": f"verify_implementation failed: {exc}", + "current_phase": "failed", + } + + # Cross-reference tasks against modified files. + addressed: list[str] = [] + unaddressed: list[str] = [] + for task in tasks: + task_id = task.get("id") or task.get("title") or task.get("description", "unknown") + if _task_addressed(task, modified_files): + addressed.append(str(task_id)) + else: + unaddressed.append(str(task_id)) + + total = len(tasks) + score = len(addressed) / total if total > 0 else 1.0 + attempt_count = state.get("implement_attempt_count", 0) + + result = VerificationResult( + completeness_score=score, + addressed_tasks=tuple(addressed), + unaddressed_tasks=tuple(unaddressed), + modified_files=tuple(modified_files), + attempt_count=attempt_count, + ) + + await self.tracker.append_log( + self.name, + f"Completeness: {score:.0%} " + f"({len(addressed)}/{total} tasks addressed, " + f"attempt {attempt_count + 1}/{MAX_IMPLEMENT_ATTEMPTS})", + ) + + if unaddressed: + await self.tracker.append_log( + self.name, + f"Unaddressed tasks: {', '.join(unaddressed[:10])}" + + ("..." if len(unaddressed) > 10 else ""), + ) + + # Decide pass/fail. + passes = score >= COMPLETENESS_THRESHOLD + max_reached = attempt_count >= MAX_IMPLEMENT_ATTEMPTS + + if passes or max_reached: + if max_reached and not passes: + await self.tracker.append_log( + self.name, + f"Max attempts ({MAX_IMPLEMENT_ATTEMPTS}) reached — " + f"proceeding with {score:.0%} completeness.", + ) + await self._emit_verified(config, state, result) + await self.complete( + outputs={ + "completeness_score": score, + "addressed_count": len(addressed), + "total_count": total, + } + ) + return { + "verification_result": result, + "verification_feedback": None, + } + + # Fail — route back to implement. + feedback_lines = [ + "The following planned tasks were not addressed by the implementation:", + "", + ] + for task in tasks: + task_id = task.get("id") or task.get("title") or task.get("description", "unknown") + if str(task_id) in unaddressed: + desc = task.get("description", task.get("title", task_id)) + fp = task.get("file_path") or task.get("file") or "" + feedback_lines.append(f"- {desc}" + (f" ({fp})" if fp else "")) + + feedback = "\n".join(feedback_lines) + + await self._emit_verification_failed(config, state, result) + await self.complete( + outputs={ + "completeness_score": score, + "addressed_count": len(addressed), + "total_count": total, + } + ) + return { + "verification_result": result, + "verification_feedback": feedback, + "implement_attempt_count": attempt_count + 1, + } + + # ------------------------------------------------------------------ + # Event emission helpers + # ------------------------------------------------------------------ + + async def _emit_verified( + self, + config: RunnableConfig, + state: ThreadWorkflowState, + result: Any, + ) -> None: + from lintel.workflows.events import ImplementationVerified + + self._emit_event( + config, + state, + ImplementationVerified, + result, + ) + + async def _emit_verification_failed( + self, + config: RunnableConfig, + state: ThreadWorkflowState, + result: Any, + ) -> None: + from lintel.workflows.events import ImplementationVerificationFailed + + self._emit_event( + config, + state, + ImplementationVerificationFailed, + result, + ) + + def _emit_event( + self, + config: RunnableConfig, + state: ThreadWorkflowState, + event_cls: type, + result: Any, + ) -> None: + """Best-effort event emission via the event bus.""" + try: + from dataclasses import asdict + + configurable = config.get("configurable", {}) + event_bus = configurable.get("event_bus") + if event_bus is None: + return + + payload = { + "stage": "verify_implementation", + "verification_result": asdict(result), + } + run_id = state.get("run_id", "") + if run_id: + payload["run_id"] = run_id + + event = event_cls( + actor_id="system", + payload=payload, + ) + # Fire-and-forget — event bus may be sync or async. + import asyncio + + loop = asyncio.get_event_loop() + if hasattr(event_bus, "publish"): + coro = event_bus.publish(event) + if asyncio.iscoroutine(coro): + loop.create_task(coro) + except Exception: + logger.debug("verify_impl_event_emit_failed", exc_info=True) + + +def route_after_verification(state: ThreadWorkflowState) -> str: + """Pure routing function for the conditional edge after verify_implementation. + + Returns: + ``'review'`` — verification passed or max retries exceeded. + ``'implement'`` — below threshold and retries remain. + ``'close'`` — error or explicit failure. + """ + if state.get("error"): + return "close" + + result = state.get("verification_result") + if result is None: + # No verification result — pass through to review. + return "review" + + score = result.completeness_score + attempt_count = state.get("implement_attempt_count", 0) + + if score >= COMPLETENESS_THRESHOLD: + return "review" + if attempt_count >= MAX_IMPLEMENT_ATTEMPTS: + return "review" + # Below threshold and attempts remain — loop back. + return "implement" diff --git a/packages/workflows/src/lintel/workflows/state.py b/packages/workflows/src/lintel/workflows/state.py index 950bc96e..05833644 100644 --- a/packages/workflows/src/lintel/workflows/state.py +++ b/packages/workflows/src/lintel/workflows/state.py @@ -3,7 +3,10 @@ from __future__ import annotations from operator import add -from typing import Annotated, Any, TypedDict +from typing import TYPE_CHECKING, Annotated, Any, TypedDict + +if TYPE_CHECKING: + from lintel.workflows.types import VerificationResult class ThreadWorkflowState(TypedDict): @@ -68,3 +71,11 @@ class ThreadWorkflowState(TypedDict): # Project conventions — concatenated CLAUDE.md file contents from the target repo. # Collected by setup_workspace and injected into implement/review agent system prompts. project_conventions: str + + # Implementation verification (verify_implementation node) + # Result of comparing plan tasks against actual sandbox file modifications. + verification_result: VerificationResult | None + # Human-readable summary of unaddressed tasks, fed back to the coder on retry. + verification_feedback: str | None + # Loop guard: number of implement attempts (incremented each loop-back). + implement_attempt_count: int diff --git a/packages/workflows/src/lintel/workflows/types.py b/packages/workflows/src/lintel/workflows/types.py index 281387b7..5af49ad8 100644 --- a/packages/workflows/src/lintel/workflows/types.py +++ b/packages/workflows/src/lintel/workflows/types.py @@ -286,3 +286,14 @@ class ApprovalDecision: approver: str = "" corrections: str = "" feedback: str = "" + + +@dataclass(frozen=True) +class VerificationResult: + """Result of comparing plan tasks against actual file modifications in the sandbox.""" + + completeness_score: float + addressed_tasks: tuple[str, ...] + unaddressed_tasks: tuple[str, ...] + modified_files: tuple[str, ...] + attempt_count: int From 11bd5b19a14c50a78d55ab1bfa6b542f5ab645e6 Mon Sep 17 00:00:00 2001 From: Bamdad Dashtban Date: Tue, 7 Apr 2026 21:32:20 +0100 Subject: [PATCH 2/3] fix: resolve lint/type errors in verify_implementation node - Replace Any with VerificationResult type in event emission methods - Convert _emit_event to async and await event_bus.publish directly - Remove from __future__ import annotations for LangGraph runtime compat - Add verify_implementation to expected graph nodes in test Co-Authored-By: Claude Opus 4.6 --- .../workflows/nodes/verify_implementation.py | 34 +++++++------------ .../workflows/src/lintel/workflows/state.py | 12 ++++--- .../tests/workflows/test_feature_to_pr.py | 1 + 3 files changed, 20 insertions(+), 27 deletions(-) diff --git a/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py b/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py index 1f522e5d..77e79cbc 100644 --- a/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py +++ b/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py @@ -6,19 +6,15 @@ remain, otherwise proceeds to review. """ -from __future__ import annotations - from pathlib import PurePosixPath -from typing import TYPE_CHECKING, Any +from typing import Any import structlog +from langchain_core.runnables import RunnableConfig from lintel.workflows.base import WorkflowNode - -if TYPE_CHECKING: - from langchain_core.runnables import RunnableConfig - - from lintel.workflows.state import ThreadWorkflowState +from lintel.workflows.state import ThreadWorkflowState +from lintel.workflows.types import VerificationResult logger = structlog.get_logger() @@ -248,11 +244,11 @@ async def _emit_verified( self, config: RunnableConfig, state: ThreadWorkflowState, - result: Any, + result: VerificationResult, ) -> None: from lintel.workflows.events import ImplementationVerified - self._emit_event( + await self._emit_event( config, state, ImplementationVerified, @@ -263,23 +259,23 @@ async def _emit_verification_failed( self, config: RunnableConfig, state: ThreadWorkflowState, - result: Any, + result: VerificationResult, ) -> None: from lintel.workflows.events import ImplementationVerificationFailed - self._emit_event( + await self._emit_event( config, state, ImplementationVerificationFailed, result, ) - def _emit_event( + async def _emit_event( self, config: RunnableConfig, state: ThreadWorkflowState, event_cls: type, - result: Any, + result: VerificationResult, ) -> None: """Best-effort event emission via the event bus.""" try: @@ -290,7 +286,7 @@ def _emit_event( if event_bus is None: return - payload = { + payload: dict[str, Any] = { "stage": "verify_implementation", "verification_result": asdict(result), } @@ -302,14 +298,8 @@ def _emit_event( actor_id="system", payload=payload, ) - # Fire-and-forget — event bus may be sync or async. - import asyncio - - loop = asyncio.get_event_loop() if hasattr(event_bus, "publish"): - coro = event_bus.publish(event) - if asyncio.iscoroutine(coro): - loop.create_task(coro) + await event_bus.publish(event) except Exception: logger.debug("verify_impl_event_emit_failed", exc_info=True) diff --git a/packages/workflows/src/lintel/workflows/state.py b/packages/workflows/src/lintel/workflows/state.py index 05833644..da2d9232 100644 --- a/packages/workflows/src/lintel/workflows/state.py +++ b/packages/workflows/src/lintel/workflows/state.py @@ -1,12 +1,14 @@ -"""Workflow state definitions. TypedDict for LangGraph compatibility.""" +"""Workflow state definitions. TypedDict for LangGraph compatibility. -from __future__ import annotations +Note: ``from __future__ import annotations`` is deliberately omitted here +because LangGraph's ``StateGraph`` calls ``get_type_hints()`` at runtime, +which requires all referenced types to be resolvable in the module namespace. +""" from operator import add -from typing import TYPE_CHECKING, Annotated, Any, TypedDict +from typing import Annotated, Any, TypedDict -if TYPE_CHECKING: - from lintel.workflows.types import VerificationResult +from lintel.workflows.types import VerificationResult # noqa: TCH001 class ThreadWorkflowState(TypedDict): diff --git a/packages/workflows/tests/workflows/test_feature_to_pr.py b/packages/workflows/tests/workflows/test_feature_to_pr.py index 117daebf..b83323c9 100644 --- a/packages/workflows/tests/workflows/test_feature_to_pr.py +++ b/packages/workflows/tests/workflows/test_feature_to_pr.py @@ -143,6 +143,7 @@ def test_graph_has_expected_nodes(self) -> None: "plan", "approval_gate_spec", "implement", + "verify_implementation", "review", "approval_gate_pr", "close", From 480e5b11814f403917b1f5df00c903a41ffdcbe4 Mon Sep 17 00:00:00 2001 From: Bamdad Dashtban Date: Tue, 7 Apr 2026 21:48:18 +0100 Subject: [PATCH 3/3] fix: resolve lint errors in verify_implementation and state imports Co-Authored-By: Claude Opus 4.6 --- .../src/lintel/workflows/nodes/verify_implementation.py | 2 +- packages/workflows/src/lintel/workflows/state.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py b/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py index 77e79cbc..6f491fc4 100644 --- a/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py +++ b/packages/workflows/src/lintel/workflows/nodes/verify_implementation.py @@ -9,8 +9,8 @@ from pathlib import PurePosixPath from typing import Any -import structlog from langchain_core.runnables import RunnableConfig +import structlog from lintel.workflows.base import WorkflowNode from lintel.workflows.state import ThreadWorkflowState diff --git a/packages/workflows/src/lintel/workflows/state.py b/packages/workflows/src/lintel/workflows/state.py index da2d9232..693a3f9f 100644 --- a/packages/workflows/src/lintel/workflows/state.py +++ b/packages/workflows/src/lintel/workflows/state.py @@ -8,7 +8,7 @@ from operator import add from typing import Annotated, Any, TypedDict -from lintel.workflows.types import VerificationResult # noqa: TCH001 +from lintel.workflows.types import VerificationResult class ThreadWorkflowState(TypedDict):