From 2b91537b8dab3ff1e6b08b67542a5f68ab7ae56c Mon Sep 17 00:00:00 2001 From: Test User Date: Thu, 29 Jan 2026 22:44:26 -0700 Subject: [PATCH 1/3] feat(cli): add `cf work follow` command for real-time execution streaming Implements live streaming of task execution output, enabling users to attach to running tasks and view output in real-time. Components: - core/streaming.py: File-based streaming infrastructure - RunOutputLogger: Writes agent output to .codeframe/runs//output.log - tail_run_output(): Polling-based file tailing generator - get_latest_lines(): Buffered output retrieval for --tail flag - cli/app.py: New `cf work follow` command - Attach to running tasks and stream output - --tail N: Show last N lines before streaming - Shows completion status for finished runs - Graceful handling of Ctrl+C - core/agent.py: Enhanced _verbose_print() to write to log file - Dual-write pattern: stdout (when verbose) + log file (always) - Enables following non-verbose runs - core/runtime.py: Creates output logger for each run - Passes logger to Agent for output capture Tests: 37 new tests covering streaming, CLI, and agent integration Closes #308 --- CLAUDE.md | 3 + codeframe/cli/app.py | 170 +++++++++++++++ codeframe/core/agent.py | 18 +- codeframe/core/runtime.py | 10 + codeframe/core/streaming.py | 220 +++++++++++++++++++ tests/cli/test_v2_cli_integration.py | 74 +++++++ tests/cli/test_work_follow.py | 246 +++++++++++++++++++++ tests/core/test_agent_streaming.py | 281 ++++++++++++++++++++++++ tests/core/test_streaming.py | 311 +++++++++++++++++++++++++++ 9 files changed, 1332 insertions(+), 1 deletion(-) create mode 100644 codeframe/core/streaming.py create mode 100644 tests/cli/test_work_follow.py create mode 100644 tests/core/test_agent_streaming.py create mode 100644 tests/core/test_streaming.py diff --git a/CLAUDE.md b/CLAUDE.md index b8680c2c..7e995c23 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -97,6 +97,7 @@ codeframe/ │ ├── diagnostics.py # Failed task analysis │ ├── diagnostic_agent.py # AI-powered task diagnosis │ ├── credentials.py # API key and credential management +│ ├── streaming.py # Real-time output streaming for cf work follow │ └── ... ├── adapters/ │ └── llm/ # LLM provider adapters @@ -265,6 +266,8 @@ cf work start --execute --verbose # With detailed output cf work start --execute --dry-run # Preview changes cf work stop # Cancel stale run cf work resume # Resume blocked work +cf work follow # Stream real-time output +cf work follow --tail 50 # Show last 50 lines then stream # Batch execution (multiple tasks) cf work batch run ... # Execute multiple tasks diff --git a/codeframe/cli/app.py b/codeframe/cli/app.py index b7a23a3d..bc2347d5 100644 --- a/codeframe/cli/app.py +++ b/codeframe/cli/app.py @@ -2409,6 +2409,176 @@ def work_update_description( raise typer.Exit(1) +@work_app.command("follow") +def work_follow( + task_id: str = typer.Argument(..., help="Task ID to follow (can be partial)"), + workspace_path: Optional[Path] = typer.Option( + None, + "--workspace", + "-w", + help="Workspace path (defaults to current directory)", + ), + tail: Optional[int] = typer.Option( + None, + "--tail", + "-n", + help="Show last N lines of buffered output before streaming", + ), + timeout: Optional[float] = typer.Option( + None, + "--timeout", + "-t", + help="Maximum seconds to wait for output (for testing)", + ), +) -> None: + """Follow real-time execution output of a running task. + + Attaches to an active task execution and streams output as it happens. + Shows buffered output when attaching to already-running executions. + + If the task has a completed run (no active run), shows the final output. + + Example: + cf work follow abc123 # Follow from current point + cf work follow abc123 --tail 50 # Show last 50 lines then stream + """ + from codeframe.core.workspace import get_workspace + from codeframe.core import tasks as tasks_module, runtime + from codeframe.core.streaming import ( + get_latest_lines_with_count, + tail_run_output, + run_output_exists, + ) + + path = workspace_path or Path.cwd() + + try: + workspace = get_workspace(path) + + # Find task by partial ID + all_tasks = tasks_module.list_tasks(workspace) + matching = [t for t in all_tasks if t.id.startswith(task_id)] + + if not matching: + console.print(f"[red]Error:[/red] No task found matching '{task_id}'") + raise typer.Exit(1) + + if len(matching) > 1: + console.print(f"[red]Error:[/red] Multiple tasks match '{task_id}':") + for t in matching[:5]: + console.print(f" {t.id[:8]} - {t.title}") + raise typer.Exit(1) + + task = matching[0] + + # Get active run for task + active_run = runtime.get_active_run(workspace, task.id) + + if not active_run: + # Check for recent completed/failed runs + recent_runs = runtime.list_runs(workspace, task_id=task.id, limit=1) + + if recent_runs: + last_run = recent_runs[0] + + # Status color + status_color = { + runtime.RunStatus.COMPLETED: "green", + runtime.RunStatus.FAILED: "red", + runtime.RunStatus.BLOCKED: "yellow", + }.get(last_run.status, "white") + + console.print( + f"[{status_color}]Run {last_run.status.value}[/{status_color}] " + f"for task: {task.title}" + ) + + # Show final output if available + if run_output_exists(workspace, last_run.id): + console.print("\n[dim]--- Final output ---[/dim]") + lines, total = get_latest_lines_with_count( + workspace, last_run.id, count=tail or 50 + ) + if tail and total > tail: + console.print(f"[dim](showing last {tail} of {total} lines)[/dim]") + for line in lines: + console.print(line.rstrip()) + else: + console.print("[dim]No output captured for this run.[/dim]") + + raise typer.Exit(0) + else: + console.print(f"[yellow]No active run found for task:[/yellow] {task.title}") + console.print("[dim]Start a run with:[/dim]") + console.print(f" cf work start {task.id[:8]} --execute") + raise typer.Exit(1) + + # We have an active run - stream it + console.print(f"[blue]Following task:[/blue] {task.title}") + console.print(f"[dim]Run: {active_run.id[:8]} | Status: {active_run.status.value}[/dim]") + + # Show buffered output if requested + start_line = 0 + if tail: + lines, total = get_latest_lines_with_count( + workspace, active_run.id, count=tail + ) + if lines: + console.print(f"\n[dim]--- Buffered output (last {len(lines)} of {total} lines) ---[/dim]") + for line in lines: + console.print(f"[dim]{line.rstrip()}[/dim]") + console.print("[dim]--- Live output ---[/dim]\n") + start_line = total # Skip already-shown lines + + # Calculate max_wait for testing + max_wait = timeout if timeout else None + + # Terminal run statuses + TERMINAL_STATUSES = { + runtime.RunStatus.COMPLETED, + runtime.RunStatus.FAILED, + runtime.RunStatus.BLOCKED, + } + + try: + # Stream output + for line in tail_run_output( + workspace, + active_run.id, + since_line=start_line, + poll_interval=0.3, + max_wait=max_wait, + ): + console.print(line.rstrip()) + + # Check if run is still active (periodically) + current_run = runtime.get_run(workspace, active_run.id) + if current_run and current_run.status in TERMINAL_STATUSES: + # Show completion message + status_color = { + runtime.RunStatus.COMPLETED: "green", + runtime.RunStatus.FAILED: "red", + runtime.RunStatus.BLOCKED: "yellow", + }.get(current_run.status, "white") + + console.print( + f"\n[{status_color}]Run {current_run.status.value}[/{status_color}]" + ) + break + + except KeyboardInterrupt: + console.print("\n[yellow]Streaming interrupted[/yellow]") + console.print(f"[dim]Run is still active. Resume with: cf work follow {task.id[:8]}[/dim]") + raise typer.Exit(0) + + except FileNotFoundError: + console.print(f"[red]Error:[/red] No workspace found at {path}") + raise typer.Exit(1) + except ValueError as e: + console.print(f"[red]Error:[/red] {e}") + raise typer.Exit(1) + + # ============================================================================= # Batch execution commands (subcommand group: cf work batch ) # ============================================================================= diff --git a/codeframe/core/agent.py b/codeframe/core/agent.py index 6a235c1c..74484255 100644 --- a/codeframe/core/agent.py +++ b/codeframe/core/agent.py @@ -33,6 +33,7 @@ if TYPE_CHECKING: from codeframe.core.conductor import GlobalFixCoordinator + from codeframe.core.streaming import RunOutputLogger # Safe shell commands that can be executed without full shell interpretation SAFE_SHELL_COMMANDS = frozenset({ @@ -397,6 +398,7 @@ def __init__( debug: bool = False, verbose: bool = False, fix_coordinator: Optional["GlobalFixCoordinator"] = None, + output_logger: Optional["RunOutputLogger"] = None, ): """Initialize the agent. @@ -409,6 +411,7 @@ def __init__( debug: If True, write detailed debug log to workspace verbose: If True, print detailed progress to stdout fix_coordinator: Optional coordinator for global fixes (for parallel execution) + output_logger: Optional logger for streaming output to file (for cf work follow) """ self.workspace = workspace self.llm = llm_provider @@ -418,6 +421,7 @@ def __init__( self.debug = debug self.verbose = verbose self.fix_coordinator = fix_coordinator + self.output_logger = output_logger self.state = AgentState() self.context: Optional[TaskContext] = None @@ -433,10 +437,22 @@ def __init__( self._setup_debug_log() def _verbose_print(self, message: str) -> None: - """Print message only if verbose mode is enabled.""" + """Print message to stdout (if verbose) and to output log file. + + The output log file is always written to (if logger provided) to enable + streaming via `cf work follow`, even when verbose=False. + + Args: + message: Message to print/log + """ + # Print to stdout if verbose mode is enabled if self.verbose: print(message) + # Always write to output log if logger is provided (for cf work follow) + if self.output_logger: + self.output_logger.write(message + "\n") + def run(self, task_id: str) -> AgentState: """Run the agent on a task. diff --git a/codeframe/core/runtime.py b/codeframe/core/runtime.py index 53262991..b0aef690 100644 --- a/codeframe/core/runtime.py +++ b/codeframe/core/runtime.py @@ -604,6 +604,10 @@ def execute_agent( "verbose": verbose, }) + # Create output logger for streaming (cf work follow) + from codeframe.core.streaming import RunOutputLogger + output_logger = RunOutputLogger(workspace, run.id) + # Create event callback to emit workspace events and log def on_agent_event(event_type: str, data: dict) -> None: events.emit_for_workspace( @@ -626,6 +630,7 @@ def on_agent_event(event_type: str, data: dict) -> None: debug=debug, verbose=verbose, fix_coordinator=fix_coordinator, + output_logger=output_logger, ) state = agent.run(run.task_id) @@ -647,6 +652,7 @@ def on_agent_event(event_type: str, data: dict) -> None: on_event=on_agent_event, debug=debug, fix_coordinator=fix_coordinator, + output_logger=output_logger, ) state = agent.run(run.task_id) @@ -743,6 +749,7 @@ def on_agent_event(event_type: str, data: dict) -> None: on_event=on_agent_event, debug=debug, fix_coordinator=fix_coordinator, + output_logger=output_logger, ) state = agent.run(run.task_id) if debug: @@ -784,6 +791,9 @@ def on_agent_event(event_type: str, data: dict) -> None: elif state.status == AgentStatus.FAILED: fail_run(workspace, run.id) + # Close output logger + output_logger.close() + return state diff --git a/codeframe/core/streaming.py b/codeframe/core/streaming.py new file mode 100644 index 00000000..87002f24 --- /dev/null +++ b/codeframe/core/streaming.py @@ -0,0 +1,220 @@ +"""Streaming infrastructure for real-time execution output. + +This module provides file-based streaming for `cf work follow`: +- RunOutputLogger: Writes agent output to a log file +- tail_run_output: Tails a log file for real-time streaming +- get_latest_lines: Reads buffered output (for --tail N) + +Output files are stored at: .codeframe/runs//output.log + +This module is headless - no FastAPI or HTTP dependencies. +""" + +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterator, Optional + +from codeframe.core.workspace import Workspace + + +def get_run_output_path(workspace: Workspace, run_id: str) -> Path: + """Get the path for a run's output log file. + + Args: + workspace: Target workspace + run_id: Run identifier + + Returns: + Path to the output log file + """ + return workspace.repo_path / ".codeframe" / "runs" / run_id / "output.log" + + +def run_output_exists(workspace: Workspace, run_id: str) -> bool: + """Check if a run's output log exists. + + Args: + workspace: Target workspace + run_id: Run identifier + + Returns: + True if the output file exists + """ + return get_run_output_path(workspace, run_id).exists() + + +class RunOutputLogger: + """Logger that writes agent output to a file for streaming. + + This class is used by the Agent to write verbose output to a log file + that can be tailed by `cf work follow`. + + Usage: + with RunOutputLogger(workspace, run_id) as logger: + logger.write("Processing step 1...") + logger.write_timestamped("Step completed") + + The log file is flushed after each write to enable real-time streaming. + """ + + def __init__(self, workspace: Workspace, run_id: str): + """Initialize the logger. + + Args: + workspace: Target workspace + run_id: Run identifier + """ + self.workspace = workspace + self.run_id = run_id + self.log_path = get_run_output_path(workspace, run_id) + + # Ensure directory exists + self.log_path.parent.mkdir(parents=True, exist_ok=True) + + # Open file in append mode + self._file = open(self.log_path, "a", encoding="utf-8") + + def write(self, message: str) -> None: + """Write a message to the log file. + + The file is flushed after each write to enable real-time streaming. + + Args: + message: Message to write (should include newline if desired) + """ + self._file.write(message) + self._file.flush() + + def write_timestamped(self, message: str) -> None: + """Write a message with a timestamp prefix. + + Format: [HH:MM:SS] message + + Args: + message: Message to write + """ + timestamp = datetime.now(timezone.utc).strftime("%H:%M:%S") + self.write(f"[{timestamp}] {message}\n") + + def close(self) -> None: + """Close the log file.""" + if self._file and not self._file.closed: + self._file.close() + + def __enter__(self) -> "RunOutputLogger": + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + """Context manager exit - close the file.""" + self.close() + + +def get_latest_lines(workspace: Workspace, run_id: str, count: int) -> list[str]: + """Get the last N lines from a run's output log. + + Used by `cf work follow --tail N` to show buffered output. + + Args: + workspace: Target workspace + run_id: Run identifier + count: Number of lines to return + + Returns: + List of the last N lines (or fewer if file has less) + """ + lines, _ = get_latest_lines_with_count(workspace, run_id, count) + return lines + + +def get_latest_lines_with_count( + workspace: Workspace, run_id: str, count: int +) -> tuple[list[str], int]: + """Get the last N lines and total line count from a run's output log. + + Args: + workspace: Target workspace + run_id: Run identifier + count: Number of lines to return + + Returns: + Tuple of (last N lines, total line count) + """ + log_path = get_run_output_path(workspace, run_id) + + if not log_path.exists(): + return [], 0 + + try: + with open(log_path, "r", encoding="utf-8") as f: + all_lines = f.readlines() + + total = len(all_lines) + + if count >= total: + return all_lines, total + + return all_lines[-count:], total + + except Exception: + return [], 0 + + +def tail_run_output( + workspace: Workspace, + run_id: str, + since_line: int = 0, + poll_interval: float = 0.5, + max_iterations: Optional[int] = None, + max_wait: Optional[float] = None, +) -> Iterator[str]: + """Tail a run's output log file, yielding new lines. + + This generator polls the log file and yields new lines as they appear. + It's designed to be used with `cf work follow` for real-time streaming. + + Args: + workspace: Target workspace + run_id: Run identifier + since_line: Start after this line number (0-based) + poll_interval: How often to check for new lines (seconds) + max_iterations: Stop after this many poll iterations (for testing) + max_wait: Maximum total wait time in seconds (for testing) + + Yields: + Lines from the log file as they appear + """ + log_path = get_run_output_path(workspace, run_id) + current_line = since_line + iterations = 0 + start_time = time.time() + + while True: + # Check termination conditions + if max_iterations is not None and iterations >= max_iterations: + break + + if max_wait is not None and (time.time() - start_time) >= max_wait: + break + + # Check if file exists + if not log_path.exists(): + time.sleep(poll_interval) + iterations += 1 + continue + + try: + with open(log_path, "r", encoding="utf-8") as f: + all_lines = f.readlines() + + # Yield new lines + while current_line < len(all_lines): + yield all_lines[current_line] + current_line += 1 + + except Exception: + pass # File might be temporarily unavailable + + time.sleep(poll_interval) + iterations += 1 diff --git a/tests/cli/test_v2_cli_integration.py b/tests/cli/test_v2_cli_integration.py index 60a6df7c..ecc354d2 100644 --- a/tests/cli/test_v2_cli_integration.py +++ b/tests/cli/test_v2_cli_integration.py @@ -376,6 +376,80 @@ def test_start_stub(self, workspace_with_ready_tasks): assert result.exit_code == 0 assert "stub" in result.output.lower() or "completed" in result.output.lower() + def test_follow_no_run(self, workspace_with_ready_tasks): + """Follow should indicate no active run for task without run.""" + ws = create_or_load_workspace(workspace_with_ready_tasks) + task_list = tasks.list_tasks(ws, status=TaskStatus.READY) + assert len(task_list) > 0 + tid = task_list[0].id[:8] + result = runner.invoke( + app, ["work", "follow", tid, "-w", str(workspace_with_ready_tasks)] + ) + assert result.exit_code != 0 + assert "no active run" in result.output.lower() + + def test_follow_completed_run(self, workspace_with_ready_tasks): + """Follow should show output for completed runs.""" + from codeframe.core import runtime + from codeframe.core.streaming import RunOutputLogger + + ws = create_or_load_workspace(workspace_with_ready_tasks) + task_list = tasks.list_tasks(ws, status=TaskStatus.READY) + assert len(task_list) > 0 + task = task_list[0] + tid = task.id[:8] + + # Start run + run = runtime.start_task_run(ws, task.id) + + # Write some output + with RunOutputLogger(ws, run.id) as logger: + logger.write("Test output line\n") + + # Complete the run + runtime.complete_run(ws, run.id) + + result = runner.invoke( + app, ["work", "follow", tid, "-w", str(workspace_with_ready_tasks)] + ) + assert result.exit_code == 0 + assert "completed" in result.output.lower() + + def test_follow_with_tail(self, workspace_with_ready_tasks): + """Follow --tail should show last N lines.""" + from codeframe.core import runtime + from codeframe.core.streaming import RunOutputLogger + + ws = create_or_load_workspace(workspace_with_ready_tasks) + task_list = tasks.list_tasks(ws, status=TaskStatus.READY) + assert len(task_list) > 0 + task = task_list[0] + tid = task.id[:8] + + run = runtime.start_task_run(ws, task.id) + + # Write multiple lines + with RunOutputLogger(ws, run.id) as logger: + for i in range(10): + logger.write(f"Line {i}\n") + + runtime.complete_run(ws, run.id) + + result = runner.invoke( + app, ["work", "follow", tid, "--tail", "3", "-w", str(workspace_with_ready_tasks)] + ) + assert result.exit_code == 0 + # Should contain the last lines + assert "Line 9" in result.output or "Line 8" in result.output + + def test_follow_nonexistent_task(self, workspace_path): + """Follow should error for nonexistent task.""" + result = runner.invoke( + app, ["work", "follow", "nonexistent", "-w", str(workspace_path)] + ) + assert result.exit_code != 0 + assert "no task found" in result.output.lower() + # --------------------------------------------------------------------------- # 8. Batch commands diff --git a/tests/cli/test_work_follow.py b/tests/cli/test_work_follow.py new file mode 100644 index 00000000..5b12cf60 --- /dev/null +++ b/tests/cli/test_work_follow.py @@ -0,0 +1,246 @@ +"""Tests for the `cf work follow` CLI command. + +Tests for real-time execution streaming of individual task runs. +""" + +import pytest +from pathlib import Path +from typer.testing import CliRunner +from unittest.mock import patch +import threading +import time + +from codeframe.cli.app import app +from codeframe.core.workspace import create_or_load_workspace, Workspace +from codeframe.core import tasks + + +runner = CliRunner() + + +@pytest.fixture +def temp_workspace(tmp_path: Path) -> Workspace: + """Create a temporary workspace with a task.""" + workspace = create_or_load_workspace(tmp_path) + return workspace + + +@pytest.fixture +def task_with_run(temp_workspace: Workspace): + """Create a task with an active run.""" + from codeframe.core import runtime + + task = tasks.create(temp_workspace, title="Test task for follow") + run = runtime.start_task_run(temp_workspace, task.id) + return task, run, temp_workspace + + +class TestWorkFollowCommand: + """Tests for cf work follow command.""" + + def test_follow_requires_task_id(self, tmp_path: Path): + """Follow command should require a task ID argument.""" + result = runner.invoke(app, ["work", "follow"]) + assert result.exit_code != 0 + assert "Missing argument" in result.output or "Usage" in result.output + + def test_follow_shows_error_for_nonexistent_task(self, temp_workspace: Workspace): + """Should show error when task doesn't exist.""" + result = runner.invoke( + app, + ["work", "follow", "nonexistent", "--workspace", str(temp_workspace.repo_path)], + ) + assert result.exit_code != 0 + assert "No task found" in result.output or "Error" in result.output + + def test_follow_shows_error_for_no_active_run(self, temp_workspace: Workspace): + """Should show message when task has no active run.""" + task = tasks.create(temp_workspace, title="Task without run") + + result = runner.invoke( + app, + ["work", "follow", task.id[:8], "--workspace", str(temp_workspace.repo_path)], + ) + + # Should indicate no active run + assert "No active run" in result.output or "not running" in result.output.lower() + + def test_follow_shows_completed_run_output(self, temp_workspace: Workspace): + """Should show final output for completed runs.""" + from codeframe.core import runtime + from codeframe.core.streaming import RunOutputLogger + + # Create task and run + task = tasks.create(temp_workspace, title="Completed task") + run = runtime.start_task_run(temp_workspace, task.id) + + # Write some output + with RunOutputLogger(temp_workspace, run.id) as logger: + logger.write("Step 1 completed\n") + logger.write("Step 2 completed\n") + logger.write("Task finished successfully\n") + + # Complete the run + runtime.complete_run(temp_workspace, run.id) + + result = runner.invoke( + app, + ["work", "follow", task.id[:8], "--workspace", str(temp_workspace.repo_path)], + ) + + # Should show completion message + assert result.exit_code == 0 + assert "completed" in result.output.lower() or "finished" in result.output.lower() + + def test_follow_with_tail_shows_buffered_output(self, task_with_run): + """Should show last N lines when --tail is specified.""" + task, run, workspace = task_with_run + from codeframe.core.streaming import RunOutputLogger + from codeframe.core import runtime + + # Write output + with RunOutputLogger(workspace, run.id) as logger: + for i in range(10): + logger.write(f"Line {i}\n") + + # Complete the run so follow doesn't wait + runtime.complete_run(workspace, run.id) + + result = runner.invoke( + app, + [ + "work", "follow", task.id[:8], + "--tail", "3", + "--workspace", str(workspace.repo_path) + ], + ) + + assert result.exit_code == 0 + # Should show last 3 lines + assert "Line 7" in result.output or "Line 8" in result.output or "Line 9" in result.output + + def test_follow_streams_output_for_active_run(self, task_with_run): + """Should stream output while run is active.""" + task, run, workspace = task_with_run + from codeframe.core.streaming import RunOutputLogger + from codeframe.core import runtime + + # This test simulates real-time streaming + # Write output in a separate thread while follow is running + + output_collected = [] + + def writer_thread(): + time.sleep(0.2) + with RunOutputLogger(workspace, run.id) as logger: + logger.write("First line\n") + time.sleep(0.2) + logger.write("Second line\n") + time.sleep(0.2) + runtime.complete_run(workspace, run.id) + + thread = threading.Thread(target=writer_thread) + thread.start() + + result = runner.invoke( + app, + [ + "work", "follow", task.id[:8], + "--workspace", str(workspace.repo_path), + "--timeout", "3", # Short timeout for test + ], + ) + + thread.join(timeout=5) + + # Should have captured some output + assert "First line" in result.output or "Second line" in result.output + + def test_follow_handles_ctrl_c_gracefully(self, task_with_run): + """Should handle keyboard interrupt gracefully.""" + task, run, workspace = task_with_run + + # Mock KeyboardInterrupt during streaming + with patch('codeframe.core.streaming.tail_run_output') as mock_tail: + def raise_interrupt(): + yield "Test line\n" + raise KeyboardInterrupt() + + mock_tail.return_value = raise_interrupt() + + result = runner.invoke( + app, + [ + "work", "follow", task.id[:8], + "--workspace", str(workspace.repo_path), + ], + catch_exceptions=False, + ) + + # Should exit cleanly with interrupt message + assert "interrupt" in result.output.lower() or "cancelled" in result.output.lower() or result.exit_code == 0 + + +class TestWorkFollowOutput: + """Tests for follow output formatting.""" + + def test_output_includes_timestamps(self, task_with_run): + """Output should include timestamps when available.""" + task, run, workspace = task_with_run + from codeframe.core.streaming import RunOutputLogger + from codeframe.core import runtime + + with RunOutputLogger(workspace, run.id) as logger: + logger.write_timestamped("Step started") + + runtime.complete_run(workspace, run.id) + + result = runner.invoke( + app, + ["work", "follow", task.id[:8], "--workspace", str(workspace.repo_path)], + ) + + # Should show timestamp format [HH:MM:SS] + assert "[" in result.output and "]" in result.output + + def test_shows_task_info_on_attach(self, task_with_run): + """Should show task info when attaching to a run.""" + task, run, workspace = task_with_run + from codeframe.core import runtime + + runtime.complete_run(workspace, run.id) + + result = runner.invoke( + app, + ["work", "follow", task.id[:8], "--workspace", str(workspace.repo_path)], + ) + + # Should show task title or ID + assert task.title in result.output or task.id[:8] in result.output + + +class TestWorkFollowWithEvents: + """Tests for follow command with run completion.""" + + def test_follow_shows_run_completion(self, task_with_run): + """Should show completion message when run finishes.""" + task, run, workspace = task_with_run + from codeframe.core import runtime + from codeframe.core.streaming import RunOutputLogger + + # Write some output + with RunOutputLogger(workspace, run.id) as logger: + logger.write("Processing task...\n") + + runtime.complete_run(workspace, run.id) + + result = runner.invoke( + app, + [ + "work", "follow", task.id[:8], + "--workspace", str(workspace.repo_path), + ], + ) + + # Should show completion information + assert "completed" in result.output.lower() diff --git a/tests/core/test_agent_streaming.py b/tests/core/test_agent_streaming.py new file mode 100644 index 00000000..e468fcb7 --- /dev/null +++ b/tests/core/test_agent_streaming.py @@ -0,0 +1,281 @@ +"""Tests for agent integration with streaming output. + +These tests verify that the Agent class correctly writes verbose output +to both stdout and the run output log file for `cf work follow`. +""" + +import pytest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from codeframe.core.workspace import create_or_load_workspace, Workspace +from codeframe.core.streaming import RunOutputLogger + + +@pytest.fixture +def temp_workspace(tmp_path: Path) -> Workspace: + """Create a temporary workspace for testing.""" + return create_or_load_workspace(tmp_path) + + +@pytest.fixture +def mock_llm_provider(): + """Create a mock LLM provider.""" + provider = MagicMock() + provider.complete.return_value = MagicMock( + content='{"steps": [], "estimated_complexity": "low"}' + ) + provider.get_model.return_value = "claude-sonnet-4" + return provider + + +class TestAgentOutputLogger: + """Tests for Agent output logging integration.""" + + def test_agent_accepts_output_logger(self, temp_workspace: Workspace, mock_llm_provider): + """Agent should accept an optional output_logger parameter.""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + # Should not raise + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + output_logger=logger, + ) + + assert agent.output_logger is logger + logger.close() + + def test_agent_verbose_print_writes_to_logger( + self, temp_workspace: Workspace, mock_llm_provider + ): + """Agent _verbose_print should write to the output logger.""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=True, + output_logger=logger, + ) + + agent._verbose_print("Test message from agent") + + logger.close() + + # Check the log file contains the message + content = logger.log_path.read_text() + assert "Test message from agent" in content + + def test_agent_verbose_print_writes_both_stdout_and_file( + self, temp_workspace: Workspace, mock_llm_provider, capsys + ): + """Agent _verbose_print should write to both stdout and file when verbose=True.""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=True, + output_logger=logger, + ) + + agent._verbose_print("Test output message") + + logger.close() + + # Check stdout + captured = capsys.readouterr() + assert "Test output message" in captured.out + + # Check file + content = logger.log_path.read_text() + assert "Test output message" in content + + def test_agent_writes_to_file_even_when_not_verbose( + self, temp_workspace: Workspace, mock_llm_provider, capsys + ): + """Agent should write to file even when verbose=False (for follow command).""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=False, # Not verbose to stdout + output_logger=logger, + ) + + agent._verbose_print("Silent message") + + logger.close() + + # Should NOT be in stdout + captured = capsys.readouterr() + assert "Silent message" not in captured.out + + # But SHOULD be in file + content = logger.log_path.read_text() + assert "Silent message" in content + + def test_agent_without_logger_still_works( + self, temp_workspace: Workspace, mock_llm_provider, capsys + ): + """Agent should work normally without an output logger.""" + from codeframe.core.agent import Agent + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=True, + ) + + # Should not raise + agent._verbose_print("Test without logger") + + captured = capsys.readouterr() + assert "Test without logger" in captured.out + + +class TestRuntimeCreatesLogger: + """Tests for runtime creating output logger for runs.""" + + def test_execute_agent_creates_output_logger(self, temp_workspace: Workspace): + """Runtime execute_agent should create an output logger for the run.""" + from codeframe.core import runtime, tasks as tasks_module + from codeframe.core.streaming import run_output_exists + + # Create task and run + task = tasks_module.create(temp_workspace, title="Test task") + run = runtime.start_task_run(temp_workspace, task.id) + + # Mock the Agent class at its definition location + with patch("codeframe.core.agent.Agent") as MockAgent, \ + patch("codeframe.adapters.llm.get_provider"): + + mock_agent = MagicMock() + mock_agent.run.return_value = MagicMock( + status=MagicMock(value="completed"), + blocker=None, + ) + MockAgent.return_value = mock_agent + + # Patch os.getenv to provide API key + with patch.dict("os.environ", {"ANTHROPIC_API_KEY": "test-key"}): + try: + runtime.execute_agent(temp_workspace, run) + except Exception: + pass # May fail on other things, but logger should be created + + # Output log should exist + assert run_output_exists(temp_workspace, run.id) + + def test_output_logger_passed_to_agent(self, temp_workspace: Workspace): + """Runtime should pass the output logger to the Agent.""" + from codeframe.core import runtime, tasks as tasks_module + from codeframe.core.streaming import RunOutputLogger + + task = tasks_module.create(temp_workspace, title="Test task") + run = runtime.start_task_run(temp_workspace, task.id) + + captured_logger = None + + def capture_agent(*args, **kwargs): + nonlocal captured_logger + captured_logger = kwargs.get("output_logger") + mock = MagicMock() + mock.run.return_value = MagicMock( + status=MagicMock(value="completed"), + blocker=None, + ) + return mock + + with patch("codeframe.core.agent.Agent", side_effect=capture_agent), \ + patch("codeframe.adapters.llm.get_provider"), \ + patch.dict("os.environ", {"ANTHROPIC_API_KEY": "test-key"}): + try: + runtime.execute_agent(temp_workspace, run) + except Exception: + pass + + # Agent should have received a logger + assert captured_logger is not None + assert isinstance(captured_logger, RunOutputLogger) + + +class TestAgentOutputContent: + """Tests for the content written to the output log.""" + + def test_agent_logs_planning_start( + self, temp_workspace: Workspace, mock_llm_provider + ): + """Agent should log when planning starts.""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=True, + output_logger=logger, + ) + + # Simulate planning status change + agent._verbose_print("[PLAN] Creating implementation plan...") + + logger.close() + + content = logger.log_path.read_text() + assert "PLAN" in content or "plan" in content.lower() + + def test_agent_logs_step_execution( + self, temp_workspace: Workspace, mock_llm_provider + ): + """Agent should log step execution.""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=True, + output_logger=logger, + ) + + agent._verbose_print("[STEP 1] Creating file: test.py") + + logger.close() + + content = logger.log_path.read_text() + assert "STEP" in content + + def test_agent_logs_verification_attempts( + self, temp_workspace: Workspace, mock_llm_provider + ): + """Agent should log verification attempts.""" + from codeframe.core.agent import Agent + + logger = RunOutputLogger(temp_workspace, "test-run-id") + + agent = Agent( + workspace=temp_workspace, + llm_provider=mock_llm_provider, + verbose=True, + output_logger=logger, + ) + + agent._verbose_print("[VERIFY] Attempt 1/3") + + logger.close() + + content = logger.log_path.read_text() + assert "VERIFY" in content diff --git a/tests/core/test_streaming.py b/tests/core/test_streaming.py new file mode 100644 index 00000000..4bd3ea94 --- /dev/null +++ b/tests/core/test_streaming.py @@ -0,0 +1,311 @@ +"""Tests for the streaming module (work follow functionality). + +This module tests the file-based streaming infrastructure used by +`cf work follow` to stream real-time execution output. +""" + +import pytest +import time +import threading +from pathlib import Path + +from codeframe.core.workspace import Workspace + + +@pytest.fixture +def temp_workspace(tmp_path: Path) -> Workspace: + """Create a temporary workspace for testing.""" + from codeframe.core.workspace import create_or_load_workspace + + workspace = create_or_load_workspace(tmp_path) + return workspace + + +@pytest.fixture +def run_id() -> str: + """Generate a test run ID.""" + return "test-run-12345678" + + +class TestRunOutputPath: + """Tests for run output path generation.""" + + def test_get_run_output_path_returns_expected_structure( + self, temp_workspace: Workspace, run_id: str + ): + """Output path should follow .codeframe/runs//output.log pattern.""" + from codeframe.core.streaming import get_run_output_path + + path = get_run_output_path(temp_workspace, run_id) + + assert path.name == "output.log" + assert path.parent.name == run_id + assert path.parent.parent.name == "runs" + assert ".codeframe" in str(path) + + def test_get_run_output_path_is_within_workspace( + self, temp_workspace: Workspace, run_id: str + ): + """Output path should be within the workspace directory.""" + from codeframe.core.streaming import get_run_output_path + + path = get_run_output_path(temp_workspace, run_id) + + assert str(path).startswith(str(temp_workspace.repo_path)) + + +class TestRunOutputLogger: + """Tests for the RunOutputLogger class.""" + + def test_create_logger_creates_directory( + self, temp_workspace: Workspace, run_id: str + ): + """Creating a logger should create the output directory.""" + from codeframe.core.streaming import RunOutputLogger + + logger = RunOutputLogger(temp_workspace, run_id) + + assert logger.log_path.parent.exists() + logger.close() + + def test_logger_writes_to_file( + self, temp_workspace: Workspace, run_id: str + ): + """Logger should write messages to the log file.""" + from codeframe.core.streaming import RunOutputLogger + + logger = RunOutputLogger(temp_workspace, run_id) + logger.write("Test message\n") + logger.close() + + content = logger.log_path.read_text() + assert "Test message" in content + + def test_logger_flushes_immediately( + self, temp_workspace: Workspace, run_id: str + ): + """Logger should flush after each write for real-time streaming.""" + from codeframe.core.streaming import RunOutputLogger + + logger = RunOutputLogger(temp_workspace, run_id) + logger.write("First message\n") + + # Read before close - content should be available + content = logger.log_path.read_text() + assert "First message" in content + logger.close() + + def test_logger_handles_context_manager( + self, temp_workspace: Workspace, run_id: str + ): + """Logger should work as context manager.""" + from codeframe.core.streaming import RunOutputLogger, get_run_output_path + + with RunOutputLogger(temp_workspace, run_id) as logger: + logger.write("Context message\n") + + path = get_run_output_path(temp_workspace, run_id) + content = path.read_text() + assert "Context message" in content + + def test_logger_writes_with_timestamp_option( + self, temp_workspace: Workspace, run_id: str + ): + """Logger should optionally include timestamps.""" + from codeframe.core.streaming import RunOutputLogger + + with RunOutputLogger(temp_workspace, run_id) as logger: + logger.write_timestamped("Timestamped message") + + content = logger.log_path.read_text() + # Should contain timestamp pattern [HH:MM:SS] + assert "[" in content and "]" in content + assert "Timestamped message" in content + + +class TestTailRunOutput: + """Tests for tailing run output files.""" + + def test_tail_yields_existing_lines( + self, temp_workspace: Workspace, run_id: str + ): + """Tail should yield lines that exist in the file.""" + from codeframe.core.streaming import RunOutputLogger, tail_run_output + + # Create log with content + with RunOutputLogger(temp_workspace, run_id) as logger: + logger.write("Line 1\n") + logger.write("Line 2\n") + + # Tail should yield these lines + lines = list(tail_run_output(temp_workspace, run_id, max_iterations=1)) + + assert len(lines) == 2 + assert "Line 1" in lines[0] + assert "Line 2" in lines[1] + + def test_tail_with_since_line_skips_old_content( + self, temp_workspace: Workspace, run_id: str + ): + """Tail with since_line should skip already-seen lines.""" + from codeframe.core.streaming import RunOutputLogger, tail_run_output + + # Create log with content + with RunOutputLogger(temp_workspace, run_id) as logger: + logger.write("Line 1\n") + logger.write("Line 2\n") + logger.write("Line 3\n") + + # Tail from line 2 (0-indexed, so skip first 2) + lines = list(tail_run_output(temp_workspace, run_id, since_line=2, max_iterations=1)) + + assert len(lines) == 1 + assert "Line 3" in lines[0] + + def test_tail_yields_new_lines_from_concurrent_writer( + self, temp_workspace: Workspace, run_id: str + ): + """Tail should yield new lines as they're written by another process.""" + from codeframe.core.streaming import RunOutputLogger, tail_run_output + + # Start with empty log + logger = RunOutputLogger(temp_workspace, run_id) + + collected_lines = [] + stop_event = threading.Event() + + def tail_collector(): + for line in tail_run_output( + temp_workspace, run_id, poll_interval=0.1, max_wait=2.0 + ): + collected_lines.append(line) + if "Final" in line: + stop_event.set() + break + + # Start tailing in background + tail_thread = threading.Thread(target=tail_collector) + tail_thread.start() + + # Write lines with small delays + time.sleep(0.2) + logger.write("First line\n") + time.sleep(0.2) + logger.write("Second line\n") + time.sleep(0.2) + logger.write("Final line\n") + logger.close() + + # Wait for tail to finish + stop_event.wait(timeout=3.0) + tail_thread.join(timeout=1.0) + + assert len(collected_lines) >= 3 + assert any("First" in line for line in collected_lines) + assert any("Final" in line for line in collected_lines) + + def test_tail_handles_missing_file_gracefully( + self, temp_workspace: Workspace + ): + """Tail should handle missing file by waiting for it to appear.""" + from codeframe.core.streaming import tail_run_output + + # File doesn't exist yet + lines = list(tail_run_output( + temp_workspace, + "nonexistent-run", + max_iterations=1, + max_wait=0.5 + )) + + assert len(lines) == 0 # No lines, but no exception + + +class TestGetLatestRunLines: + """Tests for reading buffered output (--tail N functionality).""" + + def test_get_latest_lines_returns_last_n_lines( + self, temp_workspace: Workspace, run_id: str + ): + """Should return the last N lines from the log file.""" + from codeframe.core.streaming import RunOutputLogger, get_latest_lines + + # Create log with 10 lines + with RunOutputLogger(temp_workspace, run_id) as logger: + for i in range(10): + logger.write(f"Line {i}\n") + + # Get last 3 lines + lines = get_latest_lines(temp_workspace, run_id, count=3) + + assert len(lines) == 3 + assert "Line 7" in lines[0] + assert "Line 8" in lines[1] + assert "Line 9" in lines[2] + + def test_get_latest_lines_returns_all_if_fewer_than_n( + self, temp_workspace: Workspace, run_id: str + ): + """Should return all lines if fewer than N exist.""" + from codeframe.core.streaming import RunOutputLogger, get_latest_lines + + # Create log with 2 lines + with RunOutputLogger(temp_workspace, run_id) as logger: + logger.write("Line 1\n") + logger.write("Line 2\n") + + # Ask for 10 lines + lines = get_latest_lines(temp_workspace, run_id, count=10) + + assert len(lines) == 2 + assert "Line 1" in lines[0] + assert "Line 2" in lines[1] + + def test_get_latest_lines_returns_empty_for_missing_file( + self, temp_workspace: Workspace + ): + """Should return empty list if file doesn't exist.""" + from codeframe.core.streaming import get_latest_lines + + lines = get_latest_lines(temp_workspace, "nonexistent", count=5) + + assert lines == [] + + def test_get_latest_lines_returns_line_count( + self, temp_workspace: Workspace, run_id: str + ): + """Should return the total line count along with lines.""" + from codeframe.core.streaming import RunOutputLogger, get_latest_lines_with_count + + # Create log with 10 lines + with RunOutputLogger(temp_workspace, run_id) as logger: + for i in range(10): + logger.write(f"Line {i}\n") + + lines, total = get_latest_lines_with_count(temp_workspace, run_id, count=3) + + assert len(lines) == 3 + assert total == 10 + + +class TestRunOutputExists: + """Tests for checking if run output exists.""" + + def test_output_exists_returns_false_for_missing_file( + self, temp_workspace: Workspace + ): + """Should return False if output file doesn't exist.""" + from codeframe.core.streaming import run_output_exists + + assert run_output_exists(temp_workspace, "nonexistent") is False + + def test_output_exists_returns_true_for_existing_file( + self, temp_workspace: Workspace, run_id: str + ): + """Should return True if output file exists.""" + from codeframe.core.streaming import RunOutputLogger, run_output_exists + + with RunOutputLogger(temp_workspace, run_id) as logger: + logger.write("Test\n") + + assert run_output_exists(temp_workspace, run_id) is True From 8106b6575c7a9a3e4578d0a8a4a013144c4cd2fb Mon Sep 17 00:00:00 2001 From: Test User Date: Thu, 29 Jan 2026 23:05:40 -0700 Subject: [PATCH 2/3] fix: address CodeRabbit review feedback for work follow feature - Wrap agent execution in try/finally to ensure output logger cleanup - Add time-based throttling for status checks (every 1s instead of per-line) - Add pytest.mark.v2 marker to test_work_follow.py for v2 test filtering --- codeframe/cli/app.py | 36 ++-- codeframe/core/runtime.py | 352 +++++++++++++++++----------------- tests/cli/test_work_follow.py | 4 + 3 files changed, 203 insertions(+), 189 deletions(-) diff --git a/codeframe/cli/app.py b/codeframe/cli/app.py index bc2347d5..40e1b611 100644 --- a/codeframe/cli/app.py +++ b/codeframe/cli/app.py @@ -2541,6 +2541,11 @@ def work_follow( } try: + import time + + last_status_check = time.time() + STATUS_CHECK_INTERVAL = 1.0 # Check status every 1 second + # Stream output for line in tail_run_output( workspace, @@ -2551,20 +2556,23 @@ def work_follow( ): console.print(line.rstrip()) - # Check if run is still active (periodically) - current_run = runtime.get_run(workspace, active_run.id) - if current_run and current_run.status in TERMINAL_STATUSES: - # Show completion message - status_color = { - runtime.RunStatus.COMPLETED: "green", - runtime.RunStatus.FAILED: "red", - runtime.RunStatus.BLOCKED: "yellow", - }.get(current_run.status, "white") - - console.print( - f"\n[{status_color}]Run {current_run.status.value}[/{status_color}]" - ) - break + # Check run status periodically (not on every line) + current_time = time.time() + if current_time - last_status_check >= STATUS_CHECK_INTERVAL: + last_status_check = current_time + current_run = runtime.get_run(workspace, active_run.id) + if current_run and current_run.status in TERMINAL_STATUSES: + # Show completion message + status_color = { + runtime.RunStatus.COMPLETED: "green", + runtime.RunStatus.FAILED: "red", + runtime.RunStatus.BLOCKED: "yellow", + }.get(current_run.status, "white") + + console.print( + f"\n[{status_color}]Run {current_run.status.value}[/{status_color}]" + ) + break except KeyboardInterrupt: console.print("\n[yellow]Streaming interrupted[/yellow]") diff --git a/codeframe/core/runtime.py b/codeframe/core/runtime.py index b0aef690..55d660e1 100644 --- a/codeframe/core/runtime.py +++ b/codeframe/core/runtime.py @@ -608,193 +608,195 @@ def execute_agent( from codeframe.core.streaming import RunOutputLogger output_logger = RunOutputLogger(workspace, run.id) - # Create event callback to emit workspace events and log - def on_agent_event(event_type: str, data: dict) -> None: - events.emit_for_workspace( - workspace, - events.EventType.AGENT_STEP_STARTED if "started" in event_type else events.EventType.AGENT_STEP_COMPLETED, - {"run_id": run.id, "agent_event": event_type, **data}, - print_event=True, + try: + # Create event callback to emit workspace events and log + def on_agent_event(event_type: str, data: dict) -> None: + events.emit_for_workspace( + workspace, + events.EventType.AGENT_STEP_STARTED if "started" in event_type else events.EventType.AGENT_STEP_COMPLETED, + {"run_id": run.id, "agent_event": event_type, **data}, + print_event=True, + ) + + # Also log to run logger for diagnosis + category = _event_type_to_category(event_type) + run_logger.info(category, f"Agent event: {event_type}", data) + + # Create and run agent + agent = Agent( + workspace=workspace, + llm_provider=provider, + dry_run=dry_run, + on_event=on_agent_event, + debug=debug, + verbose=verbose, + fix_coordinator=fix_coordinator, + output_logger=output_logger, ) - # Also log to run logger for diagnosis - category = _event_type_to_category(event_type) - run_logger.info(category, f"Agent event: {event_type}", data) - - # Create and run agent - agent = Agent( - workspace=workspace, - llm_provider=provider, - dry_run=dry_run, - on_event=on_agent_event, - debug=debug, - verbose=verbose, - fix_coordinator=fix_coordinator, - output_logger=output_logger, - ) + state = agent.run(run.task_id) - state = agent.run(run.task_id) - - # If agent is BLOCKED, try supervisor resolution - if state.status == AgentStatus.BLOCKED: - from codeframe.core.conductor import get_supervisor - - supervisor = get_supervisor(workspace) - if supervisor.try_resolve_blocked_task(run.task_id): - # Supervisor resolved the blocker - retry the agent - print("[Supervisor] Retrying task after auto-resolution...") - - # Create a new agent instance and retry - agent = Agent( - workspace=workspace, - llm_provider=provider, - dry_run=dry_run, - on_event=on_agent_event, - debug=debug, - fix_coordinator=fix_coordinator, - output_logger=output_logger, - ) - state = agent.run(run.task_id) - - # If agent FAILED, check if supervisor can help with common technical issues - if state.status == AgentStatus.FAILED: - from codeframe.core.conductor import get_supervisor, SUPERVISOR_TACTICAL_PATTERNS - - if debug: - logger.debug("Agent FAILED - analyzing for supervisor intervention") - logger.debug("state.blocker: %s", state.blocker) - logger.debug( - "state.step_results count: %d", - len(state.step_results) if state.step_results else 0 - ) - logger.debug( - "state.gate_results count: %d", - len(state.gate_results) if state.gate_results else 0 - ) + # If agent is BLOCKED, try supervisor resolution + if state.status == AgentStatus.BLOCKED: + from codeframe.core.conductor import get_supervisor + + supervisor = get_supervisor(workspace) + if supervisor.try_resolve_blocked_task(run.task_id): + # Supervisor resolved the blocker - retry the agent + print("[Supervisor] Retrying task after auto-resolution...") + + # Create a new agent instance and retry + agent = Agent( + workspace=workspace, + llm_provider=provider, + dry_run=dry_run, + on_event=on_agent_event, + debug=debug, + fix_coordinator=fix_coordinator, + output_logger=output_logger, + ) + state = agent.run(run.task_id) + + # If agent FAILED, check if supervisor can help with common technical issues + if state.status == AgentStatus.FAILED: + from codeframe.core.conductor import get_supervisor, SUPERVISOR_TACTICAL_PATTERNS - # Extract error message from available sources - error_msg = "" - error_source = "none" - if state.blocker: - error_msg = state.blocker.reason or state.blocker.question or "" - error_source = "blocker" - elif state.step_results: - # Check last step result for error info - last_result = state.step_results[-1] if debug: - error_preview = last_result.error[:200] if last_result.error else "None" + logger.debug("Agent FAILED - analyzing for supervisor intervention") + logger.debug("state.blocker: %s", state.blocker) logger.debug( - "Last step result: status=%s, error=%s", - last_result.status, error_preview + "state.step_results count: %d", + len(state.step_results) if state.step_results else 0 ) - if hasattr(last_result, 'error') and last_result.error: - error_msg = last_result.error - error_source = "step_result.error" - elif hasattr(last_result, 'output') and last_result.output: - error_msg = last_result.output - error_source = "step_result.output" - elif state.gate_results: - # Check gate results for failure info - for gate in state.gate_results: + logger.debug( + "state.gate_results count: %d", + len(state.gate_results) if state.gate_results else 0 + ) + + # Extract error message from available sources + error_msg = "" + error_source = "none" + if state.blocker: + error_msg = state.blocker.reason or state.blocker.question or "" + error_source = "blocker" + elif state.step_results: + # Check last step result for error info + last_result = state.step_results[-1] if debug: - logger.debug("Gate result: passed=%s", gate.passed) - if not gate.passed: - for check in gate.checks: - if debug: - output_preview = check.output[:100] if check.output else "None" - logger.debug( - " Check: %s status=%s output=%s", - check.name, check.status, output_preview - ) - if check.output: - error_msg = check.output - error_source = f"gate.{check.name}" - break - - if debug: - logger.debug("Extracted error from: %s", error_source) - error_preview = error_msg[:300] if error_msg else "EMPTY" - logger.debug("Error message (first 300 chars): %s", error_preview) - - error_msg_lower = error_msg.lower() - matched_patterns = [p for p in SUPERVISOR_TACTICAL_PATTERNS if p in error_msg_lower] - if debug: - logger.debug("Matched tactical patterns: %s", matched_patterns) - - if error_msg and matched_patterns: - supervisor = get_supervisor(workspace) - resolution = supervisor._generate_tactical_resolution(error_msg) - logger.info( - "Supervisor detected recoverable error, providing guidance: %s...", - resolution[:100] - ) + error_preview = last_result.error[:200] if last_result.error else "None" + logger.debug( + "Last step result: status=%s, error=%s", + last_result.status, error_preview + ) + if hasattr(last_result, 'error') and last_result.error: + error_msg = last_result.error + error_source = "step_result.error" + elif hasattr(last_result, 'output') and last_result.output: + error_msg = last_result.output + error_source = "step_result.output" + elif state.gate_results: + # Check gate results for failure info + for gate in state.gate_results: + if debug: + logger.debug("Gate result: passed=%s", gate.passed) + if not gate.passed: + for check in gate.checks: + if debug: + output_preview = check.output[:100] if check.output else "None" + logger.debug( + " Check: %s status=%s output=%s", + check.name, check.status, output_preview + ) + if check.output: + error_msg = check.output + error_source = f"gate.{check.name}" + break - # Create a blocker with the resolution for the agent's next run - from codeframe.core import blockers - blocker = blockers.create( - workspace, - task_id=run.task_id, - question=f"Technical error: {error_msg[:500]}", - ) - blockers.answer(workspace, blocker.id, resolution) if debug: - logger.debug("Created blocker %s and answered with resolution", blocker.id[:8]) - - # Retry the agent with the new context - logger.info("Supervisor retrying task with guidance...") - agent = Agent( - workspace=workspace, - llm_provider=provider, - dry_run=dry_run, - on_event=on_agent_event, - debug=debug, - fix_coordinator=fix_coordinator, - output_logger=output_logger, - ) - state = agent.run(run.task_id) + logger.debug("Extracted error from: %s", error_source) + error_preview = error_msg[:300] if error_msg else "EMPTY" + logger.debug("Error message (first 300 chars): %s", error_preview) + + error_msg_lower = error_msg.lower() + matched_patterns = [p for p in SUPERVISOR_TACTICAL_PATTERNS if p in error_msg_lower] if debug: - logger.debug("Retry completed with status: %s", state.status) - elif debug: - logger.debug( - "No supervisor intervention - error_msg empty=%s, no pattern match=%s", - not error_msg, not matched_patterns - ) + logger.debug("Matched tactical patterns: %s", matched_patterns) + + if error_msg and matched_patterns: + supervisor = get_supervisor(workspace) + resolution = supervisor._generate_tactical_resolution(error_msg) + logger.info( + "Supervisor detected recoverable error, providing guidance: %s...", + resolution[:100] + ) + + # Create a blocker with the resolution for the agent's next run + from codeframe.core import blockers + blocker = blockers.create( + workspace, + task_id=run.task_id, + question=f"Technical error: {error_msg[:500]}", + ) + blockers.answer(workspace, blocker.id, resolution) + if debug: + logger.debug("Created blocker %s and answered with resolution", blocker.id[:8]) + + # Retry the agent with the new context + logger.info("Supervisor retrying task with guidance...") + agent = Agent( + workspace=workspace, + llm_provider=provider, + dry_run=dry_run, + on_event=on_agent_event, + debug=debug, + fix_coordinator=fix_coordinator, + output_logger=output_logger, + ) + state = agent.run(run.task_id) + if debug: + logger.debug("Retry completed with status: %s", state.status) + elif debug: + logger.debug( + "No supervisor intervention - error_msg empty=%s, no pattern match=%s", + not error_msg, not matched_patterns + ) - # Log final status - if state.status == AgentStatus.COMPLETED: - run_logger.info(LogCategory.STATE_CHANGE, "Agent completed successfully") - elif state.status == AgentStatus.BLOCKED: - blocker_reason = state.blocker.question if state.blocker else "Unknown" - run_logger.warning(LogCategory.BLOCKER, f"Agent blocked: {blocker_reason[:200]}", { - "blocker_question": blocker_reason, - }) - elif state.status == AgentStatus.FAILED: - # Log detailed error information for diagnosis - error_info = {} - if state.step_results: - last_step = state.step_results[-1] - error_info["last_step_status"] = last_step.status.value if hasattr(last_step.status, 'value') else str(last_step.status) - error_info["last_step_error"] = last_step.error[:500] if last_step.error else None - if state.gate_results: - error_info["gate_failures"] = sum(1 for g in state.gate_results if not g.passed) - run_logger.error(LogCategory.ERROR, "Agent execution failed", error_info) - - # Update run status based on agent result - if state.status == AgentStatus.COMPLETED: - complete_run(workspace, run.id) - elif state.status == AgentStatus.BLOCKED: - # Get blocker ID from state if available - blocker_id = "" - if state.blocker and hasattr(state, "_blocker_id"): - blocker_id = state._blocker_id - block_run(workspace, run.id, blocker_id) - elif state.status == AgentStatus.FAILED: - fail_run(workspace, run.id) - - # Close output logger - output_logger.close() - - return state + # Log final status + if state.status == AgentStatus.COMPLETED: + run_logger.info(LogCategory.STATE_CHANGE, "Agent completed successfully") + elif state.status == AgentStatus.BLOCKED: + blocker_reason = state.blocker.question if state.blocker else "Unknown" + run_logger.warning(LogCategory.BLOCKER, f"Agent blocked: {blocker_reason[:200]}", { + "blocker_question": blocker_reason, + }) + elif state.status == AgentStatus.FAILED: + # Log detailed error information for diagnosis + error_info = {} + if state.step_results: + last_step = state.step_results[-1] + error_info["last_step_status"] = last_step.status.value if hasattr(last_step.status, 'value') else str(last_step.status) + error_info["last_step_error"] = last_step.error[:500] if last_step.error else None + if state.gate_results: + error_info["gate_failures"] = sum(1 for g in state.gate_results if not g.passed) + run_logger.error(LogCategory.ERROR, "Agent execution failed", error_info) + + # Update run status based on agent result + if state.status == AgentStatus.COMPLETED: + complete_run(workspace, run.id) + elif state.status == AgentStatus.BLOCKED: + # Get blocker ID from state if available + blocker_id = "" + if state.blocker and hasattr(state, "_blocker_id"): + blocker_id = state._blocker_id + block_run(workspace, run.id, blocker_id) + elif state.status == AgentStatus.FAILED: + fail_run(workspace, run.id) + + return state + + finally: + # Always close the output logger to ensure file is properly flushed + output_logger.close() def _event_type_to_category(event_type: str): diff --git a/tests/cli/test_work_follow.py b/tests/cli/test_work_follow.py index 5b12cf60..63a3a150 100644 --- a/tests/cli/test_work_follow.py +++ b/tests/cli/test_work_follow.py @@ -15,6 +15,10 @@ from codeframe.core import tasks +# Mark all tests in this module as v2 tests (CLI-first, headless functionality) +pytestmark = pytest.mark.v2 + + runner = CliRunner() From 269a78eb86f19b974afcf06cf0746e0b9972ad54 Mon Sep 17 00:00:00 2001 From: Test User Date: Thu, 29 Jan 2026 23:06:55 -0700 Subject: [PATCH 3/3] fix: address additional CodeRabbit feedback - Add verbose=verbose to supervisor retry Agent constructions - Add defensive initialization in RunOutputLogger (__init__ and close) --- codeframe/core/runtime.py | 2 ++ codeframe/core/streaming.py | 3 ++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/codeframe/core/runtime.py b/codeframe/core/runtime.py index 55d660e1..d029ab9d 100644 --- a/codeframe/core/runtime.py +++ b/codeframe/core/runtime.py @@ -652,6 +652,7 @@ def on_agent_event(event_type: str, data: dict) -> None: dry_run=dry_run, on_event=on_agent_event, debug=debug, + verbose=verbose, fix_coordinator=fix_coordinator, output_logger=output_logger, ) @@ -749,6 +750,7 @@ def on_agent_event(event_type: str, data: dict) -> None: dry_run=dry_run, on_event=on_agent_event, debug=debug, + verbose=verbose, fix_coordinator=fix_coordinator, output_logger=output_logger, ) diff --git a/codeframe/core/streaming.py b/codeframe/core/streaming.py index 87002f24..a9d5613f 100644 --- a/codeframe/core/streaming.py +++ b/codeframe/core/streaming.py @@ -68,6 +68,7 @@ def __init__(self, workspace: Workspace, run_id: str): self.workspace = workspace self.run_id = run_id self.log_path = get_run_output_path(workspace, run_id) + self._file = None # Initialize before potential mkdir/open failure # Ensure directory exists self.log_path.parent.mkdir(parents=True, exist_ok=True) @@ -99,7 +100,7 @@ def write_timestamped(self, message: str) -> None: def close(self) -> None: """Close the log file.""" - if self._file and not self._file.closed: + if hasattr(self, "_file") and self._file and not self._file.closed: self._file.close() def __enter__(self) -> "RunOutputLogger":