feat(cli): add cf work follow for live execution streaming
#319
Conversation
…ming 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/<run_id>/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
WalkthroughAdds file-based real-time streaming for work runs: new streaming module and RunOutputLogger, Agent/runtime wiring to write run output, and a CLI subcommand Changes
Sequence DiagramsequenceDiagram
participant User as User
participant CLI as "cf work follow"
participant WS as Workspace
participant Task as Task/Run
participant Stream as Stream Module
participant Logger as RunOutputLogger
User->>CLI: cf work follow <task-id> [--tail N]
CLI->>WS: resolve workspace
CLI->>Task: find task by partial id
alt task found
CLI->>Task: get active run
alt active run exists
opt tail requested
CLI->>Stream: get_latest_lines_with_count(N)
Stream-->>CLI: return buffered lines
CLI->>User: print buffered lines
end
CLI->>Stream: tail_run_output(run_id)
loop poll for new lines
Stream->>Logger: read/flush output.log
Logger-->>Stream: yield new lines
Stream-->>CLI: deliver lines
CLI->>User: print live output
end
Task->>CLI: run completes
CLI->>User: print completion message
else no active run
CLI->>Stream: get_latest_lines(run_id) (if recent)
CLI->>User: print final output or suggest start
end
else task not found
CLI->>User: error "no task found"
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
Add
|
Code Review:
|
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
|
|
||
| try: | ||
| # Stream output | ||
| for line in tail_run_output( |
There was a problem hiding this comment.
🟡 Medium
The terminal status check at line 2555 only runs after tail_run_output yields a new line. If the run completes without producing new output, the loop hangs indefinitely. Consider checking run status inside the tail_run_output generator or implementing a separate status-polling mechanism that runs independently of line yields.
🚀 Want me to fix this? Reply ex: "fix it for me".
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
codeframe/core/runtime.py (1)
648-656:⚠️ Potential issue | 🟡 MinorPreserve verbose output on supervisor retries
Line 648 and Line 745 recreate the Agent without
verbose=verbose, so stdout streaming stops after the first retry even when the user requested verbose. Pass the flag through both retry constructions.🛠️ Proposed fix
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, ) @@ 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 applies to: 745-753
🤖 Fix all issues with AI agents
In `@codeframe/core/runtime.py`:
- Around line 607-610: The RunOutputLogger instance (created as output_logger =
RunOutputLogger(workspace, run.id)) must be closed on all exit paths; wrap the
remaining execute_agent body that calls agent.run() and any supervisor retry
logic in a try/finally (or use a context manager) and call output_logger.close()
in the finally block so the file handle is always released and streaming flushed
even if agent.run() or supervisor retry raises.
In `@tests/cli/test_work_follow.py`:
- Around line 6-20: The module lacks the required pytest v2 marker; add a
module-level variable pytestmark = pytest.mark.v2 near the top of the test file
(e.g., alongside the existing imports where `from codeframe.cli.app import app`,
`runner = CliRunner()`, and other globals are defined) so the tests using the
`app` CLI are included when running `pytest -m v2`; ensure `pytest` is imported
if not already.
🧹 Nitpick comments (2)
codeframe/core/streaming.py (2)
61-76: Consider defensive initialization for robustness.If
mkdir()succeeds butopen()fails (e.g., permission issues), the object is partially initialized without_file, which could cause anAttributeErrorinclose()if called on an incompletely constructed instance.🛡️ Proposed defensive initialization
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) + self._file = None # Initialize before potential failure # 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")
100-103: Guard against uninitialized_fileattribute.If
__init__fails after the class is instantiated but before_fileis assigned, callingclose()directly or via__exit__will raiseAttributeError.🛡️ Proposed safer close implementation
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()
- 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
- Add verbose=verbose to supervisor retry Agent constructions - Add defensive initialization in RunOutputLogger (__init__ and close)
Follow-up Review: Latest Commit (8106b65)Excellent work addressing the review feedback! The latest commit successfully resolves the issues identified by CodeRabbit: ✅ Issues Resolved
This addresses both CodeRabbit's actionable comments and implements one of my performance suggestions from the initial review.
|
| print("[Supervisor] Retrying task after auto-resolution...") | ||
|
|
||
| # Create a new agent instance and retry | ||
| agent = Agent( |
There was a problem hiding this comment.
🟢 Low
The verbose parameter is missing when creating retry Agent instances (lines 649 and 746), though it's included in the initial agent. Consider adding verbose=verbose to preserve user-requested verbose mode on retries.
🚀 Want me to fix this? Reply ex: "fix it for me".
Code Review:
|
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@codeframe/cli/app.py`:
- Around line 2421-2426: The tail Typer option currently allows non-positive
values which cause counterintuitive slicing; ensure tail is constrained to >= 1
by either adding a min bound on the Option or validating after parsing.
Specifically, update the tail declaration (symbol: tail) to include a minimum
(e.g., typer.Option(..., min=1)) if supported, or add a simple check where tail
is used (e.g., if tail is not None and tail < 1: raise
typer.BadParameter("--tail must be >= 1")) to reject zero/negative inputs.
- Around line 2549-2575: The loop currently only polls run status when
tail_run_output yields a line, which can hang if the run finishes without
emitting new output; fix by decoupling output consumption from status polling:
get an iterator from tail_run_output(...) (use the same args), then run a while
loop that attempts to read the next line non-blockingly (or with a short
timeout) and prints it when available, but independently every
STATUS_CHECK_INTERVAL call runtime.get_run(workspace, active_run.id) and compare
against TERMINAL_STATUSES; update last_status_check when you poll, break and
print the colored completion message when the run is terminal, and ensure
StopIteration or generator exhaustion also triggers a final status check to
avoid hanging.
| tail: Optional[int] = typer.Option( | ||
| None, | ||
| "--tail", | ||
| "-n", | ||
| help="Show last N lines of buffered output before streaming", | ||
| ), |
There was a problem hiding this comment.
Validate --tail to avoid negative/zero surprises.
tail <= 0 leads to counterintuitive slicing (e.g., -1 returns nearly all lines). Consider constraining it to >= 1.
🛠️ Proposed validation
tail: Optional[int] = typer.Option(
None,
"--tail",
"-n",
+ min=1,
help="Show last N lines of buffered output before streaming",
),🤖 Prompt for AI Agents
In `@codeframe/cli/app.py` around lines 2421 - 2426, The tail Typer option
currently allows non-positive values which cause counterintuitive slicing;
ensure tail is constrained to >= 1 by either adding a min bound on the Option or
validating after parsing. Specifically, update the tail declaration (symbol:
tail) to include a minimum (e.g., typer.Option(..., min=1)) if supported, or add
a simple check where tail is used (e.g., if tail is not None and tail < 1: raise
typer.BadParameter("--tail must be >= 1")) to reject zero/negative inputs.
| # 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 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 |
There was a problem hiding this comment.
Follow can hang when the run finishes without emitting new output.
Status polling only happens when a line is yielded. If the run completes after the last line (or never logs), the loop never checks terminal status and can block indefinitely.
🛠️ Suggested refactor to decouple status checks from output
- last_status_check = time.time()
- STATUS_CHECK_INTERVAL = 1.0 # Check status every 1 second
-
- # 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 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
+ STATUS_CHECK_INTERVAL = 1.0 # Check status every 1 second
+ current_line = start_line
+ overall_start = time.time()
+
+ while True:
+ for line in tail_run_output(
+ workspace,
+ active_run.id,
+ since_line=current_line,
+ poll_interval=0.3,
+ max_wait=STATUS_CHECK_INTERVAL,
+ ):
+ console.print(line.rstrip())
+ current_line += 1
+
+ # Check run status even if no new output arrived
+ current_run = runtime.get_run(workspace, active_run.id)
+ if current_run and current_run.status in TERMINAL_STATUSES:
+ 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
+
+ if max_wait is not None and (time.time() - overall_start) >= max_wait:
+ break🤖 Prompt for AI Agents
In `@codeframe/cli/app.py` around lines 2549 - 2575, The loop currently only polls
run status when tail_run_output yields a line, which can hang if the run
finishes without emitting new output; fix by decoupling output consumption from
status polling: get an iterator from tail_run_output(...) (use the same args),
then run a while loop that attempts to read the next line non-blockingly (or
with a short timeout) and prints it when available, but independently every
STATUS_CHECK_INTERVAL call runtime.get_run(workspace, active_run.id) and compare
against TERMINAL_STATUSES; update last_status_check when you poll, break and
print the colored completion message when the run is terminal, and ensure
StopIteration or generator exhaustion also triggers a final status check to
avoid hanging.
Summary
Implements
cf work followcommand for real-time streaming of task execution output (#308).cf work follow <task-id>streams agent output as it happens--tail Nshows last N lines before streaming live.codeframe/runs/<run_id>/output.logKey Components
core/streaming.pycli/app.pywork_followcommand implementationcore/agent.py_verbose_print()for dual-writecore/runtime.pyUsage
Test plan
tests/core/test_streaming.py)tests/core/test_agent_streaming.py)tests/cli/test_work_follow.py)tests/cli/test_v2_cli_integration.pyCloses #308
Summary by CodeRabbit
cf work follow <task-id>to stream live task run output in the CLI.--tail <n>to preload the last n lines before streaming.✏️ Tip: You can customize this high-level summary in your review settings.