Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/stepfork/cli/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ def export_command(
str | None,
typer.Option(
"--entrypoint",
"-e",
help="Trusted local MODULE:FUNCTION the generated test executes.",
),
] = None,
Expand Down
22 changes: 13 additions & 9 deletions src/stepfork/cli/minimize.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
from rich.panel import Panel
from rich.table import Table

from stepfork.minimize.delta_debug import create_minimized_trace
from stepfork.minimize.engine import MinimizationEngine, generate_report
from stepfork.minimize.models import MinimizationReport
from stepfork.minimize.signature import ReplayNeverReachedRecordedCall
Expand Down Expand Up @@ -175,15 +174,20 @@ def minimize_command(

# Save the minimized trace
try:
from stepfork.minimize.delta_debug import EventSubset

minimal_subset = EventSubset(
events=tuple(
trace.events[i] for i in range(result.minimized_event_count or 0)
),
includes_dependency_calls=True,
if result.minimized_events is None:
console.print("\n[red]Error: no minimized events in result[/red]")
raise typer.Exit(1)

minimized_trace = Trace(
agent_name=trace.agent_name,
run_id=trace.run_id,
created_at=trace.created_at,
status=trace.status,
failure=trace.failure,
environment=trace.environment,
)
minimized_trace = create_minimized_trace(trace, minimal_subset)
for event in result.minimized_events:
minimized_trace.add(event.model_copy(deep=True))
minimized_trace.save(output, overwrite=overwrite)
except Exception as exc:
console.print(f"\n[red]Error saving minimized trace:[/red] {exc}")
Expand Down
1 change: 1 addition & 0 deletions src/stepfork/cli/replay.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ def replay_command(
str,
typer.Option(
"--entrypoint",
"-e",
help="Trusted local MODULE:FUNCTION to execute under replay.",
),
],
Expand Down
16 changes: 12 additions & 4 deletions src/stepfork/minimize/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -169,16 +169,18 @@ def minimize(self) -> MinimizationResult:
error_message="Minimized trace does not reproduce the original failure",
)

# Success
# Success - the trace is minimal when we successfully complete minimization
# (either we reduced it, or it was already at its smallest possible form)
return MinimizationResult(
success=True,
original_event_count=len(self.trace.events),
minimized_event_count=len(minimized_trace.events),
replay_attempts=self.total_replay_attempts,
preserved_failure=True,
is_minimal=len(minimized_trace.events) < len(self.trace.events),
is_minimal=True,
failure_signature=original_signature,
nondeterministic=self.nondeterministic_failures > 0,
minimized_events=tuple(minimized_trace.events),
)

def _test_subset_reproduces_failure(
Expand Down Expand Up @@ -269,8 +271,14 @@ def generate_report(
"Results may vary across runs."
)

if not result.is_minimal and result.success:
warnings.append("Trace was already minimal or could not be reduced further.")
# Check if no reduction was possible (already minimal)
already_minimal = (
result.success
and result.minimized_event_count is not None
and result.minimized_event_count == result.original_event_count
)
if already_minimal:
warnings.append("Trace was already minimal.")

reduction_percentage: float | None = None
if result.minimized_event_count is not None and result.original_event_count > 0:
Expand Down
6 changes: 5 additions & 1 deletion src/stepfork/minimize/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,13 @@
from __future__ import annotations

from dataclasses import dataclass
from typing import Literal
from typing import TYPE_CHECKING, Literal

from stepfork.trace.models import JsonValue

if TYPE_CHECKING:
from stepfork.trace.models import Event


@dataclass(frozen=True)
class FailureSignature:
Expand Down Expand Up @@ -133,6 +136,7 @@ class MinimizationResult:
failure_signature: FailureSignature | None = None
error_message: str | None = None
nondeterministic: bool = False
minimized_events: tuple[Event, ...] | None = None


@dataclass(frozen=True)
Expand Down
32 changes: 32 additions & 0 deletions tests/unit/test_cli_export.py
Original file line number Diff line number Diff line change
Expand Up @@ -362,3 +362,35 @@ def test_export_cli_invalid_expect_output_json(tmp_path: Path) -> None:

assert result.exit_code == 2
assert "not valid JSON" in result.stdout


def test_export_cli_short_flag_e_for_entrypoint(tmp_path: Path) -> None:
"""The -e short flag works for --entrypoint on export command.

Regression test for Finding 4 item 3: -e existed on minimize but not
on replay or export.
"""
trace = tmp_path / "booking.sftrace"
_record_booking(trace)
output = tmp_path / "test_regression.py"

# Use -e instead of --entrypoint
result = runner.invoke(
app,
[
"export",
str(trace),
"--pytest",
"-e", # Use short flag
FIXED_ENTRYPOINT,
"--expect-output",
str(EXPECTED_FILE),
"--output",
str(output),
],
)

assert result.exit_code == 0
assert output.is_file()
# Should not complain about unknown option -e
assert "no such option" not in result.stdout.lower()
132 changes: 132 additions & 0 deletions tests/unit/test_cli_minimize.py
Original file line number Diff line number Diff line change
Expand Up @@ -241,3 +241,135 @@ def test_minimize_command_respects_overwrite_flag(

# May succeed or fail based on minimization result, but not due to overwrite
assert result.exit_code in (0, 1)


def test_minimize_already_minimal_shows_minimal_yes(tmp_path: Path) -> None:
"""When a trace is already minimal, the report shows 'Minimal: Yes' not 'No'.

Regression test for Finding 4 item 1: 'Minimal: No' was incorrectly shown
when the trace was already at its smallest possible form.
"""
from stepfork import FailureInfo, Trace
from stepfork.trace import ErrorEvent, RunEnd, RunStart, RunStatus

# Create a minimal failing trace (just run_start, error, run_end)
trace_path = tmp_path / "minimal.sftrace"
trace = Trace(agent_name="minimal-agent")
trace.add(RunStart())
trace.add(ErrorEvent(error_type="ValueError", message="test error"))
trace.add(RunEnd(run_status=RunStatus.FAILED))
trace.status = RunStatus.FAILED
trace.failure = FailureInfo(type="ValueError", message="test error", step=1)
trace.save(trace_path)

# Agent that raises the same error
output_path = tmp_path / "out.sftrace"
result = runner.invoke(
app,
[
"minimize",
str(trace_path),
"-e",
"test_cli_minimize:_raise_value_error",
"--output",
str(output_path),
],
)

# When already minimal, report should show "Yes" not "No"
assert "Minimal" in result.stdout
assert "[green]Yes[/green]" in result.stdout or "Yes" in result.stdout
# Should NOT show contradictory "No" next to "already minimal"
assert not (
"[yellow]No[/yellow]" in result.stdout
and "already minimal" in result.stdout.lower()
)


def _raise_value_error() -> None:
"""Helper for test_minimize_already_minimal_shows_minimal_yes."""
raise ValueError("test error")


def test_minimize_failed_trace_keeps_run_end(tmp_path: Path) -> None:
"""Minimizing a failed trace preserves run_end event.

Regression test for Finding 4 item 2: run_end was being dropped from
minimized failed traces even though docs say run markers are kept.
"""
from stepfork import FailureInfo, Trace
from stepfork.trace import (
ErrorEvent,
EventType,
RunEnd,
RunStart,
RunStatus,
ToolCall,
ToolResult,
)

# Create a failing trace with multiple events
trace_path = tmp_path / "failed.sftrace"
trace = Trace(agent_name="failed-agent")
trace.add(RunStart())
trace.add(ToolCall(name="some_tool", input={"x": 1}))
trace.add(ToolResult(name="some_tool", output={"y": 2}))
trace.add(ErrorEvent(error_type="ValueError", message="boom"))
trace.add(RunEnd(run_status=RunStatus.FAILED))
trace.status = RunStatus.FAILED
trace.failure = FailureInfo(type="ValueError", message="boom", step=3)
trace.save(trace_path)

# Minimize with an agent that raises the same error
output_path = tmp_path / "minimized.sftrace"
result = runner.invoke(
app,
[
"minimize",
str(trace_path),
"-e",
"test_cli_minimize:_raise_value_error_boom",
"--output",
str(output_path),
],
)

# Check that minimization succeeded
if result.exit_code == 0 and output_path.exists():
minimized = Trace.load(output_path)
event_types = [e.type for e in minimized.events]

# run_start and run_end must be preserved per docs/minimize.md
assert EventType.RUN_START in event_types, "run_start should be preserved"
assert EventType.RUN_END in event_types, "run_end should be preserved"


def _raise_value_error_boom() -> None:
"""Helper for test_minimize_failed_trace_keeps_run_end."""
raise ValueError("boom")


def test_minimize_short_flag_e_for_entrypoint(test_trace: Path, tmp_path: Path) -> None:
"""The -e short flag works for --entrypoint on minimize command."""
expected_output_file = tmp_path / "expected.json"
expected_output_file.write_text(json.dumps({"value": 999}))
output_path = tmp_path / "minimized.sftrace"

result = runner.invoke(
app,
[
"minimize",
str(test_trace),
"-e", # Use short flag
"test_cli_minimize:simple_agent",
"--expect-output",
str(expected_output_file),
"--output",
str(output_path),
],
)

# Should work with -e flag (may succeed or fail on minimization result)
assert result.exit_code in (0, 1)
# Should not complain about unknown option -e
assert "no such option" not in result.stdout.lower()
22 changes: 22 additions & 0 deletions tests/unit/test_cli_replay.py
Original file line number Diff line number Diff line change
Expand Up @@ -365,3 +365,25 @@ def test_replay_cli_reports_unexpected_exception_type(
assert result.exit_code == 1
assert "TypeError" in result.stdout
assert "DIVERGED" in result.stdout


def test_replay_cli_short_flag_e_for_entrypoint(tmp_path: Path) -> None:
"""The -e short flag works for --entrypoint on replay command.

Regression test for Finding 4 item 3: -e existed on minimize but not
on replay or export.
"""
trace = tmp_path / "booking.sftrace"
_record_booking(trace)

# Use -e instead of --entrypoint
result = runner.invoke(
app,
["replay", str(trace), "-e", ENTRYPOINT, "--mode", "frozen"],
catch_exceptions=False,
)

assert result.exit_code == 0
assert "COMPLETED" in result.stdout
# Should not complain about unknown option -e
assert "no such option" not in result.stdout.lower()