diff --git a/src/agent/api/api.py b/src/agent/api/api.py index d785989..59ef489 100644 --- a/src/agent/api/api.py +++ b/src/agent/api/api.py @@ -5,6 +5,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware +from agent.api.routes.compact import router as compact_router from agent.api.routes.health import router as health_router from agent.api.routes.streaming import router as streaming_router @@ -25,6 +26,7 @@ app.include_router(health_router) app.include_router(streaming_router) +app.include_router(compact_router) if __name__ == "__main__": diff --git a/src/agent/api/routes/compact.py b/src/agent/api/routes/compact.py new file mode 100644 index 0000000..05999cc --- /dev/null +++ b/src/agent/api/routes/compact.py @@ -0,0 +1,19 @@ +"""Chat compaction route.""" + +from __future__ import annotations + +from fastapi import APIRouter + +from agent.api.schemas.compact import CompactRequest, CompactResponse +from agent.services.compaction import compact_conversation + +router = APIRouter(tags=["Compaction"]) + + +@router.post("/compact", response_model=CompactResponse) +async def compact_chat(request: CompactRequest): + """Compact a conversation into a shorter summary.""" + summary = await compact_conversation( + request.messages, request.last_summary, request.llm + ) + return CompactResponse(summary=summary) diff --git a/src/agent/api/routes/streaming.py b/src/agent/api/routes/streaming.py index 26634a7..67a0f5c 100644 --- a/src/agent/api/routes/streaming.py +++ b/src/agent/api/routes/streaming.py @@ -9,7 +9,12 @@ from fastapi.responses import StreamingResponse from langgraph.types import Command -from agent.api.schemas.streaming import ResumeRequest, StreamRequest, build_messages +from agent.api.schemas.streaming import ( + ClarificationResumeRequest, + PermissionResumeRequest, + StreamRequest, + build_messages, +) from agent.streaming.runner import graph, request_cancel, stream_graph @@ -44,6 +49,7 @@ async def generate_stream(request: StreamRequest): "plan_attempt_count": 0, "pending_question": {}, "human_clarifications": [], + "write_rejected": False, "current_yaml_draft": "", "required_secrets": [], "validator_errors": [], @@ -72,14 +78,13 @@ async def generate_stream(request: StreamRequest): ) -@router.post("/generate/{thread_id}/stream/resume") -async def resume_stream(thread_id: str, request: ResumeRequest): - """Resume an interrupted workflow generation stream. +@router.post("/generate/{thread_id}/resume/permission") +async def resume_permission(thread_id: str, request: PermissionResumeRequest): + """Resume interrupted stream with permission decision. Raises: - HTTPException: 404 if thread_id does not exist + HTTPException: 404 if thread not found, 409 if validation fails """ - # Validate thread exists before attempting resume config = { "configurable": { "thread_id": thread_id, @@ -89,18 +94,93 @@ async def resume_stream(thread_id: str, request: ResumeRequest): state = graph.get_state(config) - # MemorySaver returns empty state for unknown threads if not state.values: raise HTTPException( status_code=404, - detail=f"Thread '{thread_id}' not found. Cannot resume non-existent thread.", + detail=f"Thread '{thread_id}' not found", + ) + + pending = state.values.get("pending_permission_request", {}) + if not pending: + raise HTTPException( + status_code=409, + detail="No pending permission request", + ) + + if request.request_id != pending.get("request_id"): + raise HTTPException( + status_code=409, + detail=f"Request ID mismatch. Expected '{pending.get('request_id')}', got '{request.request_id}'", + ) + + pending_q = state.values.get("pending_question", {}) + if pending_q.get("type") != "permission": + raise HTTPException( + status_code=409, + detail=f"Expected permission question, but pending type is '{pending_q.get('type')}'", + ) + + run_id = f"run_{uuid4().hex[:12]}" + + return StreamingResponse( + stream_graph( + Command(resume=request.decision), + config, + thread_id, + run_id, + is_resume=True, + ), + media_type="application/x-ndjson", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Thread-Id": thread_id, + "X-Run-Id": run_id, + }, + ) + + +@router.post("/generate/{thread_id}/resume/clarification") +async def resume_clarification(thread_id: str, request: ClarificationResumeRequest): + """Resume interrupted stream with clarification answer. + + Raises: + HTTPException: 404 if thread not found, 409 if validation fails + """ + config = { + "configurable": { + "thread_id": thread_id, + "llm_config": request.llm.model_dump(), + } + } + + state = graph.get_state(config) + + if not state.values: + raise HTTPException( + status_code=404, + detail=f"Thread '{thread_id}' not found", + ) + + pending_q = state.values.get("pending_question", {}) + if not pending_q: + raise HTTPException( + status_code=409, + detail="No pending question", + ) + + q_type = pending_q.get("type", "clarification") + if q_type == "permission": + raise HTTPException( + status_code=409, + detail="Use /resume/permission endpoint for permission requests", ) run_id = f"run_{uuid4().hex[:12]}" return StreamingResponse( stream_graph( - Command(resume=request.response), + Command(resume=request.answer), config, thread_id, run_id, diff --git a/src/agent/api/schemas/__init__.py b/src/agent/api/schemas/__init__.py index 6ac0f94..90d1eb1 100644 --- a/src/agent/api/schemas/__init__.py +++ b/src/agent/api/schemas/__init__.py @@ -1,10 +1,21 @@ +from .compact import CompactMessageItem, CompactRequest, CompactResponse from .health import HealthResponse -from .streaming import MessageItem, ResumeRequest, StreamRequest, build_messages +from .streaming import ( + ClarificationResumeRequest, + MessageItem, + PermissionResumeRequest, + StreamRequest, + build_messages, +) __all__ = [ + "CompactMessageItem", + "CompactRequest", + "CompactResponse", "HealthResponse", "MessageItem", - "ResumeRequest", + "PermissionResumeRequest", + "ClarificationResumeRequest", "StreamRequest", "build_messages", ] diff --git a/src/agent/api/schemas/compact.py b/src/agent/api/schemas/compact.py new file mode 100644 index 0000000..9079063 --- /dev/null +++ b/src/agent/api/schemas/compact.py @@ -0,0 +1,35 @@ +"""Schemas for the chat compaction endpoint.""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, Field + +from agent.api.schemas.streaming import LLMConfig + + +class CompactMessageItem(BaseModel): + """A single conversation turn to be compacted.""" + + role: Literal["user", "assistant"] = Field(..., description="Speaker role") + content: str = Field(..., description="Message text") + + +class CompactRequest(BaseModel): + """Request body for POST /compact.""" + + messages: list[CompactMessageItem] = Field( + ..., description="Conversation turns to compact (user + assistant)" + ) + last_summary: str = Field( + default="", + description="Previous compaction summary text. Empty on first compaction.", + ) + llm: LLMConfig = Field(..., description="Per-request LLM configuration (required).") + + +class CompactResponse(BaseModel): + """Compaction result returned to the backend.""" + + summary: str = Field(..., description="Compressed summary text") diff --git a/src/agent/api/schemas/streaming.py b/src/agent/api/schemas/streaming.py index 3c9b268..616e4b9 100644 --- a/src/agent/api/schemas/streaming.py +++ b/src/agent/api/schemas/streaming.py @@ -38,6 +38,13 @@ class LLMConfig(BaseModel): temperature: float = Field(default=0, description="Sampling temperature. 0 = " "deterministic, the right default for structured " "YAML generation. Caller can override per request.") + reasoning_effort: Literal["none", "low", "medium", "high"] = Field( + default="none", + description="Model reasoning (thinking) budget. 'none' = off (default, " + "thinking tokens are billable). 'low'/'medium'/'high' allocate " + "an increasing reasoning budget. Mapped per-provider by LiteLLM; " + "output presence is provider-dependent.", + ) @model_validator(mode="after") def _validate_provider_requirements(self) -> "LLMConfig": @@ -92,8 +99,16 @@ class StreamRequest(BaseModel): llm: LLMConfig = Field(..., description="Per-request LLM configuration (required).") -class ResumeRequest(BaseModel): - """Request body for resuming an interrupted stream.""" +class PermissionResumeRequest(BaseModel): + """Request body for resuming with permission decision.""" + + decision: Literal["allow", "deny"] = Field(..., description="Permission decision") + request_id: str = Field(..., description="Request ID from permission event metadata") + llm: LLMConfig = Field(..., description="Per-request LLM configuration (required).") + + +class ClarificationResumeRequest(BaseModel): + """Request body for resuming with clarification answer.""" - response: str = Field(..., description="User answer to the pending question") + answer: str = Field(..., description="Answer to the clarification question") llm: LLMConfig = Field(..., description="Per-request LLM configuration (required).") diff --git a/src/agent/graph/graph.py b/src/agent/graph/graph.py index 6ba8ed8..7285dd8 100644 --- a/src/agent/graph/graph.py +++ b/src/agent/graph/graph.py @@ -3,7 +3,8 @@ The supervisor hub sits at the center. After each node executes, control returns to the supervisor, which reads the updated state -and routes to the next node. Only the writer node exits to END. +and routes to the next node. The supervisor exits to END once a run +reaches a terminal state (the writer sets terminal_status on success). """ from __future__ import annotations @@ -44,17 +45,16 @@ def build_graph(): sg.add_edge("analyzer", "supervisor") sg.add_edge("generator", "supervisor") sg.add_edge("human_interaction", "supervisor") + sg.add_edge("writer", "supervisor") - # Writer is the terminal node - sg.add_edge("writer", END) - - # --- Supervisor conditional routing --- + # Supervisor exits to END once a run reaches a terminal state. sg.add_conditional_edges("supervisor", supervisor_decision, { "planner": "planner", "analyzer": "analyzer", "generator": "generator", "human_interaction": "human_interaction", "writer": "writer", + "__end__": END, }) return sg.compile(checkpointer=MemorySaver()) diff --git a/src/agent/graph/nodes/planner.py b/src/agent/graph/nodes/planner.py index 376c7b8..d776022 100644 --- a/src/agent/graph/nodes/planner.py +++ b/src/agent/graph/nodes/planner.py @@ -18,7 +18,7 @@ from agent.graph.state import State from agent.llm.helpers import get_llm_from_config from agent.llm.resilience import build_resilient_llm, classify_llm_error -from agent.streaming.emitter import emit_plan, emit_title, emit_update, emit_llm_error +from agent.streaming.emitter import emit_plan, emit_thinking, emit_title, emit_update, emit_llm_error def _build_context_block(state: State) -> str: @@ -105,6 +105,21 @@ def planner_node(state: State, config: RunnableConfig) -> dict: Returns a partial state dict with only the fields this node writes. """ + if state.get("write_rejected"): + return { + "pending_question": { + "type": "clarification", + "source": "planner", + "message": ( + "You rejected the generated workflow. What should I change? " + "(e.g. 'use a different trigger', 'bump node to 20', 'add a caching step')" + ), + }, + "write_rejected": False, + "replan_requested": False, + "replan_reason": "", + } + # Deadlock prevention: if planner repeatedly returns empty plan without # asking for analysis or human input, escalate to human. if state.get("plan_attempt_count", 0) >= MAX_EMPTY_PLAN_ATTEMPTS: @@ -138,6 +153,7 @@ def planner_node(state: State, config: RunnableConfig) -> dict: "replan_requested": False, "replan_reason": "", "replan_count": 0, + "current_yaml_draft": "", } llm = get_llm_from_config(config) @@ -156,7 +172,7 @@ def planner_node(state: State, config: RunnableConfig) -> dict: conversation.append(HumanMessage(content=context_block)) try: - response: PlannerOutput = structured_llm.invoke([ + response, raw = structured_llm.invoke([ SystemMessage(content=system_prompt), *conversation, ]) @@ -172,6 +188,10 @@ def planner_node(state: State, config: RunnableConfig) -> dict: # Re-raise to let the runner handle terminal failure raise + thinking = (raw.additional_kwargs.get("reasoning_content") if raw else "") or "" + if thinking: + emit_thinking(thinking) + if response.public_update: emit_update(response.public_update) diff --git a/src/agent/graph/nodes/supervisor.py b/src/agent/graph/nodes/supervisor.py index 3585eab..f288453 100644 --- a/src/agent/graph/nodes/supervisor.py +++ b/src/agent/graph/nodes/supervisor.py @@ -14,12 +14,17 @@ def supervisor_decision(state: State) -> str: """Decide the next node based on current state flags. Returns one of: - "analyzer" — an analysis query is pending - "planner" — need a plan (first time or replan) - "generator" — have a plan, need to generate YAML - "human_interaction" — draft ready for review, or circuit breaker hit - "writer" — draft approved, save to disk + "analyzer" — an analysis query is pending + "planner" — need a plan (first time or replan) + "generator" — have a plan, need to generate YAML + "human_interaction" — a clarification or permission prompt is pending + "writer" — draft present, request write approval + "__end__" — run reached a terminal state """ + # 0. Terminal state? exit the graph. + if state.get("terminal_status") in {"success", "failed"}: + return "__end__" + # 1. Pending human question? always route to HITL gateway first if state.get("pending_question"): return "human_interaction" @@ -40,5 +45,5 @@ def supervisor_decision(state: State) -> str: if not state.get("current_yaml_draft"): return "generator" - # 6. Draft ready → save & end (no final approval gate) + # 6. Draft ready → request write approval return "writer" diff --git a/src/agent/graph/nodes/writer.py b/src/agent/graph/nodes/writer.py index 5565c11..11fc663 100644 --- a/src/agent/graph/nodes/writer.py +++ b/src/agent/graph/nodes/writer.py @@ -1,22 +1,29 @@ """ -Writer node — deterministic file writer for approved YAML drafts. - -The planner decides the filepath; this node resolves the absolute path, -avoids overwrites by appending a number, and writes the file. -No LLM calls — pure Python. +Writer node - deterministic file writer for approved YAML drafts. + +Two-phase write approval gate (mirrors analyzer_node's permission pattern): + Run 1 (draft present, no pending permission): resolve the output path, + emit the draft for preview, then request write permission and return + WITHOUT writing. The graph pauses in human_interaction_node. + Run 2 (pending permission present): read the allow/deny answer. + allow: write to disk, emit run.completed. + deny: route to planner via write_rejected so it asks for clarification. +No LLM calls - pure Python. """ from __future__ import annotations from pathlib import Path +from uuid import uuid4 from langchain_core.messages import AIMessage from langgraph.config import get_stream_writer from agent.graph.state import State -from agent.streaming.emitter import emit_run_completed +from agent.streaming.emitter import emit_run_completed, emit_yaml_draft from agent.streaming.events import StreamEvent from agent.tools import write_tool +from agent.tools.permission.protocol import parse_allow_deny, permission_answer_for_request MAX_FILENAME_ATTEMPTS = 100 @@ -28,7 +35,7 @@ def _resolve_output_path(project_path: str, planned_filepath: str) -> str: if not base_path.exists(): return str(base_path) - # Append incrementing number: ci.yml → ci-2.yml → ci-3.yml + # Append incrementing number: ci.yml -> ci-2.yml -> ci-3.yml stem = base_path.stem suffix = base_path.suffix parent = base_path.parent @@ -45,7 +52,7 @@ def _resolve_output_path(project_path: str, planned_filepath: str) -> str: def write_yaml(project_path: str, yaml_draft: str, planned_filepath: str) -> str: """Write the YAML draft to disk and return the absolute path. - + Raises: OSError: If the file cannot be written (disk full, permissions, etc.) RuntimeError: If a unique filename cannot be found @@ -56,9 +63,9 @@ def write_yaml(project_path: str, yaml_draft: str, planned_filepath: str) -> str def writer_node(state: State) -> dict: - """Parent graph node: saves the draft and returns output_path.""" project_path = state.get("project_path", "") yaml_draft = state.get("current_yaml_draft", "") + planned_filepath = state.get("planned_filepath", "") if not yaml_draft: return {} @@ -66,48 +73,86 @@ def writer_node(state: State) -> dict: if not project_path: raise ValueError("writer_node requires 'project_path' in state") - planned_filepath = state.get("planned_filepath", "") - - # Fallback if planner didn't provide a filepath if not planned_filepath: - target_platform = state.get("target_platform", "github_actions") - planned_filepath = ( - ".gitlab-ci.yml" - if target_platform == "gitlab_ci" - else ".github/workflows/ci.yml" + raise ValueError("writer_node requires 'planned_filepath' in state") + + pending_permission = state.get("pending_permission_request") or {} + + # Run 2: We already asked for permission, now check the user's answer + if pending_permission: + request_id = pending_permission.get("request_id") + decision = parse_allow_deny(permission_answer_for_request(state, request_id)) + + # User gave invalid/mismatched answer - ask again + if decision is None: + return { + "pending_question": { + "type": "permission", + "source": "writer", + "message": f"Allow writer to save the workflow to {pending_permission['command']}?", + "metadata": pending_permission, + } + } + + # User denied - route to planner for clarification + if not decision: + return { + "write_rejected": True, + "replan_requested": True, + "replan_reason": "User rejected the generated workflow draft.", + "current_yaml_draft": "", + "pending_question": {}, + "pending_permission_request": {}, + } + + # User allowed - write to disk + try: + path = write_yaml(project_path, yaml_draft, planned_filepath) + except (OSError, RuntimeError) as e: + get_stream_writer()(StreamEvent( + type="run.failed", + node="writer", + data={ + "message": f"Failed to write workflow file: {e}", + "detail": type(e).__name__, + "draft_content": yaml_draft, + }, + ).model_dump(mode="json")) + raise + + emit_run_completed( + output_path=path, + pipeline_plan=state.get("pipeline_plan", []), + warnings=state.get("validator_warnings", []), ) - try: - path = write_yaml( - project_path=project_path, - yaml_draft=yaml_draft, - planned_filepath=planned_filepath, - ) - except (OSError, RuntimeError) as e: - # OSError: Disk write failure (permissions, disk full, read-only mount) - # RuntimeError: Filename collision exhausted (extremely rare) - # Emit failure event with draft content so client can offer download - writer = get_stream_writer() - writer(StreamEvent( - type="run.failed", - node="writer", - data={ - "message": f"Failed to write workflow file: {e}", - "detail": type(e).__name__, - "resolved_path": str(Path(project_path) / planned_filepath), - "draft_content": yaml_draft, - }, - ).model_dump(mode="json")) - raise - - emit_run_completed( - output_path=path, - pipeline_plan=state.get("pipeline_plan", []), - warnings=state.get("validator_warnings", []), - ) + return { + "output_path": path, + "messages": [AIMessage(content=f"Workflow saved to `{path}`.")], + "terminal_status": "success", + "pending_permission_request": {}, + } + + # Run 1: First time - request permission to write + resolved_path = _resolve_output_path(project_path, planned_filepath) + + permission = { + "type": "permission", + "request_id": str(uuid4()), + "source": "writer", + "command": resolved_path, + "cmd_type": "modify", + "working_directory": project_path, + } + + emit_yaml_draft(filepath=resolved_path, content=yaml_draft) return { - "output_path": path, - "messages": [AIMessage(content=f"Workflow saved to `{path}`.")], - "terminal_status": "success", + "pending_question": { + "type": "permission", + "source": "writer", + "message": f"Allow writer to save the workflow to {resolved_path}?", + "metadata": permission, + }, + "pending_permission_request": permission, } diff --git a/src/agent/graph/state.py b/src/agent/graph/state.py index e9f0066..877705c 100644 --- a/src/agent/graph/state.py +++ b/src/agent/graph/state.py @@ -53,6 +53,12 @@ class State(TypedDict): pending_question: PendingQuestion human_clarifications: list[dict] + # --- Write Approval --- + # True when the user denied the generated draft at the writer's permission + # gate. Forces the planner to ask for clarification instead of replanning + # blindly. Cleared once the planner emits that clarification. + write_rejected: bool + # --- Generation & Reflexion --- current_yaml_draft: str required_secrets: list[dict] diff --git a/src/agent/graph/subgraphs/analyzer.py b/src/agent/graph/subgraphs/analyzer.py index 21ca724..40f0ea6 100644 --- a/src/agent/graph/subgraphs/analyzer.py +++ b/src/agent/graph/subgraphs/analyzer.py @@ -3,6 +3,11 @@ Uses LangChain's create_agent() with inspection tools to answer specific questions about a project's structure, dependencies, build system, etc. + +The agent runs as a buffered per-turn stream so model reasoning (thinking) is +surfaced turn-by-turn, interleaved with the live tool events the tools already +emit. Layer-2 tenacity retry wraps the whole streaming run; on retry the loop +re-runs from scratch (a small number of tool events may re-emit). """ from __future__ import annotations @@ -10,17 +15,17 @@ from datetime import datetime, timezone from langchain.agents import create_agent -from langchain_core.messages import HumanMessage +from langchain_core.messages import AIMessageChunk, HumanMessage, ToolMessage from langchain_core.runnables import RunnableConfig from tenacity import retry +from agent.exceptions import PermissionRequiredException from agent.graph.prompts import ANALYZER_SYSTEM_PROMPT from agent.graph.schemas import AnalyzerResultEntry, PendingQuestion from agent.graph.state import State from agent.llm.helpers import get_llm_from_config from agent.llm.resilience import retry_policy -from agent.tools.permission.protocol import parse_allow_deny -from agent.exceptions import PermissionRequiredException +from agent.streaming.emitter import emit_thinking from agent.tools import ( create_bash_tool, create_glob_tool, @@ -28,6 +33,7 @@ create_list_tool, read_tool, ) +from agent.tools.permission.protocol import parse_allow_deny, permission_answer_for_request def _normalize_message_content(content) -> str: @@ -44,21 +50,6 @@ def _normalize_message_content(content) -> str: return str(content) -def _permission_answer_for_request(state: State, request_id: str | None) -> str: - clarifications = state.get("human_clarifications", []) or [] - - if request_id: - for record in reversed(clarifications): - if record.get("type") != "permission": - continue - if str(record.get("request_id", "")) == str(request_id): - return str(record.get("answer", "")) - return "" - - # if request_id is missing, force explicit re-prompt. - return "" - - def build_analyzer_agent(project_path: str, llm=None, approved_commands: set[str] | None = None): """Build a compiled ReAct agent for codebase analysis. @@ -92,7 +83,7 @@ def build_analyzer_agent(project_path: str, llm=None, approved_commands: set[str return create_agent(model=llm, tools=tools, system_prompt=prompt) -def run_analyzer( +async def run_analyzer_streaming( project_path: str, query: str, llm=None, @@ -100,31 +91,73 @@ def run_analyzer( ) -> dict: """Run a single analysis query and return the text answer. + Streams the ReAct loop with stream_mode="messages" so reasoning (thinking) + is buffered per turn and emitted as blobs, interleaved with the live tool + events the tools already emit. Final answer is the last AIMessage content. + + Layer-2 tenacity retry wraps the whole streaming run. On retry the loop + re-runs from scratch; the per-turn thinking buffer is local to each + attempt so it auto-discards, and a small number of tool events may re-emit. + Args: project_path: Absolute path to the project. query: Question about the project. llm: Optional LLM override. + approved_commands: Pre-approved shell commands (skip permission HITL). Returns: - The agent's final text answer. + {"answer": str, "permission_request": dict | None} """ - agent = build_analyzer_agent(project_path, llm=llm, approved_commands=approved_commands) + agent = build_analyzer_agent( + project_path, llm=llm, approved_commands=approved_commands + ) - # Layer-2 retry around the ReAct loop. Each attempt runs the full agent - # invoke; retryable transport/parse errors back off (spanning a provider - # rate-limit window) and emit "retrying (N)..." via the shared policy's - # before_sleep callback. Final failure falls through to the except below. @retry(**retry_policy()) - def _invoke_agent(): - return agent.invoke({"messages": [HumanMessage(content=query)]}) + async def _stream_agent() -> str: + # Per-turn reasoning buffer. A ToolMessage marks a turn boundary: + # the model's reasoning for that turn is complete -> flush it. + turn_thinking: list[str] = [] + answer = "" + + async for chunk, _metadata in agent.astream( + {"messages": [HumanMessage(content=query)]}, + stream_mode="messages", + ): + # Check for ToolMessage first (turn boundary marker) + if isinstance(chunk, ToolMessage): + # Turn boundary: flush buffered thinking before tool events. + if turn_thinking: + emit_thinking("".join(turn_thinking)) + turn_thinking = [] + # AIMessageChunk is a subclass of AIMessage; check the specific + # chunk type first so we read streamed reasoning deltas. + elif isinstance(chunk, AIMessageChunk): + delta = chunk.additional_kwargs.get("reasoning_content") or "" + if delta: + turn_thinking.append(delta) + # For chunks, accumulate the content + if chunk.content: + content = _normalize_message_content(chunk.content) + if content: + answer = content + # Handle any other message types that have content + elif hasattr(chunk, 'content') and chunk.content: + # Complete message - capture the answer + content = _normalize_message_content(chunk.content) + if content: + answer = content + + # Flush any trailing reasoning from the final turn (no tool follows it). + if turn_thinking: + emit_thinking("".join(turn_thinking)) + + return answer answer = "" permission_request = None try: - result = _invoke_agent() - messages = result.get("messages", []) - answer = _normalize_message_content(messages[-1].content) if messages else "" + answer = await _stream_agent() except PermissionRequiredException as e: permission_request = e.payload answer = "Analysis halted: Permission required." @@ -141,7 +174,7 @@ def _invoke_agent(): } -def analyzer_node(state: State, config: RunnableConfig) -> dict: +async def analyzer_node(state: State, config: RunnableConfig) -> dict: """Parent graph node: reads query from state, runs analyzer, writes result. Registered as: ``graph.add_node("analyzer", analyzer_node)`` @@ -162,7 +195,7 @@ def analyzer_node(state: State, config: RunnableConfig) -> dict: if permission_request: request_id = permission_request.get("request_id") - permission_answer = _permission_answer_for_request(state, request_id) + permission_answer = permission_answer_for_request(state, request_id) decision = parse_allow_deny(permission_answer) if decision is None: @@ -197,7 +230,7 @@ def analyzer_node(state: State, config: RunnableConfig) -> dict: if approved_command: approved_commands.add(approved_command) - result = run_analyzer( + result = await run_analyzer_streaming( project_path=project_path, query=query, llm=llm, diff --git a/src/agent/graph/subgraphs/generator_reflexion/generator.py b/src/agent/graph/subgraphs/generator_reflexion/generator.py index 027fc22..cd06acd 100644 --- a/src/agent/graph/subgraphs/generator_reflexion/generator.py +++ b/src/agent/graph/subgraphs/generator_reflexion/generator.py @@ -20,7 +20,7 @@ from agent.graph.state import State from agent.llm.helpers import get_llm_from_config from agent.llm.resilience import build_resilient_llm, classify_llm_error -from agent.streaming.emitter import emit_update, emit_llm_error +from agent.streaming.emitter import emit_thinking, emit_update, emit_llm_error from agent.utils import clean_yaml_fences @@ -97,7 +97,7 @@ def generator_node(state: State, config: RunnableConfig) -> dict: conversation.append(HumanMessage(content=user_message)) try: - response: GeneratorOutput = structured_llm.invoke([ + response, raw = structured_llm.invoke([ SystemMessage(content=system_prompt), *conversation, ]) @@ -113,6 +113,10 @@ def generator_node(state: State, config: RunnableConfig) -> dict: # Re-raise to let the runner handle terminal failure raise + thinking = (raw.additional_kwargs.get("reasoning_content") if raw else "") or "" + if thinking: + emit_thinking(thinking) + if response.public_update: emit_update(response.public_update) diff --git a/src/agent/llm/factory.py b/src/agent/llm/factory.py index c45d9b9..cd58383 100644 --- a/src/agent/llm/factory.py +++ b/src/agent/llm/factory.py @@ -47,24 +47,26 @@ def _make_hashable_config(cfg: "LLMConfig") -> tuple: cfg.api_key or "", cfg.base_url or "", cfg.temperature, + cfg.reasoning_effort, ) @lru_cache(maxsize=32) def _cached_build_llm(cfg_tuple: tuple) -> BaseChatModel: """Internal cached builder. Do not call directly. - + Raises: AgentConfigError: If LLM construction fails due to config issues """ # Reconstruct a minimal config-like object - provider, model, api_key, base_url, temperature = cfg_tuple + provider, model, api_key, base_url, temperature, reasoning_effort = cfg_tuple cfg = LLMConfig( - provider=provider, + provider=provider, model=model, api_key=api_key or None, base_url=base_url or None, temperature=temperature, + reasoning_effort=reasoning_effort, ) try: @@ -77,6 +79,12 @@ def _cached_build_llm(cfg_tuple: tuple) -> BaseChatModel: kwargs["api_key"] = cfg.api_key if cfg.base_url: kwargs["api_base"] = cfg.base_url # LiteLLM's param is named api_base + if cfg.reasoning_effort != "none": + # LiteLLM auto-maps this to each provider's native thinking knob + # (anthropic thinking/budget_tokens, gemini thinking_level, openai + # native, openai_compatible passthrough). Output normalized to + # additional_kwargs["reasoning_content"]. + kwargs["reasoning_effort"] = cfg.reasoning_effort return ChatLiteLLM(**kwargs) except Exception as e: diff --git a/src/agent/llm/resilience.py b/src/agent/llm/resilience.py index 4d53a63..d7b0869 100644 --- a/src/agent/llm/resilience.py +++ b/src/agent/llm/resilience.py @@ -155,55 +155,83 @@ def build_resilient_llm( max_retries: int = 2, ) -> Any: """Build a structured-output LLM with built-in retry + notification. - + This provides two-layer error handling: - + Layer 1 (LiteLLM num_retries, configured in factory.py): - Silently retries transport errors (429/5xx/timeout) - Fast recovery for transient blips - Respects provider Retry-After headers - User sees nothing if it succeeds - + Layer 2 (tenacity retry, this function): - Catches transport errors that exhausted Layer 1 (sustained issues) - Also catches parse errors (schema mismatch) - Emits llm.error events via before_sleep callback - - User sees "retrying (N)…" notification - + - User sees "retrying (N)..." notification + On final failure after all retries, the exception propagates to the - call site for classification via classify_llm_error() → LLMInvocationError. - + call site for classification via classify_llm_error() -> LLMInvocationError. + + include_raw=True is used so the raw AIMessage is available on + ``last_raw_message`` for reasoning_content extraction (thinking). The + wrapper isolates the unwrap internally and still returns the clean parsed + schema to callers, so node call sites are unchanged. + Args: llm: Base LangChain chat model (configured with num_retries in factory) schema: Pydantic model for structured output max_retries: Maximum retry attempts for Layer 2 (default: 2) Combined with num_retries=3, max 6 total HTTP attempts - + Returns: - A Runnable that parses into ``schema`` with built-in resilience - + A Runnable that parses into ``schema`` with built-in resilience. + Exposes ``last_raw_message`` (set on success only) for thinking capture. + """ - base_runnable = llm.with_structured_output(schema, method="function_calling") + base_runnable = llm.with_structured_output( + schema, method="function_calling", include_raw=True + ) # Layer-2 tenacity settings shared by both sync and async paths. # max_retries is retries, retry_policy takes total attempts -> +1. _retry_kwargs = retry_policy(max_retries + 1) + def _unwrap(result: Any) -> tuple[Any, Any]: + """Isolate include_raw translation. + + Returns (parsed_schema, raw_message). Parse failures are re-raised so + Layer-2 tenacity still retries them (the include_raw dict would + otherwise swallow them). Falls through with a None raw for the + bare-schema shape returned by some mocks. + """ + if not isinstance(result, dict) or "parsed" not in result: + return result, None # bare schema (test mocks pass through) + + parsing_error = result.get("parsing_error") + if parsing_error is not None: + raise parsing_error # let tenacity retry, classify_llm_error later + + return result["parsed"], result.get("raw") + @retry(**_retry_kwargs) - def _invoke(input: Any, config: Any = None, **kwargs: Any) -> Any: - return base_runnable.invoke(input, config, **kwargs) + def _invoke(input: Any, config: Any = None, **kwargs: Any) -> tuple[Any, Any]: + return _unwrap(base_runnable.invoke(input, config, **kwargs)) @retry(**_retry_kwargs) - async def _ainvoke(input: Any, config: Any = None, **kwargs: Any) -> Any: - return await base_runnable.ainvoke(input, config, **kwargs) + async def _ainvoke(input: Any, config: Any = None, **kwargs: Any) -> tuple[Any, Any]: + return _unwrap(await base_runnable.ainvoke(input, config, **kwargs)) class _ResilientRunnable: - """Thin Runnable-compatible wrapper backed by tenacity retry.""" + """Thin Runnable-compatible wrapper backed by tenacity retry. + + Returns tuple of (parsed_schema, raw_message) from invoke/ainvoke. + """ - def invoke(self, input: Any, config: Any = None, **kwargs: Any) -> Any: + def invoke(self, input: Any, config: Any = None, **kwargs: Any) -> tuple[Any, Any]: return _invoke(input, config, **kwargs) - async def ainvoke(self, input: Any, config: Any = None, **kwargs: Any) -> Any: + async def ainvoke(self, input: Any, config: Any = None, **kwargs: Any) -> tuple[Any, Any]: return await _ainvoke(input, config, **kwargs) return _ResilientRunnable() diff --git a/src/agent/services/__init__.py b/src/agent/services/__init__.py index e69de29..4e6414b 100644 --- a/src/agent/services/__init__.py +++ b/src/agent/services/__init__.py @@ -0,0 +1,3 @@ +from agent.services.compaction import compact_conversation + +__all__ = ["compact_conversation"] diff --git a/src/agent/services/compaction/__init__.py b/src/agent/services/compaction/__init__.py new file mode 100644 index 0000000..a5c83ec --- /dev/null +++ b/src/agent/services/compaction/__init__.py @@ -0,0 +1,4 @@ +from agent.services.compaction.compaction import compact_conversation +from agent.services.compaction.compaction_prompt import COMPACTION_SYSTEM_PROMPT + +__all__ = ["compact_conversation", "COMPACTION_SYSTEM_PROMPT"] diff --git a/src/agent/services/compaction/compaction.py b/src/agent/services/compaction/compaction.py new file mode 100644 index 0000000..ae55232 --- /dev/null +++ b/src/agent/services/compaction/compaction.py @@ -0,0 +1,40 @@ +"""Chat compaction service -- compresses conversations via LLM.""" + +from __future__ import annotations + +import logging + +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage + +from agent.api.schemas.compact import CompactMessageItem +from agent.api.schemas.streaming import LLMConfig +from agent.llm.factory import build_llm +from agent.services.compaction.compaction_prompt import COMPACTION_SYSTEM_PROMPT + +logger = logging.getLogger(__name__) + + +async def compact_conversation( + messages: list[CompactMessageItem], + last_summary: str, + llm_config: LLMConfig, +) -> str: + """Run the compaction pipeline and return compressed summary text.""" + llm = build_llm(llm_config) + + llm_messages = [SystemMessage(content=COMPACTION_SYSTEM_PROMPT)] + + if last_summary: + llm_messages.append( + AIMessage(content=f"[CONVERSATION SUMMARY]\n{last_summary}") + ) + + for msg in messages: + if msg.role == "user": + llm_messages.append(HumanMessage(content=msg.content)) + else: + llm_messages.append(AIMessage(content=msg.content)) + + response = await llm.ainvoke(llm_messages) + content = response.content + return content if isinstance(content, str) else str(content) diff --git a/src/agent/services/compaction/compaction_prompt.py b/src/agent/services/compaction/compaction_prompt.py new file mode 100644 index 0000000..5d62b60 --- /dev/null +++ b/src/agent/services/compaction/compaction_prompt.py @@ -0,0 +1,43 @@ +"""Prompt for the chat compaction summarizer.""" + +COMPACTION_SYSTEM_PROMPT = """\ +You are a conversation compactor for a coding assistant. Your job is to \ +compress a conversation into a shorter summary that preserves all \ +information relevant to future interactions. + +If the conversation includes a message tagged [CONVERSATION SUMMARY], \ +it contains a summary of earlier conversation history. Merge it with \ +the new conversation into a single coherent summary. + +PRESERVE (never drop): +- All user preferences and constraints +- Project requirements and architectural decisions +- Implementation decisions and their rationale +- Unresolved tasks, TODOs, and open questions +- Important names: classes, APIs, endpoints, variables, file paths +- Technology choices and version constraints +- Anything that could influence future responses + +REMOVE: +- Greetings, thanks, acknowledgements +- Repeated explanations of the same concept +- Conversational filler and pleasantries +- Outdated decisions that were explicitly superseded + +OUTPUT FORMAT: +- Produce a single coherent summary as plain text +- Preserve both user and assistant context +- When a user changed a decision, keep only the final decision with brief \ +context on why it changed +- Write in a neutral, factual tone +- If the summary is still too long after removing noise, prefer trimming \ +assistant explanations over user decisions and requirements + +COMPRESSION STRATEGY: +- Compress as much as possible without losing information that could affect \ +future responses +- Casual or simple exchanges can be heavily compressed +- Technical discussions with specific requirements should retain more detail +- Do NOT target a fixed compression ratio +- The output can be as short or as long as needed to preserve all relevant context +""" diff --git a/src/agent/streaming/emitter.py b/src/agent/streaming/emitter.py index 335d9c8..ef4ee1e 100644 --- a/src/agent/streaming/emitter.py +++ b/src/agent/streaming/emitter.py @@ -51,6 +51,11 @@ def emit_update(message: str, *, extra: dict[str, Any] | None = None) -> None: _emit("assistant.update", data) +def emit_thinking(message: str) -> None: + """Emit model reasoning (thinking). Best-effort: provider-dependent.""" + _emit("assistant.thinking", {"message": message}) + + def emit_plan(steps: list[str], planned_filepath: str = "") -> None: """Emit a structured plan event.""" _emit("plan.created", {"steps": steps, "planned_filepath": planned_filepath}) @@ -61,6 +66,15 @@ def emit_title(title: str) -> None: _emit("title.generated", {"title": title}) +def emit_yaml_draft(filepath: str, content: str) -> None: + """Emit the validated YAML draft before it is written to disk. + + Lets the client render a preview/diff ahead of the write-approval + permission prompt. + """ + _emit("yaml.draft", {"filepath": filepath, "content": content}) + + def emit_validation_result( *, passed: bool, diff --git a/src/agent/streaming/events.py b/src/agent/streaming/events.py index 2445dd8..8a676f7 100644 --- a/src/agent/streaming/events.py +++ b/src/agent/streaming/events.py @@ -14,13 +14,16 @@ "run.failed", "run.cancelled", "assistant.update", + "assistant.thinking", "plan.created", "title.generated", + "yaml.draft", "tool.started", "tool.completed", "tool.failed", "tool.interrupted", - "input.required", + "permission.required", + "clarification.required", "validation.result", "llm.error", ] diff --git a/src/agent/streaming/runner.py b/src/agent/streaming/runner.py index cfb43b9..e072131 100644 --- a/src/agent/streaming/runner.py +++ b/src/agent/streaming/runner.py @@ -43,19 +43,32 @@ def _interrupt_event(update_data: object) -> dict[str, Any] | None: if not isinstance(payload, dict): payload = {"message": str(payload)} + question_type = payload.get("type", "clarification") metadata = payload.get("metadata", {}) if not isinstance(metadata, dict): metadata = {} - return { - "type": "input.required", - "node": "human_interaction", - "data": { - "question": payload.get("message", "Clarification needed"), - "question_type": payload.get("type", "clarification"), - "metadata": metadata, - }, - } + if question_type == "permission": + return { + "type": "permission.required", + "node": "human_interaction", + "data": { + "request_id": metadata.get("request_id"), + "command": metadata.get("command"), + "cmd_type": metadata.get("cmd_type"), + "message": payload.get("message", "Permission required"), + "source": metadata.get("source"), + }, + } + else: + return { + "type": "clarification.required", + "node": "human_interaction", + "data": { + "question": payload.get("message", "Clarification needed"), + "source": payload.get("source", "unknown"), + }, + } def _event_line( diff --git a/src/agent/tools/permission/protocol.py b/src/agent/tools/permission/protocol.py index df5fabf..e4598ab 100644 --- a/src/agent/tools/permission/protocol.py +++ b/src/agent/tools/permission/protocol.py @@ -33,3 +33,29 @@ def parse_allow_deny(answer: str) -> bool | None: if normalized.startswith(("deny", "no", "n", "reject", "blocked")): return False return None + + +def permission_answer_for_request( + state: dict, + request_id: str | None, +) -> str: + """Return the user's allow/deny answer for a permission request_id, or "". + + Looks up ``human_clarifications`` (records written by human_interaction_node) + for the most recent permission record matching ``request_id``. + + A missing or unmatched request_id forces "" so callers re-prompt explicitly. + Shared by analyzer_node and writer_node. + """ + clarifications = state.get("human_clarifications", []) or [] + + if request_id: + for record in reversed(clarifications): + if record.get("type") != "permission": + continue + if str(record.get("request_id", "")) == str(request_id): + return str(record.get("answer", "")) + return "" + + # if request_id is missing, force explicit re-prompt. + return "" diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..31b038a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,9 @@ +"""Root pytest configuration for all tests.""" + +from pathlib import Path +from dotenv import load_dotenv + +# Load .env file from project root for integration tests +env_file = Path(__file__).parent.parent / ".env" +if env_file.exists(): + load_dotenv(env_file) diff --git a/tests/integration/agent/api/test_compaction.py b/tests/integration/agent/api/test_compaction.py new file mode 100644 index 0000000..094d46f --- /dev/null +++ b/tests/integration/agent/api/test_compaction.py @@ -0,0 +1,94 @@ +"""Integration tests for the chat compaction API route.""" + +from __future__ import annotations + +import pytest +import httpx +from agent.api.api import app +from agent.services.compaction import compaction as compaction_service + + +class FakeLLM: + """Records the messages passed to ainvoke and returns a canned summary.""" + + def __init__(self) -> None: + self.captured: list = [] + + async def ainvoke(self, messages, **kwargs): + self.captured = list(messages) + return type("Response", (), {"content": "COMPACTED SUMMARY"}) + + +@pytest.mark.asyncio +async def test_compact_endpoint_builds_ordered_tagged_messages( + monkeypatch: pytest.MonkeyPatch, +): + fake_llm = FakeLLM() + monkeypatch.setattr(compaction_service, "build_llm", lambda _: fake_llm) + + llm_config = { + "provider": "openai", + "model": "gpt-4o", + "api_key": "sk-test", + "temperature": 0, + } + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/compact", + json={ + "messages": [ + {"role": "user", "content": "build a Flask API"}, + {"role": "assistant", "content": "setting up Flask"}, + ], + "last_summary": "PREVIOUS SUMMARY TEXT", + "llm": llm_config, + }, + ) + + assert response.status_code == 200 + assert response.json() == {"summary": "COMPACTED SUMMARY"} + + # System prompt leads, prior summary becomes a tagged AIMessage, + # then each turn maps to its native message type in order. + types = [type(m).__name__ for m in fake_llm.captured] + assert types == ["SystemMessage", "AIMessage", "HumanMessage", "AIMessage"] + + assert "[CONVERSATION SUMMARY]" in fake_llm.captured[1].content + assert "PREVIOUS SUMMARY TEXT" in fake_llm.captured[1].content + assert fake_llm.captured[2].content == "build a Flask API" + assert fake_llm.captured[3].content == "setting up Flask" + + +@pytest.mark.asyncio +async def test_compact_endpoint_omits_summary_when_empty( + monkeypatch: pytest.MonkeyPatch, +): + fake_llm = FakeLLM() + monkeypatch.setattr(compaction_service, "build_llm", lambda _: fake_llm) + + import httpx + + llm_config = { + "provider": "openai", + "model": "gpt-4o", + "api_key": "sk-test", + "temperature": 0, + } + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/compact", + json={ + "messages": [{"role": "user", "content": "hello"}], + "llm": llm_config, + }, + ) + + assert response.status_code == 200 + + # No last_summary -> no tagged AIMessage, only system + the single turn. + types = [type(m).__name__ for m in fake_llm.captured] + assert types == ["SystemMessage", "HumanMessage"] diff --git a/tests/integration/agent/api/test_streaming_routes.py b/tests/integration/agent/api/test_streaming_routes.py index f06d812..a997c75 100644 --- a/tests/integration/agent/api/test_streaming_routes.py +++ b/tests/integration/agent/api/test_streaming_routes.py @@ -114,6 +114,7 @@ async def test_generate_stream_endpoint_returns_ndjson_headers_and_events( "api_key": "test-key", "base_url": None, "temperature": 0, + "reasoning_effort": "none", }, } } @@ -140,7 +141,7 @@ async def test_generate_stream_endpoint_returns_ndjson_headers_and_events( @pytest.mark.asyncio -async def test_resume_stream_endpoint_uses_command_and_skips_run_started( +async def test_resume_permission_endpoint_validates_request_id( monkeypatch: pytest.MonkeyPatch, ): fake_graph = FakeGraph( @@ -149,23 +150,26 @@ async def test_resume_stream_endpoint_uses_command_and_skips_run_started( "type": "custom", "data": { "type": "assistant.update", - "node": "planner", - "data": {"message": "Continuing with the approved command."}, + "node": "analyzer", + "data": {"message": "Command approved, continuing."}, }, } ] ) - # Patch BOTH bindings: the runner module global (used by stream_graph) - # and the route module's own import (used by resume_stream's get_state). + fake_graph._state = _FakeState({ + "pending_permission_request": {"request_id": "req-123", "command": "mkdir build"}, + "pending_question": {"type": "permission"}, + }) monkeypatch.setattr(streaming_runner, "graph", fake_graph) monkeypatch.setattr(streaming_route, "graph", fake_graph) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( - "/generate/thread-123/stream/resume", + "/generate/thread-123/resume/permission", json={ - "response": "allow", + "decision": "allow", + "request_id": "req-123", "llm": { "provider": "openai", "model": "gpt-4o", @@ -176,26 +180,93 @@ async def test_resume_stream_endpoint_uses_command_and_skips_run_started( ) assert response.status_code == 200 - assert response.headers["content-type"].startswith("application/x-ndjson") assert response.headers["x-thread-id"] == "thread-123" - assert response.headers["x-run-id"].startswith("run_") assert getattr(fake_graph.graph_input, "resume") == "allow" - assert fake_graph.config == { - "configurable": { - "thread_id": "thread-123", - "llm_config": { - "provider": "openai", - "model": "gpt-4o", - "api_key": "test-key", - "base_url": None, - "temperature": 0, + + events = _parse_ndjson(response) + assert len(events) == 1 + assert events[0]["type"] == "assistant.update" + + +@pytest.mark.asyncio +async def test_resume_permission_rejects_mismatched_request_id( + monkeypatch: pytest.MonkeyPatch, +): + fake_graph = FakeGraph([]) + fake_graph._state = _FakeState({ + "pending_permission_request": {"request_id": "req-123", "command": "mkdir build"}, + "pending_question": {"type": "permission"}, + }) + monkeypatch.setattr(streaming_runner, "graph", fake_graph) + monkeypatch.setattr(streaming_route, "graph", fake_graph) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/generate/thread-123/resume/permission", + json={ + "decision": "allow", + "request_id": "req-wrong", + "llm": { + "provider": "openai", + "model": "gpt-4o", + "api_key": "test-key", + "temperature": 0, + }, }, - } - } + ) + + assert response.status_code == 409 + detail = response.json()["detail"] + assert "Request ID mismatch" in detail + assert "req-123" in detail + assert "req-wrong" in detail + + +@pytest.mark.asyncio +async def test_resume_clarification_endpoint_accepts_answer( + monkeypatch: pytest.MonkeyPatch, +): + fake_graph = FakeGraph( + [ + { + "type": "custom", + "data": { + "type": "assistant.update", + "node": "planner", + "data": {"message": "Using Python 3.11 as specified."}, + }, + } + ] + ) + fake_graph._state = _FakeState({ + "pending_question": {"type": "clarification", "message": "What Python version?"}, + }) + monkeypatch.setattr(streaming_runner, "graph", fake_graph) + monkeypatch.setattr(streaming_route, "graph", fake_graph) + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/generate/thread-456/resume/clarification", + json={ + "answer": "Python 3.11", + "llm": { + "provider": "openai", + "model": "gpt-4o", + "api_key": "test-key", + "temperature": 0, + }, + }, + ) + + assert response.status_code == 200 + assert response.headers["x-thread-id"] == "thread-456" + assert getattr(fake_graph.graph_input, "resume") == "Python 3.11" events = _parse_ndjson(response) - assert [event["seq"] for event in events] == [1] - assert [event["type"] for event in events] == ["assistant.update"] + assert len(events) == 1 + assert events[0]["type"] == "assistant.update" @pytest.mark.asyncio diff --git a/tests/integration/agent/graph/subgraphs/test_analyzer.py b/tests/integration/agent/graph/subgraphs/test_analyzer.py index 6e48c03..00fca8c 100644 --- a/tests/integration/agent/graph/subgraphs/test_analyzer.py +++ b/tests/integration/agent/graph/subgraphs/test_analyzer.py @@ -11,11 +11,10 @@ import pytest import time -from agent.graph.subgraphs.analyzer import run_analyzer +from agent.graph.subgraphs.analyzer import run_analyzer_streaming from tests.integration.helpers import get_test_llm - @pytest.fixture(autouse=True) def _rate_limit_delay(): """Automatically sleep before every test to prevent LLM Tokens-Per-Minute rate limits (429 errors).""" @@ -47,7 +46,7 @@ def python_flask_project(tmp_path): "COPY requirements.txt .\n" "RUN pip install -r requirements.txt\n" "COPY . .\n" - "CMD [\"python\", \"app.py\"]\n" + 'CMD ["python", "app.py"]\n' ) src = tmp_path / "src" src.mkdir() @@ -119,24 +118,27 @@ def go_project(tmp_path): class TestAnalyzerDetectsLanguage: """The agent should correctly identify the programming language.""" - def test_detects_python(self, python_flask_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_detects_python(self, python_flask_project): + result = await run_analyzer_streaming( str(python_flask_project), "What programming language is this project written in?", llm=get_test_llm(), ) assert "python" in result["answer"].lower() - def test_detects_node(self, node_express_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_detects_node(self, node_express_project): + result = await run_analyzer_streaming( str(node_express_project), "What programming language is this project written in?", llm=get_test_llm(), ) assert "javascript" in result["answer"].lower() or "node" in result["answer"].lower() - def test_detects_go(self, go_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_detects_go(self, go_project): + result = await run_analyzer_streaming( str(go_project), "What programming language is this project written in?", llm=get_test_llm(), @@ -151,32 +153,36 @@ def test_detects_go(self, go_project): class TestAnalyzerDetectsDependencies: """The agent should find frameworks and dependencies.""" - def test_finds_flask(self, python_flask_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_flask(self, python_flask_project): + result = await run_analyzer_streaming( str(python_flask_project), "What web framework does this project use?", llm=get_test_llm(), ) assert "flask" in result["answer"].lower() - def test_finds_express(self, node_express_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_express(self, node_express_project): + result = await run_analyzer_streaming( str(node_express_project), "What web framework does this project use?", llm=get_test_llm(), ) assert "express" in result["answer"].lower() - def test_finds_test_framework(self, python_flask_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_test_framework(self, python_flask_project): + result = await run_analyzer_streaming( str(python_flask_project), "What test framework is configured in this project?", llm=get_test_llm(), ) assert "pytest" in result["answer"].lower() - def test_finds_node_test_runner(self, node_express_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_node_test_runner(self, node_express_project): + result = await run_analyzer_streaming( str(node_express_project), "What test runner does this project use?", llm=get_test_llm(), @@ -191,16 +197,18 @@ def test_finds_node_test_runner(self, node_express_project): class TestAnalyzerDetectsConfig: """The agent should detect build tools and deployment config.""" - def test_finds_dockerfile(self, python_flask_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_dockerfile(self, python_flask_project): + result = await run_analyzer_streaming( str(python_flask_project), "Does this project have a Dockerfile? If yes, what base image does it use?", llm=get_test_llm(), ) assert "python" in result["answer"].lower() - def test_finds_npm_scripts(self, node_express_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_npm_scripts(self, node_express_project): + result = await run_analyzer_streaming( str(node_express_project), "What npm scripts are defined in this project?", llm=get_test_llm(), @@ -208,8 +216,9 @@ def test_finds_npm_scripts(self, node_express_project): assert "start" in result["answer"].lower() assert "test" in result["answer"].lower() - def test_finds_makefile_targets(self, go_project): - result = run_analyzer( + @pytest.mark.asyncio + async def test_finds_makefile_targets(self, go_project): + result = await run_analyzer_streaming( str(go_project), "What build targets are defined in the Makefile?", llm=get_test_llm(), diff --git a/tests/integration/agent/graph/test_hitl_flow.py b/tests/integration/agent/graph/test_hitl_flow.py index d64af81..bca1535 100644 --- a/tests/integration/agent/graph/test_hitl_flow.py +++ b/tests/integration/agent/graph/test_hitl_flow.py @@ -65,15 +65,34 @@ def _events_to_response(events: list[dict]) -> SimpleNamespace: events=events, ) - input_events = [event for event in events if event.get("type") == "input.required"] - if input_events: - data = input_events[-1].get("data", {}) + permission_events = [event for event in events if event.get("type") == "permission.required"] + if permission_events: + data = permission_events[-1].get("data", {}) return SimpleNamespace( status="interrupted", question={ - "type": data.get("question_type", "clarification"), + "type": "permission", + "message": data.get("message", "Permission required"), + "metadata": { + "request_id": data.get("request_id"), + "command": data.get("command"), + "cmd_type": data.get("cmd_type"), + "source": data.get("source"), + }, + }, + errors=[], + events=events, + ) + + clarification_events = [event for event in events if event.get("type") == "clarification.required"] + if clarification_events: + data = clarification_events[-1].get("data", {}) + return SimpleNamespace( + status="interrupted", + question={ + "type": "clarification", "message": data.get("question", "Clarification needed"), - "metadata": data.get("metadata", {}), + "metadata": {}, }, errors=[], events=events, @@ -146,20 +165,27 @@ async def test_hitl_permission_pause_and_resume_with_real_llm(sample_project): if q_type == "permission": metadata = question.get("metadata") or {} - assert metadata.get("request_id") - user_reply = "allow" + request_id = metadata.get("request_id") + assert request_id, "Permission request must have request_id" + stream_response = await client.post( + f"/generate/{thread_id}/resume/permission", + json={ + "decision": "allow", + "request_id": request_id, + "llm": get_llm_config(), + } + ) elif q_type in {"clarification", "replan_limit"}: - user_reply = "Use sensible defaults and continue" + stream_response = await client.post( + f"/generate/{thread_id}/resume/clarification", + json={ + "answer": "Use sensible defaults and continue", + "llm": get_llm_config(), + } + ) else: pytest.fail(f"Unexpected interruption type: {q_type}") - stream_response = await client.post( - f"/generate/{thread_id}/stream/resume", - json={ - "response": user_reply, - "llm": get_llm_config(), - } - ) assert stream_response.status_code == 200 response = _events_to_response(await _collect_stream_events(stream_response)) @@ -187,7 +213,8 @@ async def test_hitl_permission_pause_and_resume_with_real_llm(sample_project): assert str(record.get("answer", "")).strip().lower().startswith("allow") -def test_hitl_normal_clarification_pause_and_resume_real_graph(sample_project): +@pytest.mark.asyncio +async def test_hitl_normal_clarification_pause_and_resume_real_graph(sample_project): graph = build_graph() thread_id = f"it-clarification-{uuid4()}" config = { @@ -211,9 +238,19 @@ def test_hitl_normal_clarification_pause_and_resume_real_graph(sample_project): "current_yaml_draft": "name: ci\non:\n push:\n", "planned_filepath": ".github/workflows/ci.yml", "validator_errors": ["simulated previous validation error"], + "analyzer_result": [ + { + "query": "project context", + "result": "Python project with pyproject.toml and README", + "requester": "planner", + "timestamp": "2024-01-01T00:00:00Z", + } + ], + "analyzer_query": "", + "query_requester": "", } - graph.invoke(initial_state, config) + await graph.ainvoke(initial_state, config) interrupted = graph.get_state(config) assert interrupted.next @@ -225,16 +262,57 @@ def test_hitl_normal_clarification_pause_and_resume_real_graph(sample_project): assert payload.get("source") == "planner" assert payload.get("message") + # Resume with user answer - may trigger additional clarifications from other nodes user_answer = "Keep the workflow minimal and stable" - graph.invoke(Command(resume=user_answer), config) - resumed = graph.get_state(config) - - assert not resumed.next - - clarifications = resumed.values.get("human_clarifications", []) or [] - assert clarifications - last = clarifications[-1] - assert last.get("type") == "replan_limit" - assert last.get("source") == "planner" - assert last.get("answer") == user_answer - assert resumed.values.get("pending_question", {}) == {} + await graph.ainvoke(Command(resume=user_answer), config) + + # Handle potential additional clarifications until completion + # (e.g., generator may ask follow-up questions) + max_clarification_rounds = 5 + clarification_count = 0 + + while True: + resumed = graph.get_state(config) + + # If workflow completed, break + if not resumed.next: + break + + # If still interrupted, provide a generic answer and continue + if resumed.next and resumed.tasks and resumed.tasks[0].interrupts: + clarification_count += 1 + if clarification_count > max_clarification_rounds: + pytest.fail( + f"Exceeded {max_clarification_rounds} clarification rounds. " + f"Last interruption: {resumed.tasks[0].interrupts[0].value}" + ) + + payload = resumed.tasks[0].interrupts[0].value + if payload.get("type") == "permission": + follow_up_answer = "allow" + else: + # Provide a sensible default answer to any clarification + follow_up_answer = "Use standard Python project defaults. No special requirements." + await graph.ainvoke(Command(resume=follow_up_answer), config) + else: + break + + # Verify the workflow completed + final_state = graph.get_state(config) + assert not final_state.next, "Workflow should have completed" + + # Verify the original clarification was recorded + clarifications = final_state.values.get("human_clarifications", []) or [] + assert clarifications, "Should have at least one clarification" + + # Find the replan_limit clarification (should be first) + replan_clarification = next( + (c for c in clarifications if c.get("type") == "replan_limit"), + None + ) + assert replan_clarification is not None, "Should have recorded the replan_limit clarification" + assert replan_clarification.get("source") == "planner" + assert replan_clarification.get("answer") == user_answer + + # Verify pending question was cleared + assert final_state.values.get("pending_question", {}) == {} diff --git a/tests/unit/agent/graph/nodes/test_writer.py b/tests/unit/agent/graph/nodes/test_writer.py index 63aab45..04de1e8 100644 --- a/tests/unit/agent/graph/nodes/test_writer.py +++ b/tests/unit/agent/graph/nodes/test_writer.py @@ -75,9 +75,6 @@ def test_disk_failure_emits_run_failed_with_draft(self, monkeypatch, tmp_path): lambda: (lambda payload: emitted.append(payload)), ) - # Replace the write_tool reference with a stub whose invoke fails. - # (Can't monkeypatch .invoke directly — StructuredTool is a frozen - # Pydantic model.) class _FailingTool: def invoke(self, payload): raise OSError("disk full") @@ -90,18 +87,141 @@ def invoke(self, payload): "planned_filepath": ".github/workflows/ci.yml", "pipeline_plan": [], "validator_warnings": [], + "pending_permission_request": { + "type": "permission", + "request_id": "test-req-123", + "source": "writer", + "command": str(tmp_path / ".github" / "workflows" / "ci.yml"), + "cmd_type": "modify", + "working_directory": str(tmp_path), + }, + "human_clarifications": [ + { + "type": "permission", + "request_id": "test-req-123", + "answer": "allow", + } + ], } - - with pytest.raises(OSError): writer_module.writer_node(state) - # Exactly one event emitted, and it carries the draft for recovery. assert len(emitted) == 1 payload = emitted[0] assert payload["type"] == "run.failed" assert payload["node"] == "writer" assert payload["data"]["detail"] == "OSError" assert payload["data"]["draft_content"] == "name: CI\non: push\n" - assert "ci.yml" in payload["data"]["resolved_path"] + + +class TestWriterNodePermissionFlow: + + def test_run1_emits_draft_and_requests_permission(self, monkeypatch, tmp_path): + emitted = [] + monkeypatch.setattr( + "agent.streaming.emitter.get_stream_writer", + lambda: (lambda payload: emitted.append(payload)), + ) + + state = { + "project_path": str(tmp_path), + "current_yaml_draft": "name: CI\non: push\n", + "planned_filepath": ".github/workflows/ci.yml", + } + + result = writer_module.writer_node(state) + + assert "pending_question" in result + assert result["pending_question"]["type"] == "permission" + assert result["pending_question"]["source"] == "writer" + assert "ci.yml" in result["pending_question"]["message"] + + assert "pending_permission_request" in result + permission = result["pending_permission_request"] + assert permission["type"] == "permission" + assert permission["source"] == "writer" + assert permission["cmd_type"] == "modify" + assert permission["request_id"] + assert "ci.yml" in permission["command"] + + yaml_draft_events = [e for e in emitted if e.get("type") == "yaml.draft"] + assert len(yaml_draft_events) == 1 + draft_event = yaml_draft_events[0] + assert draft_event["data"]["content"] == "name: CI\non: push\n" + assert "ci.yml" in draft_event["data"]["filepath"] + + assert not (tmp_path / ".github" / "workflows" / "ci.yml").exists() + + def test_run2_allow_writes_file(self, monkeypatch, tmp_path): + emitted = [] + monkeypatch.setattr( + "agent.streaming.emitter.get_stream_writer", + lambda: (lambda payload: emitted.append(payload)), + ) + + state = { + "project_path": str(tmp_path), + "current_yaml_draft": "name: CI\non: push\n", + "planned_filepath": ".github/workflows/ci.yml", + "pipeline_plan": ["step 1"], + "validator_warnings": [], + "pending_permission_request": { + "type": "permission", + "request_id": "test-req-456", + "source": "writer", + "command": str(tmp_path / ".github" / "workflows" / "ci.yml"), + "cmd_type": "modify", + "working_directory": str(tmp_path), + }, + "human_clarifications": [ + { + "type": "permission", + "request_id": "test-req-456", + "answer": "allow", + } + ], + } + + result = writer_module.writer_node(state) + + assert "output_path" in result + assert result["terminal_status"] == "success" + assert (tmp_path / ".github" / "workflows" / "ci.yml").exists() + assert (tmp_path / ".github" / "workflows" / "ci.yml").read_text() == "name: CI\non: push\n" + + run_completed = [e for e in emitted if e.get("type") == "run.completed"] + assert len(run_completed) == 1 + + def test_run2_deny_sets_write_rejected_flag(self, tmp_path): + state = { + "project_path": str(tmp_path), + "current_yaml_draft": "name: CI\non: push\n", + "planned_filepath": ".github/workflows/ci.yml", + "pending_permission_request": { + "type": "permission", + "request_id": "test-req-789", + "source": "writer", + "command": str(tmp_path / ".github" / "workflows" / "ci.yml"), + "cmd_type": "modify", + "working_directory": str(tmp_path), + }, + "human_clarifications": [ + { + "type": "permission", + "request_id": "test-req-789", + "answer": "deny", + } + ], + } + + result = writer_module.writer_node(state) + + assert result["write_rejected"] is True + assert result["replan_requested"] is True + assert result["replan_reason"] == "User rejected the generated workflow draft." + assert result["current_yaml_draft"] == "" + assert result["pending_question"] == {} + assert result["pending_permission_request"] == {} + + assert not (tmp_path / ".github" / "workflows" / "ci.yml").exists() diff --git a/tests/unit/agent/graph/subgraphs/test_analyzer.py b/tests/unit/agent/graph/subgraphs/test_analyzer.py index a74ed60..ad51965 100644 --- a/tests/unit/agent/graph/subgraphs/test_analyzer.py +++ b/tests/unit/agent/graph/subgraphs/test_analyzer.py @@ -1,5 +1,9 @@ """Unit tests for analyzer permission HITL behavior.""" +import asyncio + +import pytest + from agent.graph.subgraphs import analyzer as analyzer_module @@ -24,18 +28,19 @@ def _config() -> dict: "model": "gpt-4o", "api_key": "test-key", "temperature": 0, + "reasoning_effort": "none", } } } def test_returns_empty_when_no_query(): - result = analyzer_module.analyzer_node(_state(analyzer_query=""), _config()) + result = asyncio.run(analyzer_module.analyzer_node(_state(analyzer_query=""), _config())) assert result == {} def test_maps_permission_signal_to_pending_question(monkeypatch): - def fake_run_analyzer(**kwargs): + async def fake_run_analyzer(**kwargs): return { "answer": "", "permission_request": { @@ -48,9 +53,9 @@ def fake_run_analyzer(**kwargs): }, } - monkeypatch.setattr(analyzer_module, "run_analyzer", fake_run_analyzer) + monkeypatch.setattr(analyzer_module, "run_analyzer_streaming", fake_run_analyzer) - result = analyzer_module.analyzer_node(_state(), _config()) + result = asyncio.run(analyzer_module.analyzer_node(_state(), _config())) assert result["pending_question"]["type"] == "permission" assert result["pending_question"]["source"] == "analyzer" @@ -70,7 +75,7 @@ def test_denied_permission_clears_query_and_records_result(): ], ) - result = analyzer_module.analyzer_node(state, _config()) + result = asyncio.run(analyzer_module.analyzer_node(state, _config())) assert result["analyzer_query"] == "" assert result["pending_permission_request"] == {} @@ -81,14 +86,14 @@ def test_denied_permission_clears_query_and_records_result(): def test_allowed_permission_retries_with_preapproved_command(monkeypatch): captured = {} - def fake_run_analyzer(**kwargs): + async def fake_run_analyzer(**kwargs): captured["approved_commands"] = kwargs.get("approved_commands") return { "answer": "Repository uses Poetry", "permission_request": None, } - monkeypatch.setattr(analyzer_module, "run_analyzer", fake_run_analyzer) + monkeypatch.setattr(analyzer_module, "run_analyzer_streaming", fake_run_analyzer) state = _state( pending_permission_request={"request_id": "req-1", "command": "mkdir build", "cmd_type": "modify"}, @@ -101,7 +106,7 @@ def fake_run_analyzer(**kwargs): ], ) - result = analyzer_module.analyzer_node(state, _config()) + result = asyncio.run(analyzer_module.analyzer_node(state, _config())) assert captured["approved_commands"] == {"mkdir build"} assert result["analyzer_query"] == "" @@ -115,14 +120,14 @@ def fake_run_analyzer(**kwargs): def test_permission_answer_uses_matching_request_id(monkeypatch): captured = {} - def fake_run_analyzer(**kwargs): + async def fake_run_analyzer(**kwargs): captured["approved_commands"] = kwargs.get("approved_commands") return { "answer": "Repository uses Poetry", "permission_request": None, } - monkeypatch.setattr(analyzer_module, "run_analyzer", fake_run_analyzer) + monkeypatch.setattr(analyzer_module, "run_analyzer_streaming", fake_run_analyzer) state = _state( pending_permission_request={"request_id": "req-new", "command": "mkdir build", "cmd_type": "modify"}, @@ -132,7 +137,7 @@ def fake_run_analyzer(**kwargs): ], ) - result = analyzer_module.analyzer_node(state, _config()) + result = asyncio.run(analyzer_module.analyzer_node(state, _config())) assert "approved_commands" not in captured assert result["analyzer_query"] == "" @@ -141,10 +146,10 @@ def fake_run_analyzer(**kwargs): def test_permission_mismatch_request_id_reprompts_instead_of_fallback(monkeypatch): - def fake_run_analyzer(**kwargs): - raise AssertionError("run_analyzer should not be called when decision is unresolved") + async def fake_run_analyzer(**kwargs): + raise AssertionError("run_analyzer_streaming should not be called when decision is unresolved") - monkeypatch.setattr(analyzer_module, "run_analyzer", fake_run_analyzer) + monkeypatch.setattr(analyzer_module, "run_analyzer_streaming", fake_run_analyzer) state = _state( pending_permission_request={"request_id": "req-current", "command": "mkdir build", "cmd_type": "modify"}, @@ -153,7 +158,7 @@ def fake_run_analyzer(**kwargs): ], ) - result = analyzer_module.analyzer_node(state, _config()) + result = asyncio.run(analyzer_module.analyzer_node(state, _config())) assert result["pending_question"]["type"] == "permission" assert "allow' or 'deny'" in result["pending_question"]["message"] @@ -162,18 +167,21 @@ def fake_run_analyzer(**kwargs): def test_run_analyzer_returns_degraded_answer_on_generic_exception(monkeypatch): """A non-permission error (e.g. transient LLM failure) must not crash the - analyzer — it returns a degraded answer the planner/generator can use.""" + analyzer -- it returns a degraded answer the planner/generator can use.""" - def _raise(self, payload): + async def _raise(*a, **k): raise RuntimeError("simulated provider outage") + yield # Make it an async generator - fake_compiled = type("FakeAgent", (), {"invoke": _raise})() + fake_compiled = type("FakeAgent", (), {"astream": _raise})() monkeypatch.setattr( analyzer_module, "build_analyzer_agent", lambda *a, **k: fake_compiled ) - result = analyzer_module.run_analyzer( - project_path="d:/repo", query="inspect", llm=object() + result = asyncio.run( + analyzer_module.run_analyzer_streaming( + project_path="d:/repo", query="inspect", llm=object() + ) ) assert result["permission_request"] is None diff --git a/tests/unit/agent/llm/test_resilience.py b/tests/unit/agent/llm/test_resilience.py index 4add8d3..9643103 100644 --- a/tests/unit/agent/llm/test_resilience.py +++ b/tests/unit/agent/llm/test_resilience.py @@ -134,18 +134,18 @@ def _make_chain(llm_mock, max_retries=2): class TestBuildResilientLlm: def test_success_path_returns_value(self): - """Happy path: invoke returns the parsed value with no retry.""" + """Happy path: invoke returns tuple of (parsed, raw).""" inner = MagicMock() inner.invoke.return_value = _Schema(answer="ok") llm_mock = MagicMock() llm_mock.with_structured_output.return_value = inner chain = _make_chain(llm_mock) - result = chain.invoke([]) + parsed, raw = chain.invoke([]) - assert result.answer == "ok" + assert parsed.answer == "ok" llm_mock.with_structured_output.assert_called_once_with( - _Schema, method="function_calling" + _Schema, method="function_calling", include_raw=True ) def test_retries_on_retryable_exception_then_succeeds(self, monkeypatch): @@ -171,9 +171,9 @@ def _invoke(msgs, config=None, **kw): llm_mock.with_structured_output.return_value = inner chain = resilience.build_resilient_llm(llm_mock, _Schema, max_retries=2) - result = chain.invoke([]) + parsed, raw = chain.invoke([]) - assert result.answer == "recovered" + assert parsed.answer == "recovered" assert call_count == 2 assert len(emitted) == 1 assert emitted[0]["category"] == "parse" diff --git a/tests/unit/agent/streaming/test_events.py b/tests/unit/agent/streaming/test_events.py index e25f412..fa56c9b 100644 --- a/tests/unit/agent/streaming/test_events.py +++ b/tests/unit/agent/streaming/test_events.py @@ -32,13 +32,16 @@ def test_event_type_includes_expected_public_events() -> None: "run.failed", "run.cancelled", "assistant.update", + "assistant.thinking", "plan.created", "title.generated", + "yaml.draft", "tool.started", "tool.completed", "tool.failed", "tool.interrupted", - "input.required", + "permission.required", + "clarification.required", "validation.result", "llm.error", } diff --git a/tests/unit/agent/streaming/test_runner.py b/tests/unit/agent/streaming/test_runner.py index bdc3d9f..d7c723d 100644 --- a/tests/unit/agent/streaming/test_runner.py +++ b/tests/unit/agent/streaming/test_runner.py @@ -72,13 +72,13 @@ async def test_stream_graph_sequences_redacts_and_converts_interrupt(monkeypatch assert [line["type"] for line in lines] == [ "run.started", "assistant.update", - "input.required", + "clarification.required", ] assert lines[0]["data"] == {"run_id": "run-1", "thread_id": "thread-1"} assert lines[1]["data"]["api_key"] == "[REDACTED]" assert lines[2]["node"] == "human_interaction" assert lines[2]["data"]["question"] == "Which deploy target?" - assert lines[2]["data"]["metadata"]["token"] == "[REDACTED]" + assert lines[2]["data"]["source"] == "unknown" @pytest.mark.asyncio