Implement chat compaction API and separate resume endpoints

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__":

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)

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,

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",
 ]

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")

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).")

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())

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)
 

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"