From ad0f95793f1649c8c937c2176464e29140ad33f3 Mon Sep 17 00:00:00 2001 From: PhilosophiMoonbeam <94211695+PhilosophiMoonbeam@users.noreply.github.com> Date: Mon, 29 Jun 2026 21:05:09 +0000 Subject: [PATCH 1/4] fix(codex): harden native structured output handling --- swe_af/hitl/ask_user.py | 22 +++-- swe_af/reasoners/execution_agents.py | 12 ++- swe_af/runtime/codex_harness_patch.py | 137 ++++++++++++++++++++++---- tests/test_codex_harness_patch.py | 88 ++++++++++++++++- tests/test_coding_loop_regressions.py | 64 +++++++++++- 5 files changed, 293 insertions(+), 30 deletions(-) diff --git a/swe_af/hitl/ask_user.py b/swe_af/hitl/ask_user.py index 0b247fb6..703535c2 100644 --- a/swe_af/hitl/ask_user.py +++ b/swe_af/hitl/ask_user.py @@ -72,6 +72,13 @@ def approval_webhook_url(app: Any) -> str | None: ] +class AskUserFormOption(BaseModel): + """One selectable option for select, radio, or checkbox_group fields.""" + + value: str = Field(description="Submitted value for this option.") + label: str = Field(description="Human-readable label shown to the user.") + + class AskUserFormField(BaseModel): """One field in a form the agent is constructing for the user.""" @@ -103,7 +110,7 @@ class AskUserFormField(BaseModel): default=None, description="Pre-filled value if the user submits without changing it.", ) - options: list[dict[str, str]] | None = Field( + options: list[AskUserFormOption] | None = Field( default=None, description=( "Required for 'select', 'radio', 'checkbox_group'. Each entry is " @@ -217,6 +224,7 @@ def _field_to_form_builder_call(form: Any, field: AskUserFormField) -> None: common["default_value"] = field.default_value ftype = field.type + options = [option.model_dump() for option in field.options or []] if ftype == "input": form.input(field.id, **common) @@ -243,19 +251,19 @@ def _field_to_form_builder_call(form: Any, field: AskUserFormField) -> None: kwargs["step"] = field.step form.slider(field.id, **kwargs) elif ftype == "select": - if not field.options: + if not options: raise ValueError(f"select field '{field.id}' requires options") - form.select(field.id, options=field.options, **common) + form.select(field.id, options=options, **common) elif ftype == "radio": - if not field.options: + if not options: raise ValueError(f"radio field '{field.id}' requires options") - form.radio_group(field.id, options=field.options, **common) + form.radio_group(field.id, options=options, **common) elif ftype == "checkbox_group": - if not field.options: + if not options: raise ValueError( f"checkbox_group field '{field.id}' requires options" ) - form.checkbox_group(field.id, options=field.options, **common) + form.checkbox_group(field.id, options=options, **common) elif ftype == "checkbox": common.pop("placeholder", None) form.checkbox(field.id, checkbox_label=field.label, **common) diff --git a/swe_af/reasoners/execution_agents.py b/swe_af/reasoners/execution_agents.py index d6399b26..2f8ce9d6 100644 --- a/swe_af/reasoners/execution_agents.py +++ b/swe_af/reasoners/execution_agents.py @@ -1250,13 +1250,21 @@ async def run_qa_synthesizer( workspace_manifest=ws_manifest, ) + provider = runtime_to_harness_adapter(ai_provider) + cwd = worktree_path or target_repo or "." + try: - result = await router.ai( + result = await router.harness( task_prompt, - system=QA_SYNTHESIZER_SYSTEM_PROMPT, + system_prompt=QA_SYNTHESIZER_SYSTEM_PROMPT, schema=QASynthesisResult, model=model, + provider=provider, + cwd=cwd, + max_turns=DEFAULT_AGENT_MAX_TURNS, + permission_mode=permission_mode or None, ) + check_fatal_harness_error(result) if result.parsed is not None: router.note( f"QA synthesizer complete: action={result.parsed.action.value}, " diff --git a/swe_af/runtime/codex_harness_patch.py b/swe_af/runtime/codex_harness_patch.py index 96eaa3fb..722985ca 100644 --- a/swe_af/runtime/codex_harness_patch.py +++ b/swe_af/runtime/codex_harness_patch.py @@ -4,6 +4,8 @@ import contextvars import json import os +import shutil +import tempfile from pathlib import Path from typing import Any @@ -16,6 +18,9 @@ active_provider: contextvars.ContextVar[str | None] = contextvars.ContextVar( "swe_af_codex_active_provider", default=None ) +active_output_paths: contextvars.ContextVar[dict[str, str] | None] = contextvars.ContextVar( + "swe_af_codex_output_paths", default=None +) _ORIGINAL_BUILD_PROMPT_SUFFIX: Any = None @@ -23,6 +28,10 @@ def _codex_strict_json_schema(schema: dict[str, Any]) -> dict[str, Any]: if not isinstance(schema, dict): return schema + if not schema: + # Codex/OpenAI structured output rejects unconstrained `{}` schemas, + # including Pydantic `Any` branches inside `anyOf`. + return {"type": "string"} strict = dict(schema) schema_type = strict.get("type") if schema_type == "object": @@ -39,6 +48,23 @@ def _codex_strict_json_schema(schema: dict[str, Any]) -> dict[str, Any]: strict["properties"] = cleaned strict["required"] = list(cleaned.keys()) strict["additionalProperties"] = False + else: + additional = strict.get("additionalProperties") + if additional is True: + # Codex/OpenAI strict structured output does not accept + # free-form maps. Keep the field object-shaped for Pydantic, + # but require it to be empty. + strict["properties"] = {} + strict["required"] = [] + strict["additionalProperties"] = False + elif isinstance(additional, dict): + strict["properties"] = {} + strict["required"] = [] + strict["additionalProperties"] = False + else: + strict["properties"] = {} + strict["required"] = [] + strict["additionalProperties"] = False if schema_type == "array": items = strict.get("items") if isinstance(items, dict): @@ -81,6 +107,42 @@ def _augment_codex_error_message(message: str, detail: str) -> str: return message +def _codex_no_final_message_error(records: Any) -> tuple[str, bool]: + if not isinstance(records, list): + return ("Codex CLI completed without a final assistant message.", False) + + for record in records: + if not isinstance(record, dict): + continue + payload = record.get("payload") + event = payload if isinstance(payload, dict) else record + if event.get("type") != "token_count": + continue + rate_limits = event.get("rate_limits") + if not isinstance(rate_limits, dict): + continue + credits = rate_limits.get("credits") + if isinstance(credits, dict) and credits.get("has_credits") is False: + limit_id = rate_limits.get("limit_id") or "unknown" + balance = credits.get("balance") + balance_note = f", balance={balance}" if balance is not None else "" + return ( + "Codex CLI completed without a final assistant message because " + f"Codex reported unavailable credits/rate-limit capacity " + f"(limit_id={limit_id}{balance_note}).", + True, + ) + rate_limit_type = rate_limits.get("rate_limit_reached_type") + if rate_limit_type: + return ( + "Codex CLI completed without a final assistant message because " + f"Codex reported a rate limit ({rate_limit_type}).", + True, + ) + + return ("Codex CLI completed without a final assistant message.", False) + + async def _run_codex_cli_with_stdin( cmd: list[str], prompt_for_codex: str, @@ -110,6 +172,7 @@ def apply_codex_harness_patch() -> None: from agentfield.agent import Agent from agentfield.harness import _runner, _schema from agentfield.harness._cli import ( + apply_subprocess_env, estimate_cli_cost, extract_final_text, parse_jsonl, @@ -136,8 +199,17 @@ def build_prompt_suffix_with_schema_file(schema: Any, cwd: str) -> str: _codex_strict_json_schema(_schema.schema_to_json_schema(schema)), indent=2, ) - _schema.write_schema_file(schema_json, cwd) - schema_path = _schema.get_schema_path(cwd) + output_dir = tempfile.mkdtemp(prefix=".agentfield-codex-", dir=cwd) + schema_path = Path(output_dir) / "schema.json" + output_path = Path(output_dir) / "output.json" + schema_path.write_text(schema_json, encoding="utf-8") + active_output_paths.set( + { + "schema": str(schema_path), + "output": str(output_path), + "dir": output_dir, + } + ) return ( "\n\n---\n" "CRITICAL CODEX STRUCTURED OUTPUT REQUIREMENTS:\n" @@ -148,13 +220,15 @@ def build_prompt_suffix_with_schema_file(schema: Any, cwd: str) -> str: ) async def execute_with_native_structured_output(self: Any, prompt: str, options: dict[str, object]) -> Any: - cwd = str(options.get("cwd")) if isinstance(options.get("cwd"), str) else None + root = options.get("project_dir") or options.get("cwd") + cwd = str(root) if isinstance(root, str) else None model = options.get("model") permission_mode = options.get("permission_mode") env_value = options.get("env") merged_env = {**os.environ} if isinstance(env_value, dict): merged_env.update({str(k): str(v) for k, v in env_value.items() if isinstance(k, str)}) + apply_subprocess_env(merged_env) cmd = [self._bin, "exec", "--json", "--skip-git-repo-check"] if cwd: @@ -170,20 +244,24 @@ async def execute_with_native_structured_output(self: Any, prompt: str, options: cmd.extend(["--sandbox", "workspace-write"]) prompt_for_codex = prompt - if cwd: + output_paths = active_output_paths.get() + schema_path = output_paths.get("schema") if output_paths else None + output_path = output_paths.get("output") if output_paths else None + if not schema_path and cwd: schema_path = _schema.get_schema_path(cwd) output_path = _schema.get_output_path(cwd) - if Path(schema_path).exists(): - cmd.extend(["--output-schema", schema_path]) - cmd.extend(["--output-last-message", output_path]) - prompt_for_codex += ( - "\n\n---\n" - "CODEX STRUCTURED OUTPUT CONTRACT:\n" - f"The Codex CLI will save your final response to: {output_path}\n" - f"Your final response MUST be a single JSON object conforming to: {schema_path}\n" - "Return the JSON object as your final answer. Do not write " - "the output file yourself or make the output file the task." - ) + + if schema_path and output_path and Path(schema_path).exists(): + cmd.extend(["--output-schema", schema_path]) + cmd.extend(["--output-last-message", output_path]) + prompt_for_codex += ( + "\n\n---\n" + "CODEX STRUCTURED OUTPUT CONTRACT:\n" + f"The Codex CLI will save your final response to: {output_path}\n" + f"Your final response MUST be a single JSON object conforming to: {schema_path}\n" + "Return the JSON object as your final answer. Do not write " + "the output file yourself or make the output file the task." + ) try: start = asyncio.get_running_loop().time() @@ -233,8 +311,7 @@ async def execute_with_native_structured_output(self: Any, prompt: str, options: records = parse_jsonl(stdout or "") result_text = extract_final_text(records) or "" - if not result_text and cwd: - output_path = _schema.get_output_path(cwd) + if not result_text and output_path: output_file = Path(output_path) if output_file.exists(): try: @@ -245,10 +322,26 @@ async def execute_with_native_structured_output(self: Any, prompt: str, options: is_error = returncode != 0 error_message = "" failure_type = FailureType.NONE + if not result_text: + error_message, is_api_error = _codex_no_final_message_error(records) + is_error = True + failure_type = FailureType.API_ERROR if is_api_error else FailureType.NO_OUTPUT if is_error: - base_error = stderr_clean or "Codex CLI failed" + stdout_error = "" + if isinstance(records, list): + for record in records: + if isinstance(record, dict) and record.get("type") in { + "error", + "turn.failed", + }: + stdout_error = json.dumps(record, ensure_ascii=False) + break + base_error = "\n".join( + part for part in (stderr_clean, stdout_error) if part + ) or error_message or "Codex CLI failed" error_message = _augment_codex_error_message(base_error, base_error) - failure_type = FailureType.CRASH + if returncode != 0: + failure_type = FailureType.CRASH return RawResult( result=result_text, @@ -289,9 +382,15 @@ async def _harness_with_provider_context( ) -> Any: provider_value = kwargs.get("provider") token = active_provider.set(str(provider_value) if provider_value else None) + output_token = active_output_paths.set(None) try: return await _orig_agent_harness(self, prompt, *args, **kwargs) finally: + output_paths = active_output_paths.get() + tmp_dir = output_paths.get("dir") if output_paths else None + if tmp_dir: + shutil.rmtree(tmp_dir, ignore_errors=True) + active_output_paths.reset(output_token) active_provider.reset(token) _schema.build_prompt_suffix = build_prompt_suffix_dispatching diff --git a/tests/test_codex_harness_patch.py b/tests/test_codex_harness_patch.py index 3919e977..47e0f607 100644 --- a/tests/test_codex_harness_patch.py +++ b/tests/test_codex_harness_patch.py @@ -1,8 +1,12 @@ from __future__ import annotations +import shutil + from swe_af.runtime.codex_harness_patch import ( _augment_codex_error_message, + _codex_no_final_message_error, _codex_strict_json_schema, + active_output_paths, active_provider, apply_codex_harness_patch, ) @@ -50,6 +54,26 @@ def test_codex_strict_json_schema_recurses_into_defs() -> None: assert "default" not in item["properties"]["count"] +def test_codex_strict_json_schema_seals_free_form_maps() -> None: + schema = { + "type": "object", + "properties": { + "agent_retro": { + "title": "Agent Retro", + "type": "object", + "additionalProperties": {"type": "string"}, + }, + }, + } + + strict = _codex_strict_json_schema(schema) + + agent_retro = strict["properties"]["agent_retro"] + assert agent_retro["properties"] == {} + assert agent_retro["required"] == [] + assert agent_retro["additionalProperties"] is False + + def test_codex_git_metadata_error_gets_actionable_hint() -> None: message = _augment_codex_error_message( "fatal: cannot create .git/index.lock", @@ -64,12 +88,65 @@ def test_codex_unrelated_error_is_unchanged() -> None: assert _augment_codex_error_message("plain error", "plain error") == "plain error" +def test_codex_no_final_message_reports_unavailable_credits() -> None: + message, is_api_error = _codex_no_final_message_error( + [ + { + "type": "event_msg", + "payload": { + "type": "token_count", + "rate_limits": { + "limit_id": "premium", + "credits": { + "has_credits": False, + "balance": "0", + "unlimited": False, + }, + }, + }, + } + ] + ) + + assert is_api_error is True + assert "unavailable credits/rate-limit capacity" in message + assert "limit_id=premium" in message + assert "balance=0" in message + + +def test_codex_no_final_message_reports_rate_limit_type() -> None: + message, is_api_error = _codex_no_final_message_error( + [ + { + "payload": { + "type": "token_count", + "rate_limits": { + "rate_limit_reached_type": "requests", + }, + }, + } + ] + ) + + assert is_api_error is True + assert "rate limit (requests)" in message + + +def test_codex_no_final_message_without_rate_limit_is_no_output() -> None: + message, is_api_error = _codex_no_final_message_error([]) + + assert is_api_error is False + assert message == "Codex CLI completed without a final assistant message." + + def test_codex_prompt_suffix_uses_final_json_not_write_tool(tmp_path) -> None: from agentfield.harness import _schema apply_codex_harness_patch() token = active_provider.set("codex") + output_token = active_output_paths.set(None) + output_paths = None try: suffix = _schema.build_prompt_suffix( { @@ -78,12 +155,21 @@ def test_codex_prompt_suffix_uses_final_json_not_write_tool(tmp_path) -> None: }, str(tmp_path), ) + output_paths = active_output_paths.get() finally: + if output_paths is not None: + shutil.rmtree(output_paths.get("dir", ""), ignore_errors=True) + active_output_paths.reset(output_token) active_provider.reset(token) assert "Return a single final JSON object" in suffix assert "Write tool" not in suffix - assert (tmp_path / ".agentfield_schema.json").exists() + assert output_paths is not None + assert output_paths["schema"] != str(tmp_path / ".agentfield_schema.json") + assert output_paths["output"] != str(tmp_path / ".agentfield_output.json") + assert output_paths["schema"].startswith(str(tmp_path / ".agentfield-codex-")) + assert output_paths["output"].startswith(str(tmp_path / ".agentfield-codex-")) + assert (tmp_path / ".agentfield_schema.json").exists() is False def test_non_codex_prompt_suffix_keeps_agentfield_write_tool_default(tmp_path) -> None: diff --git a/tests/test_coding_loop_regressions.py b/tests/test_coding_loop_regressions.py index b9af44d7..013c11e3 100644 --- a/tests/test_coding_loop_regressions.py +++ b/tests/test_coding_loop_regressions.py @@ -3,7 +3,12 @@ from pathlib import Path from swe_af.execution.coding_loop import run_coding_loop -from swe_af.execution.schemas import DAGState, ExecutionConfig, IssueOutcome +from swe_af.execution.schemas import ( + DAGState, + ExecutionConfig, + IssueOutcome, + QASynthesisResult, +) def _make_dag_state(tmp_path: Path, build_id: str) -> DAGState: @@ -98,3 +103,60 @@ async def call_fn(target: str, **kwargs): assert result.outcome == IssueOutcome.COMPLETED for agent_name in ("run_coder", "run_qa", "run_code_reviewer", "run_qa_synthesizer"): assert observed_modes[agent_name] == "bypassPermissions" + + +def test_run_qa_synthesizer_uses_provider_aware_harness_for_codex( + tmp_path: Path, + monkeypatch, +) -> None: + from swe_af.reasoners import execution_agents + + observed: dict[str, object] = {} + + class FakeAgent: + async def harness(self, prompt: str, **kwargs): + observed["prompt"] = prompt + observed.update(kwargs) + + class Result: + parsed = QASynthesisResult( + action="approve", + summary="ok", + stuck=False, + ) + + return Result() + + async def ai(self, *args, **kwargs): # pragma: no cover - should never run + raise AssertionError("QA synthesizer must use router.harness, not router.ai") + + def note(self, *args, **kwargs) -> None: + return None + + monkeypatch.setattr(execution_agents.router, "_agent", FakeAgent()) + + result = asyncio.run( + execution_agents.run_qa_synthesizer( + qa_result={"passed": True, "summary": "qa ok", "test_failures": []}, + review_result={ + "approved": True, + "blocking": False, + "summary": "review ok", + "debt_items": [], + }, + iteration_history=[], + iteration_id="it1", + worktree_path=str(tmp_path), + model="gpt-5.5", + permission_mode="auto", + ai_provider="codex", + ) + ) + + assert result["action"] == "approve" + assert result["iteration_id"] == "it1" + assert observed["model"] == "gpt-5.5" + assert observed["provider"] == "codex" + assert observed["cwd"] == str(tmp_path) + assert observed["permission_mode"] == "auto" + assert observed["schema"] is QASynthesisResult From 7bd4b512e17ae0cc36cc4d3664322a3f342bd04b Mon Sep 17 00:00:00 2001 From: PhilosophiMoonbeam <94211695+PhilosophiMoonbeam@users.noreply.github.com> Date: Wed, 1 Jul 2026 00:49:13 +0000 Subject: [PATCH 2/4] fix(codex): avoid workspace-write defaults --- swe_af/execution/dag_executor.py | 12 ++++++++++-- swe_af/runtime/codex_harness_patch.py | 13 +++++++------ tests/test_codex_harness_patch.py | 19 +++++++++++++++++++ 3 files changed, 36 insertions(+), 8 deletions(-) diff --git a/swe_af/execution/dag_executor.py b/swe_af/execution/dag_executor.py index 4a8a5d5f..23320ee9 100644 --- a/swe_af/execution/dag_executor.py +++ b/swe_af/execution/dag_executor.py @@ -88,6 +88,7 @@ async def _setup_worktrees( artifacts_dir=dag_state.artifacts_dir, level=dag_state.current_level, model=config.git_model, + permission_mode=config.permission_mode, ai_provider=config.ai_provider, build_id=build_id, ) @@ -258,6 +259,7 @@ async def _merge_level_branches( artifacts_dir=dag_state.artifacts_dir, level=level_result.level_index, model=config.merger_model, + permission_mode=config.permission_mode, ai_provider=config.ai_provider, ) @@ -348,6 +350,7 @@ async def _call_merger_for_repo( artifacts_dir=dag_state.artifacts_dir, level=level_result.level_index, model=config.merger_model, + permission_mode=config.permission_mode, ai_provider=config.ai_provider, ) return result @@ -459,6 +462,7 @@ async def _run_integration_tests( artifacts_dir=dag_state.artifacts_dir, level=level_result.level_index, model=config.integration_tester_model, + permission_mode=config.permission_mode, ai_provider=config.ai_provider, workspace_manifest=dag_state.workspace_manifest, ) @@ -491,6 +495,7 @@ async def _cleanup_worktrees( level: int = 0, model: str = "sonnet", ai_provider: str = "claude", + permission_mode: str = "", completed_results: list | None = None, ) -> None: """Remove worktrees and clean up branches after merge. @@ -527,7 +532,7 @@ async def _cleanup_worktrees( await _cleanup_single_repo( call_fn, node_id, ws_repo.absolute_path, repo_worktrees_dir, repo_branches, dag_state.artifacts_dir, level, model, ai_provider, - note_fn, + note_fn, permission_mode, ) return @@ -535,7 +540,7 @@ async def _cleanup_worktrees( await _cleanup_single_repo( call_fn, node_id, dag_state.repo_path, dag_state.worktrees_dir, branches_to_clean, dag_state.artifacts_dir, level, model, ai_provider, - note_fn, + note_fn, permission_mode, ) @@ -550,6 +555,7 @@ async def _cleanup_single_repo( model: str, ai_provider: str, note_fn: Callable | None = None, + permission_mode: str = "", ) -> None: """Clean up worktrees for a single repo. Retries once on failure.""" for attempt in range(2): # up to 1 retry @@ -562,6 +568,7 @@ async def _cleanup_single_repo( artifacts_dir=artifacts_dir, level=level, model=model, + permission_mode=permission_mode, ai_provider=ai_provider, ) if result.get("success"): @@ -1591,6 +1598,7 @@ async def _memory_fn(action: str, key: str, value=None): level=dag_state.current_level, model=config.git_model, ai_provider=config.ai_provider, + permission_mode=config.permission_mode, completed_results=level_result.completed, ) ) diff --git a/swe_af/runtime/codex_harness_patch.py b/swe_af/runtime/codex_harness_patch.py index 722985ca..53a5bafe 100644 --- a/swe_af/runtime/codex_harness_patch.py +++ b/swe_af/runtime/codex_harness_patch.py @@ -143,6 +143,12 @@ def _codex_no_final_message_error(records: Any) -> tuple[str, bool]: return ("Codex CLI completed without a final assistant message.", False) +def _codex_permission_args(permission_mode: object) -> list[str]: + if permission_mode in {"read-only", "workspace-write"}: + return ["--sandbox", str(permission_mode)] + return ["--dangerously-bypass-approvals-and-sandbox"] + + async def _run_codex_cli_with_stdin( cmd: list[str], prompt_for_codex: str, @@ -236,12 +242,7 @@ async def execute_with_native_structured_output(self: Any, prompt: str, options: if model: cmd.extend(["-m", str(model)]) - if permission_mode == "auto": - cmd.append("--dangerously-bypass-approvals-and-sandbox") - elif permission_mode in {"read-only", "workspace-write", "danger-full-access"}: - cmd.extend(["--sandbox", str(permission_mode)]) - else: - cmd.extend(["--sandbox", "workspace-write"]) + cmd.extend(_codex_permission_args(permission_mode)) prompt_for_codex = prompt output_paths = active_output_paths.get() diff --git a/tests/test_codex_harness_patch.py b/tests/test_codex_harness_patch.py index 47e0f607..7a00878e 100644 --- a/tests/test_codex_harness_patch.py +++ b/tests/test_codex_harness_patch.py @@ -5,6 +5,7 @@ from swe_af.runtime.codex_harness_patch import ( _augment_codex_error_message, _codex_no_final_message_error, + _codex_permission_args, _codex_strict_json_schema, active_output_paths, active_provider, @@ -88,6 +89,24 @@ def test_codex_unrelated_error_is_unchanged() -> None: assert _augment_codex_error_message("plain error", "plain error") == "plain error" +def test_codex_default_permission_mode_bypasses_sandbox() -> None: + assert _codex_permission_args(None) == ["--dangerously-bypass-approvals-and-sandbox"] + assert _codex_permission_args("") == ["--dangerously-bypass-approvals-and-sandbox"] + assert _codex_permission_args("auto") == ["--dangerously-bypass-approvals-and-sandbox"] + assert _codex_permission_args("default") == ["--dangerously-bypass-approvals-and-sandbox"] + assert _codex_permission_args("danger-full-access") == [ + "--dangerously-bypass-approvals-and-sandbox" + ] + assert _codex_permission_args("bypassPermissions") == [ + "--dangerously-bypass-approvals-and-sandbox" + ] + + +def test_codex_explicit_narrow_permission_modes_are_preserved() -> None: + assert _codex_permission_args("read-only") == ["--sandbox", "read-only"] + assert _codex_permission_args("workspace-write") == ["--sandbox", "workspace-write"] + + def test_codex_no_final_message_reports_unavailable_credits() -> None: message, is_api_error = _codex_no_final_message_error( [ From e8d9c308d7d8585b84d3165a17d025afaf46fe65 Mon Sep 17 00:00:00 2001 From: PhilosophiMoonbeam <94211695+PhilosophiMoonbeam@users.noreply.github.com> Date: Thu, 2 Jul 2026 01:40:13 +0000 Subject: [PATCH 3/4] fix(resume): split execute and build recovery --- README.md | 7 +- swe_af/app.py | 752 ++++++++++++++++++++++++++++++++-- tests/test_planner_execute.py | 175 +++++++- 3 files changed, 893 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index ad76371a..c64b1604 100644 --- a/README.md +++ b/README.md @@ -169,7 +169,7 @@ Most agent frameworks wrap a single coder loop. SWE-AF is a coordinated engineer - **Agent-scale parallelism** — dependency-level scheduling + isolated git worktrees allow large fan-out without branch collisions. - **Fleet-scale orchestration** — many SWE-AF nodes can run continuously in parallel via AgentField, driving thousands of agent invocations across concurrent builds. - **Explicit compromise tracking** — when scope is relaxed, debt is typed, severity-rated, and propagated. -- **Long-run reliability** — checkpointed execution supports `resume_build` after crashes or interruptions. +- **Long-run reliability** — checkpointed execution supports `resume_execute` for DAG-only recovery and `resume_build` for full build recovery through verification, finalization, PR creation, and CI gating. ## In Action @@ -623,7 +623,10 @@ POST /api/v1/execute/async/swe-planner.plan # Execute a prebuilt plan POST /api/v1/execute/async/swe-planner.execute -# Resume after interruption +# Resume DAG execution after interruption +POST /api/v1/execute/async/swe-planner.resume_execute + +# Resume full build after interruption POST /api/v1/execute/async/swe-planner.resume_build ``` diff --git a/swe_af/app.py b/swe_af/app.py index af855a3c..607b51fa 100644 --- a/swe_af/app.py +++ b/swe_af/app.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import json import os import subprocess import uuid @@ -58,6 +59,8 @@ def __init__(self, message: str, *, result=None, error_details=None) -> None: app.include_router(router) +BUILD_STATE_FILENAME = "build_state.json" + # --------------------------------------------------------------------------- # Auto-inject scoped credentials into every router.harness call. @@ -485,6 +488,607 @@ def _is_empty_build(success: bool, ever_completed: int, ever_merged: int) -> boo return not success and ever_completed == 0 and ever_merged == 0 +def _absolute_artifacts_dir(repo_path: str, artifacts_dir: str) -> str: + if os.path.isabs(artifacts_dir): + return artifacts_dir + return os.path.join(os.path.abspath(repo_path), artifacts_dir) + + +def _build_state_path(repo_path: str, artifacts_dir: str) -> str: + return os.path.join(_absolute_artifacts_dir(repo_path, artifacts_dir), BUILD_STATE_FILENAME) + + +def _save_build_state(repo_path: str, artifacts_dir: str, state: dict) -> None: + path = _build_state_path(repo_path, artifacts_dir) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fp: + json.dump(state, fp, indent=2) + + +def _load_build_state(repo_path: str, artifacts_dir: str) -> dict: + path = _build_state_path(repo_path, artifacts_dir) + if not os.path.exists(path): + return {} + with open(path, encoding="utf-8") as fp: + return json.load(fp) + + +def _build_config_from_saved_state( + stored_config: dict, + overrides: dict | None = None, +) -> BuildConfig: + data = dict(stored_config or {}) + data.update(overrides or {}) + + # BuildConfig normalizes repo_url into repos, so model_dump() may contain + # both fields. Rehydrate through the canonical multi-repo form. + if data.get("repo_url") and data.get("repos"): + data.pop("repo_url") + + return BuildConfig(**data) if data else BuildConfig() + + +def _checkpoint_path(repo_path: str, artifacts_dir: str) -> str: + return os.path.join(_absolute_artifacts_dir(repo_path, artifacts_dir), "execution", "checkpoint.json") + + +def _load_execution_checkpoint(repo_path: str, artifacts_dir: str) -> dict: + path = _checkpoint_path(repo_path, artifacts_dir) + if not os.path.exists(path): + raise RuntimeError(f"No checkpoint found at {path}. Cannot resume.") + with open(path, encoding="utf-8") as fp: + return json.load(fp) + + +def _plan_result_from_checkpoint(checkpoint: dict) -> dict: + artifacts_dir = checkpoint.get("artifacts_dir", "") + return { + "prd": {}, + "architecture": {}, + "review": {}, + "issues": checkpoint.get("all_issues", []), + "levels": checkpoint.get("levels", []), + "file_conflicts": [], + "artifacts_dir": artifacts_dir, + "rationale": checkpoint.get("original_plan_summary", ""), + } + + +def _git_config_from_checkpoint(checkpoint: dict) -> dict | None: + integration_branch = checkpoint.get("git_integration_branch", "") + if not integration_branch: + return None + return { + "integration_branch": integration_branch, + "original_branch": checkpoint.get("git_original_branch", ""), + "initial_commit_sha": checkpoint.get("git_initial_commit", ""), + "mode": checkpoint.get("git_mode", ""), + "remote_url": checkpoint.get("git_remote_url", ""), + "remote_default_branch": checkpoint.get("git_remote_default_branch", ""), + } + + +def _read_plan_docs(plan_result: dict) -> tuple[str, str]: + plan_dir = os.path.join(plan_result.get("artifacts_dir", ""), "plan") + docs: dict[str, str] = {"prd.md": "", "architecture.md": ""} + for name in docs: + path = os.path.join(plan_dir, name) + if os.path.isfile(path): + try: + with open(path, encoding="utf-8") as fp: + docs[name] = fp.read() + except OSError: + pass + return docs["prd.md"], docs["architecture.md"] + + +def _resume_incomplete_summary(result: dict) -> str: + """Return a short failure summary when a resumed DAG is still incomplete.""" + failed = result.get("failed_issues", []) or [] + skipped = result.get("skipped_issues", []) or [] + if not failed and not skipped: + return "" + + failed_names = [ + item.get("issue_name", "") if isinstance(item, dict) else str(item) + for item in failed + ] + skipped_names = [ + item.get("issue_name", "") if isinstance(item, dict) else str(item) + for item in skipped + ] + completed = len(result.get("completed_issues", []) or []) + total = len(result.get("all_issues", []) or []) + parts = [f"Resume incomplete: {completed}/{total} issues completed"] + if failed_names: + parts.append(f"failed={failed_names}") + if skipped_names: + parts.append(f"skipped={skipped_names}") + return "; ".join(parts) + + +def _existing_pr_for_branch(repo_path: str, branch: str, base_branch: str) -> dict: + """Return an open PR for branch when one already exists, else empty dict.""" + if not branch: + return {} + cmd = [ + "gh", "pr", "list", + "--head", branch, + "--base", base_branch, + "--state", "open", + "--json", "url,number", + "--limit", "1", + ] + res = subprocess.run(cmd, cwd=repo_path, capture_output=True, text=True) + if res.returncode != 0: + return {} + try: + matches = json.loads(res.stdout or "[]") + except json.JSONDecodeError: + return {} + if not matches: + return {} + first = matches[0] + return { + "success": True, + "pr_url": first.get("url", ""), + "pr_number": first.get("number", 0), + } + + +def _push_existing_pr_branch(repo_path: str, branch: str) -> None: + if not branch: + return + subprocess.run( + ["git", "push", "origin", branch], + cwd=repo_path, + capture_output=True, + text=True, + ) + + +def _append_plan_docs_to_pr( + *, + repo_path: str, + pr_number: int, + prd_markdown: str, + architecture_markdown: str, +) -> None: + if not pr_number or not (prd_markdown or architecture_markdown): + return + current_body = subprocess.run( + ["gh", "pr", "view", str(pr_number), "--json", "body", "--jq", ".body"], + cwd=repo_path, + capture_output=True, + text=True, + check=True, + ).stdout.strip() + + if ( + "PRD (Product Requirements Document)" in current_body + or "Architecture" in current_body + ): + return + + plan_sections = "\n\n---\n" + if prd_markdown: + plan_sections += ( + "\n
📋 PRD (Product Requirements Document)" + "\n\n" + + prd_markdown + + "\n\n
\n" + ) + if architecture_markdown: + plan_sections += ( + "\n
🏗️ Architecture\n\n" + + architecture_markdown + + "\n\n
\n" + ) + + subprocess.run( + ["gh", "pr", "edit", str(pr_number), "--body", current_body + plan_sections], + cwd=repo_path, + capture_output=True, + text=True, + check=True, + ) + + +async def _continue_build_tail( + *, + goal: str, + repo_path: str, + artifacts_dir: str, + cfg: BuildConfig, + resolved: dict[str, str], + plan_result: dict, + dag_result: dict, + git_config: dict | None, + manifest: WorkspaceManifest | None, + ever_completed: int, + ever_merged: int, + build_state: dict | None = None, +) -> dict: + """Run the post-DAG build tail shared by build() and resume_build().""" + build_state = dict(build_state or {}) + if manifest and dag_result.get("workspace_manifest"): + manifest = WorkspaceManifest(**dag_result["workspace_manifest"]) + + exec_config = cfg.to_execution_config_dict() + verification = build_state.get("verification") + for cycle in range(cfg.max_verify_fix_cycles + 1): + app.note(f"Verification cycle {cycle}", tags=["build", "verify"]) + verification = _unwrap(await app.call( + f"{NODE_ID}.run_verifier", + prd=plan_result.get("prd", {}), + repo_path=repo_path, + artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), + completed_issues=[r for r in dag_result.get("completed_issues", [])], + failed_issues=[r for r in dag_result.get("failed_issues", [])], + skipped_issues=dag_result.get("skipped_issues", []), + model=resolved["verifier_model"], + permission_mode=cfg.permission_mode, + ai_provider=cfg.ai_provider, + workspace_manifest=manifest.model_dump() if manifest else None, + ), "run_verifier") + build_state["verification"] = verification + + if verification.get("passed", False) or cycle >= cfg.max_verify_fix_cycles: + break + + failed_criteria = [ + c for c in verification.get("criteria_results", []) + if not c.get("passed", True) + ] + if not failed_criteria: + app.note("Verification failed but no specific criteria failures found", tags=["build", "verify"]) + break + + app.note( + f"Verification failed ({len(failed_criteria)} criteria), " + f"{cfg.max_verify_fix_cycles - cycle} fix cycles remaining", + tags=["build", "verify", "retry"], + ) + fix_result = _unwrap(await app.call( + f"{NODE_ID}.generate_fix_issues", + failed_criteria=failed_criteria, + dag_state=dag_result, + prd=plan_result.get("prd", {}), + artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), + model=resolved["verifier_model"], + permission_mode=cfg.permission_mode, + ai_provider=cfg.ai_provider, + workspace_manifest=manifest.model_dump() if manifest else None, + ), "generate_fix_issues") + + for debt in fix_result.get("debt_items", []): + dag_result.setdefault("accumulated_debt", []).append({ + "type": "unmet_acceptance_criterion", + "criterion": debt.get("criterion", ""), + "reason": debt.get("reason", ""), + "severity": debt.get("severity", "high"), + }) + + fix_issues = fix_result.get("fix_issues", []) + if not fix_issues: + app.note("No fixable issues generated — accepting with debt", tags=["build", "verify"]) + break + + fix_plan = { + "prd": plan_result.get("prd", {}), + "architecture": plan_result.get("architecture", {}), + "review": plan_result.get("review", {}), + "issues": fix_issues, + "levels": [[fi.get("name", f"fix-{i}") for i, fi in enumerate(fix_issues)]], + "file_conflicts": [], + "artifacts_dir": plan_result.get("artifacts_dir", artifacts_dir), + "rationale": f"Fix issues for verification cycle {cycle + 1}", + } + dag_result = _unwrap(await app.call( + f"{NODE_ID}.execute", + plan_result=fix_plan, + repo_path=repo_path, + config=exec_config, + git_config=git_config, + workspace_manifest=manifest.model_dump() if manifest else None, + ), "execute_fixes") + ever_completed = max(ever_completed, len(dag_result.get("completed_issues", []) or [])) + ever_merged = max(ever_merged, len(dag_result.get("merged_branches", []) or [])) + build_state["dag_result"] = dag_result + + success = verification.get("passed", False) if verification else False + completed = len(dag_result.get("completed_issues", [])) + total = len(dag_result.get("all_issues", [])) + app.note( + f"Build {'succeeded' if success else 'completed with issues'}: " + f"{completed}/{total} issues, verification={'passed' if success else 'failed'}", + tags=["build", "complete"], + ) + + prd_markdown, architecture_markdown = _read_plan_docs(plan_result) + + if manifest and len(manifest.repos) > 1: + app.note( + f"Phase 3b: Multi-repo finalization ({len(manifest.repos)} repos)", + tags=["build", "finalize", "multi-repo"], + ) + for ws_repo in manifest.repos: + try: + finalize_result = _unwrap(await app.call( + f"{NODE_ID}.run_repo_finalize", + repo_path=ws_repo.absolute_path, + artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), + model=resolved["git_model"], + permission_mode=cfg.permission_mode, + ai_provider=cfg.ai_provider, + ), f"run_repo_finalize ({ws_repo.repo_name})") + if finalize_result.get("success"): + app.note( + f"Repo finalized ({ws_repo.repo_name}): {finalize_result.get('summary', '')}", + tags=["build", "finalize", "complete"], + ) + else: + app.note( + f"Repo finalize incomplete ({ws_repo.repo_name}): {finalize_result.get('summary', '')}", + tags=["build", "finalize", "warning"], + ) + except Exception as e: + app.note( + f"Repo finalize failed for {ws_repo.repo_name} (non-blocking): {e}", + tags=["build", "finalize", "error"], + ) + else: + app.note("Phase 3b: Repo finalization", tags=["build", "finalize"]) + try: + finalize_result = _unwrap(await app.call( + f"{NODE_ID}.run_repo_finalize", + repo_path=repo_path, + artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), + model=resolved["git_model"], + permission_mode=cfg.permission_mode, + ai_provider=cfg.ai_provider, + ), "run_repo_finalize") + if finalize_result.get("success"): + app.note( + f"Repo finalized: {finalize_result.get('summary', '')}", + tags=["build", "finalize", "complete"], + ) + else: + app.note( + f"Repo finalize incomplete: {finalize_result.get('summary', '')}", + tags=["build", "finalize", "warning"], + ) + except Exception as e: + app.note( + f"Repo finalize failed (non-blocking): {e}", + tags=["build", "finalize", "error"], + ) + + pr_results: list[RepoPRResult] = [] + ci_gate_results: list[dict] = [] + build_summary = ( + f"{'Success' if success else 'Partial'}: {completed}/{total} issues completed" + + (f", verification: {verification.get('summary', '')}" if verification else "") + ) + + if manifest and len(manifest.repos) > 1: + app.note("Phase 4: Multi-repo Push + PRs", tags=["build", "github_pr", "multi-repo"]) + for ws_repo in manifest.repos: + if not ws_repo.create_pr or not cfg.enable_github_pr: + continue + repo_git_init = ws_repo.git_init_result or {} + repo_remote_url = repo_git_init.get("remote_url", "") or ws_repo.repo_url + if not repo_remote_url: + continue + repo_integration_branch = repo_git_init.get("integration_branch", "") + if not repo_integration_branch: + continue + repo_base_branch = ( + cfg.github_pr_base + or repo_git_init.get("remote_default_branch", "") + or "main" + ) + try: + existing = _existing_pr_for_branch( + ws_repo.absolute_path, repo_integration_branch, repo_base_branch + ) + if existing: + _push_existing_pr_branch(ws_repo.absolute_path, repo_integration_branch) + pr_r = existing + else: + pr_r = _unwrap(await app.call( + f"{NODE_ID}.run_github_pr", + repo_path=ws_repo.absolute_path, + integration_branch=repo_integration_branch, + base_branch=repo_base_branch, + goal=goal, + build_summary=build_summary, + completed_issues=[ + r for r in dag_result.get("completed_issues", []) + if not r.get("repo_name") or r.get("repo_name") == ws_repo.repo_name + ], + accumulated_debt=dag_result.get("accumulated_debt", []), + artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), + model=resolved["git_model"], + permission_mode=cfg.permission_mode, + ai_provider=cfg.ai_provider, + ), "run_github_pr") + pr_results.append(RepoPRResult( + repo_name=ws_repo.repo_name, + repo_url=ws_repo.repo_url, + success=pr_r.get("success", False), + pr_url=pr_r.get("pr_url", ""), + pr_number=pr_r.get("pr_number", 0), + error_message=pr_r.get("error_message", ""), + )) + if pr_r.get("pr_url"): + app.note( + f"PR ready for {ws_repo.repo_name}: {pr_r.get('pr_url')}", + tags=["build", "github_pr", "complete"], + ) + if cfg.check_ci and pr_r.get("pr_number"): + gate = await _run_ci_gate( + repo_path=ws_repo.absolute_path, + pr_number=pr_r.get("pr_number", 0), + pr_url=pr_r.get("pr_url", ""), + integration_branch=repo_integration_branch, + base_branch=repo_base_branch, + cfg=cfg, + resolved_models=resolved, + goal=goal, + completed_issues=[ + r for r in dag_result.get("completed_issues", []) + if not r.get("repo_name") or r.get("repo_name") == ws_repo.repo_name + ], + ) + ci_gate_results.append({"repo_name": ws_repo.repo_name, **gate}) + except Exception as e: + pr_results.append(RepoPRResult( + repo_name=ws_repo.repo_name, + repo_url=ws_repo.repo_url, + success=False, + error_message=str(e), + )) + app.note( + f"PR creation failed for {ws_repo.repo_name}: {e}", + tags=["build", "github_pr", "error"], + ) + else: + remote_url = git_config.get("remote_url", "") if git_config else "" + if remote_url and cfg.enable_github_pr: + app.note("Phase 4: Push + PR", tags=["build", "github_pr"]) + base_branch = ( + cfg.github_pr_base + or (git_config.get("remote_default_branch") if git_config else "") + or "main" + ) + pr_url = "" + try: + existing = _existing_pr_for_branch( + repo_path, git_config["integration_branch"], base_branch + ) + pr_existed = bool(existing) + if existing: + _push_existing_pr_branch(repo_path, git_config["integration_branch"]) + pr_result = existing + else: + pr_result = _unwrap(await app.call( + f"{NODE_ID}.run_github_pr", + repo_path=repo_path, + integration_branch=git_config["integration_branch"], + base_branch=base_branch, + goal=goal, + build_summary=build_summary, + completed_issues=dag_result.get("completed_issues", []), + accumulated_debt=dag_result.get("accumulated_debt", []), + artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), + model=resolved["git_model"], + permission_mode=cfg.permission_mode, + ai_provider=cfg.ai_provider, + ), "run_github_pr") + pr_url = pr_result.get("pr_url", "") + if pr_url: + app.note(f"PR ready: {pr_url}", tags=["build", "github_pr", "complete"]) + if not pr_existed: + try: + _append_plan_docs_to_pr( + repo_path=repo_path, + pr_number=pr_result.get("pr_number", 0), + prd_markdown=prd_markdown, + architecture_markdown=architecture_markdown, + ) + app.note( + "Plan docs appended to PR body", + tags=["build", "github_pr", "plan_docs"], + ) + except subprocess.CalledProcessError as e: + app.note( + f"Failed to append plan docs to PR (non-fatal): {e}", + tags=["build", "github_pr", "plan_docs", "warning"], + ) + else: + app.note( + f"PR creation failed: {pr_result.get('error_message', 'unknown')}", + tags=["build", "github_pr", "error"], + ) + if pr_url: + pr_results.append(RepoPRResult( + repo_name=_repo_name_from_url(cfg.repo_url) if cfg.repo_url else "repo", + repo_url=cfg.repo_url, + success=True, + pr_url=pr_url, + pr_number=pr_result.get("pr_number", 0), + )) + if cfg.check_ci and pr_result.get("pr_number"): + gate = await _run_ci_gate( + repo_path=repo_path, + pr_number=pr_result.get("pr_number", 0), + pr_url=pr_url, + integration_branch=git_config["integration_branch"], + base_branch=base_branch, + cfg=cfg, + resolved_models=resolved, + goal=goal, + completed_issues=dag_result.get("completed_issues", []), + ) + ci_gate_results.append({ + "repo_name": ( + _repo_name_from_url(cfg.repo_url) + if cfg.repo_url else "repo" + ), + **gate, + }) + except Exception as e: + app.note(f"PR creation failed: {e}", tags=["build", "github_pr", "error"]) + + if manifest and manifest.workspace_root: + try: + import shutil + shutil.rmtree(manifest.workspace_root, ignore_errors=True) + app.note( + f"Workspace cleaned up: {manifest.workspace_root}", + tags=["build", "cleanup"], + ) + except Exception: + pass + + build_result = BuildResult( + plan_result=plan_result, + dag_state=dag_result, + verification=verification, + success=success, + summary=f"{'Success' if success else 'Partial'}: {completed}/{total} issues completed" + + (f", verification: {verification.get('summary', '')}" if verification else ""), + pr_results=pr_results, + ci_gate_results=ci_gate_results, + ).model_dump() + + build_state.update({ + "goal": goal, + "repo_path": repo_path, + "repo_url": cfg.repo_url, + "artifacts_dir": artifacts_dir, + "config": cfg.model_dump(), + "plan_result": plan_result, + "dag_result": dag_result, + "git_config": git_config, + "workspace_manifest": manifest.model_dump() if manifest else None, + "verification": verification, + "pr_results": [r.model_dump() for r in pr_results], + "ci_gate_results": ci_gate_results, + "build_result": build_result, + }) + _save_build_state(repo_path, artifacts_dir, build_state) + + if _is_empty_build(success, ever_completed, ever_merged): + raise ReasonerFailed( + f"Build failed: 0/{total} issues completed, no branches merged", + result=build_result, + ) + + return build_result + + @app.reasoner() async def build( goal: str, @@ -961,6 +1565,19 @@ async def build( summary=f"Plan {approval_result.decision}: {reason}", ).model_dump() + build_state = { + "goal": goal, + "repo_path": repo_path, + "repo_url": cfg.repo_url, + "artifacts_dir": artifacts_dir, + "config": cfg.model_dump(), + "build_id": build_id, + "plan_result": plan_result, + "git_config": git_config, + "workspace_manifest": manifest.model_dump() if manifest else None, + } + _save_build_state(repo_path, artifacts_dir, build_state) + # 2. EXECUTE exec_config = cfg.to_execution_config_dict() @@ -984,6 +1601,8 @@ async def build( # genuinely shipped something. ever_completed = len(dag_result.get("completed_issues", []) or []) ever_merged = len(dag_result.get("merged_branches", []) or []) + build_state["dag_result"] = dag_result + _save_build_state(repo_path, artifacts_dir, build_state) # Refresh manifest with git_init_result populated by _init_all_repos() in # the DAG executor. Must happen before the verify/fix loop which can @@ -1386,6 +2005,14 @@ async def build( pr_results=pr_results, ci_gate_results=ci_gate_results, ).model_dump() + build_state.update({ + "dag_result": dag_result, + "verification": verification, + "pr_results": [r.model_dump() for r in pr_results], + "ci_gate_results": ci_gate_results, + "build_result": build_result, + }) + _save_build_state(repo_path, artifacts_dir, build_state) # An empty build — verification failed AND nothing was ever completed # or merged across the original run and every fix cycle — must not @@ -2060,60 +2687,109 @@ async def _post_thread_replies_and_resolve( return results -@app.reasoner() -async def resume_build( +async def _resume_execute_impl( repo_path: str, artifacts_dir: str = ".artifacts", config: dict | None = None, git_config: dict | None = None, ) -> dict: - """Resume a crashed build from the last checkpoint. - - Loads the plan result from artifacts and calls execute with resume=True. - """ - import json - - base = os.path.join(os.path.abspath(repo_path), artifacts_dir) + checkpoint = _load_execution_checkpoint(repo_path, artifacts_dir) + build_state = _load_build_state(repo_path, artifacts_dir) + plan_result = build_state.get("plan_result") or _plan_result_from_checkpoint(checkpoint) + effective_git_config = git_config or build_state.get("git_config") or _git_config_from_checkpoint(checkpoint) - # Reconstruct plan_result from saved artifacts - plan_path = os.path.join(base, "execution", "checkpoint.json") - if not os.path.exists(plan_path): - raise RuntimeError( - f"No checkpoint found at {plan_path}. Cannot resume." - ) + app.note("Resuming DAG execution from checkpoint", tags=["execute", "resume"]) - # Load the original plan artifacts to reconstruct plan_result - prd_path = os.path.join(base, "plan", "prd.md") - arch_path = os.path.join(base, "plan", "architecture.md") - rationale_path = os.path.join(base, "rationale.md") + result = _unwrap(await app.call( + f"{NODE_ID}.execute", + plan_result=plan_result, + repo_path=repo_path, + config=config, + git_config=effective_git_config, + resume=True, + ), "execute") - # We need the plan_result dict — reconstruct from checkpoint's DAGState - with open(plan_path, "r") as f: - checkpoint = json.load(f) + incomplete = _resume_incomplete_summary(result) + if incomplete: + raise ReasonerFailed(incomplete, result=result) - plan_result = { - "prd": {}, # Not needed for resume — DAGState has summaries - "architecture": {}, - "review": {}, - "issues": checkpoint.get("all_issues", []), - "levels": checkpoint.get("levels", []), - "file_conflicts": [], - "artifacts_dir": checkpoint.get("artifacts_dir", base), - "rationale": checkpoint.get("original_plan_summary", ""), - } + return result - app.note("Resuming build from checkpoint", tags=["build", "resume"]) - result = await app.call( - f"{NODE_ID}.execute", - plan_result=plan_result, +@app.reasoner() +async def resume_execute( + repo_path: str, + artifacts_dir: str = ".artifacts", + config: dict | None = None, + git_config: dict | None = None, +) -> dict: + """Resume DAG execution only from ``.artifacts/execution/checkpoint.json``.""" + return await _resume_execute_impl( repo_path=repo_path, + artifacts_dir=artifacts_dir, config=config, git_config=git_config, - resume=True, ) - return result + +@app.reasoner() +async def resume_build( + repo_path: str, + artifacts_dir: str = ".artifacts", + config: dict | None = None, + git_config: dict | None = None, + goal: str = "", + repo_url: str = "", +) -> dict: + """Resume DAG execution, then continue verifier/finalize/PR/CI build tail.""" + checkpoint = _load_execution_checkpoint(repo_path, artifacts_dir) + build_state = _load_build_state(repo_path, artifacts_dir) + overrides = dict(config or {}) + if repo_url: + overrides["repo_url"] = repo_url + cfg = _build_config_from_saved_state(build_state.get("config") or {}, overrides) + + effective_goal = goal or build_state.get("goal", "") + plan_result = build_state.get("plan_result") or _plan_result_from_checkpoint(checkpoint) + effective_git_config = git_config or build_state.get("git_config") or _git_config_from_checkpoint(checkpoint) + manifest_data = build_state.get("workspace_manifest") or checkpoint.get("workspace_manifest") + manifest = WorkspaceManifest(**manifest_data) if manifest_data else None + + app.note("Resuming full build from checkpoint", tags=["build", "resume"]) + exec_config = cfg.to_execution_config_dict() + dag_result = await _resume_execute_impl( + repo_path=repo_path, + artifacts_dir=artifacts_dir, + config=exec_config, + git_config=effective_git_config, + ) + build_state.update({ + "goal": effective_goal, + "repo_path": repo_path, + "repo_url": cfg.repo_url, + "artifacts_dir": artifacts_dir, + "config": cfg.model_dump(), + "plan_result": plan_result, + "dag_result": dag_result, + "git_config": effective_git_config, + "workspace_manifest": manifest.model_dump() if manifest else None, + }) + _save_build_state(repo_path, artifacts_dir, build_state) + + return await _continue_build_tail( + goal=effective_goal, + repo_path=repo_path, + artifacts_dir=artifacts_dir, + cfg=cfg, + resolved=cfg.resolved_models(), + plan_result=plan_result, + dag_result=dag_result, + git_config=effective_git_config, + manifest=manifest, + ever_completed=len(dag_result.get("completed_issues", []) or []), + ever_merged=len(dag_result.get("merged_branches", []) or []), + build_state=build_state, + ) def main(): diff --git a/tests/test_planner_execute.py b/tests/test_planner_execute.py index dd04fe46..6301a802 100644 --- a/tests/test_planner_execute.py +++ b/tests/test_planner_execute.py @@ -11,7 +11,9 @@ from __future__ import annotations -from unittest.mock import AsyncMock, MagicMock, patch, call as mock_call +import json +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -47,6 +49,12 @@ def _make_plan_result(issues: list[dict] | None = None) -> dict: } +def _write_checkpoint(repo_path: Path, checkpoint: dict) -> None: + checkpoint_path = repo_path / ".artifacts" / "execution" / "checkpoint.json" + checkpoint_path.parent.mkdir(parents=True, exist_ok=True) + checkpoint_path.write_text(json.dumps(checkpoint), encoding="utf-8") + + def _make_dag_state(completed: list[str], failed: list[str]) -> DAGState: """Build a DAGState with given completed / failed issue names.""" return DAGState( @@ -70,6 +78,171 @@ def _make_dag_state(completed: list[str], failed: list[str]) -> DAGState: ) +def test_resume_incomplete_summary_empty_for_finished_dag(): + import swe_af.app as app_module + + result = { + "all_issues": [{"name": "one"}], + "completed_issues": [{"issue_name": "one"}], + "failed_issues": [], + "skipped_issues": [], + } + + assert app_module._resume_incomplete_summary(result) == "" + + +def test_resume_incomplete_summary_flags_failed_and_skipped_dag(): + import swe_af.app as app_module + + result = { + "all_issues": [{"name": "one"}, {"name": "two"}, {"name": "three"}], + "completed_issues": [{"issue_name": "one"}], + "failed_issues": [{"issue_name": "two"}], + "skipped_issues": ["three"], + } + + summary = app_module._resume_incomplete_summary(result) + + assert "Resume incomplete: 1/3 issues completed" in summary + assert "failed=['two']" in summary + assert "skipped=['three']" in summary + + +def test_build_config_saved_state_round_trips_normalized_repo_url(): + import swe_af.app as app_module + from swe_af.execution.schemas import BuildConfig + + saved_config = BuildConfig( + repo_url="https://github.com/example/repo.git", + ).model_dump() + + cfg = app_module._build_config_from_saved_state(saved_config) + + assert cfg.repo_url == "https://github.com/example/repo.git" + assert len(cfg.repos) == 1 + + +@pytest.mark.asyncio +async def test_resume_execute_only_resumes_dag(tmp_path: Path): + import swe_af.app as app_module + + repo_path = tmp_path / "repo" + repo_path.mkdir() + plan_result = _make_plan_result() + completed_result = { + "all_issues": plan_result["issues"], + "completed_issues": [{"issue_name": "implement-feature"}], + "failed_issues": [], + "skipped_issues": [], + "merged_branches": ["feature-branch"], + } + _write_checkpoint( + repo_path, + { + "all_issues": plan_result["issues"], + "levels": plan_result["levels"], + "artifacts_dir": str(repo_path / ".artifacts"), + "git_integration_branch": "integration", + }, + ) + + async def fake_call(target: str, **kwargs): + assert target == f"{app_module.NODE_ID}.execute" + assert kwargs["resume"] is True + return completed_result + + with patch.object(app_module.app, "call", new=AsyncMock(side_effect=fake_call)): + result = await app_module.resume_execute(repo_path=str(repo_path)) + + assert result == completed_result + + +@pytest.mark.asyncio +async def test_resume_build_continues_full_tail(tmp_path: Path): + import swe_af.app as app_module + + repo_path = tmp_path / "repo" + repo_path.mkdir() + plan_result = _make_plan_result() + dag_result = { + "all_issues": plan_result["issues"], + "completed_issues": [{"issue_name": "implement-feature"}], + "failed_issues": [], + "skipped_issues": [], + "merged_branches": ["feature-branch"], + "accumulated_debt": [], + } + git_config = { + "integration_branch": "integration", + "original_branch": "main", + "initial_commit_sha": "abc123", + "mode": "existing", + "remote_url": "https://github.com/example/repo.git", + "remote_default_branch": "main", + } + _write_checkpoint( + repo_path, + { + "all_issues": plan_result["issues"], + "levels": plan_result["levels"], + "artifacts_dir": str(repo_path / ".artifacts"), + "git_integration_branch": "integration", + }, + ) + app_module._save_build_state( + str(repo_path), + ".artifacts", + { + "goal": "ship feature", + "repo_path": str(repo_path), + "repo_url": "https://github.com/example/repo.git", + "artifacts_dir": ".artifacts", + "config": { + "runtime": "codex", + "models": {"default": "gpt-5.5"}, + "enable_github_pr": True, + "check_ci": False, + }, + "plan_result": plan_result, + "git_config": git_config, + }, + ) + + calls: list[str] = [] + + async def fake_call(target: str, **kwargs): + calls.append(target) + if target == f"{app_module.NODE_ID}.execute": + assert kwargs["resume"] is True + return dag_result + if target == f"{app_module.NODE_ID}.run_verifier": + return {"passed": True, "summary": "ok", "criteria_results": []} + if target == f"{app_module.NODE_ID}.run_repo_finalize": + return {"success": True, "summary": "clean"} + if target == f"{app_module.NODE_ID}.run_github_pr": + return { + "success": True, + "pr_url": "https://github.com/example/repo/pull/1", + "pr_number": 1, + } + raise AssertionError(f"unexpected call: {target}") + + with ( + patch.object(app_module.app, "call", new=AsyncMock(side_effect=fake_call)), + patch.object(app_module, "_existing_pr_for_branch", return_value={}), + patch.object(app_module, "_append_plan_docs_to_pr"), + ): + result = await app_module.resume_build(repo_path=str(repo_path)) + + assert result["success"] is True + assert calls == [ + f"{app_module.NODE_ID}.execute", + f"{app_module.NODE_ID}.run_verifier", + f"{app_module.NODE_ID}.run_repo_finalize", + f"{app_module.NODE_ID}.run_github_pr", + ] + + # --------------------------------------------------------------------------- # test_execute_single_issue # --------------------------------------------------------------------------- From 35e26a81b9f987727421b7538fed897d6bacf7b8 Mon Sep 17 00:00:00 2001 From: PhilosophiMoonbeam <94211695+PhilosophiMoonbeam@users.noreply.github.com> Date: Thu, 2 Jul 2026 02:39:00 +0000 Subject: [PATCH 4/4] Revert "fix(resume): split execute and build recovery" This reverts commit e8d9c308d7d8585b84d3165a17d025afaf46fe65. --- README.md | 7 +- swe_af/app.py | 752 ++-------------------------------- tests/test_planner_execute.py | 175 +------- 3 files changed, 41 insertions(+), 893 deletions(-) diff --git a/README.md b/README.md index c64b1604..ad76371a 100644 --- a/README.md +++ b/README.md @@ -169,7 +169,7 @@ Most agent frameworks wrap a single coder loop. SWE-AF is a coordinated engineer - **Agent-scale parallelism** — dependency-level scheduling + isolated git worktrees allow large fan-out without branch collisions. - **Fleet-scale orchestration** — many SWE-AF nodes can run continuously in parallel via AgentField, driving thousands of agent invocations across concurrent builds. - **Explicit compromise tracking** — when scope is relaxed, debt is typed, severity-rated, and propagated. -- **Long-run reliability** — checkpointed execution supports `resume_execute` for DAG-only recovery and `resume_build` for full build recovery through verification, finalization, PR creation, and CI gating. +- **Long-run reliability** — checkpointed execution supports `resume_build` after crashes or interruptions. ## In Action @@ -623,10 +623,7 @@ POST /api/v1/execute/async/swe-planner.plan # Execute a prebuilt plan POST /api/v1/execute/async/swe-planner.execute -# Resume DAG execution after interruption -POST /api/v1/execute/async/swe-planner.resume_execute - -# Resume full build after interruption +# Resume after interruption POST /api/v1/execute/async/swe-planner.resume_build ``` diff --git a/swe_af/app.py b/swe_af/app.py index 607b51fa..af855a3c 100644 --- a/swe_af/app.py +++ b/swe_af/app.py @@ -9,7 +9,6 @@ from __future__ import annotations import asyncio -import json import os import subprocess import uuid @@ -59,8 +58,6 @@ def __init__(self, message: str, *, result=None, error_details=None) -> None: app.include_router(router) -BUILD_STATE_FILENAME = "build_state.json" - # --------------------------------------------------------------------------- # Auto-inject scoped credentials into every router.harness call. @@ -488,607 +485,6 @@ def _is_empty_build(success: bool, ever_completed: int, ever_merged: int) -> boo return not success and ever_completed == 0 and ever_merged == 0 -def _absolute_artifacts_dir(repo_path: str, artifacts_dir: str) -> str: - if os.path.isabs(artifacts_dir): - return artifacts_dir - return os.path.join(os.path.abspath(repo_path), artifacts_dir) - - -def _build_state_path(repo_path: str, artifacts_dir: str) -> str: - return os.path.join(_absolute_artifacts_dir(repo_path, artifacts_dir), BUILD_STATE_FILENAME) - - -def _save_build_state(repo_path: str, artifacts_dir: str, state: dict) -> None: - path = _build_state_path(repo_path, artifacts_dir) - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w", encoding="utf-8") as fp: - json.dump(state, fp, indent=2) - - -def _load_build_state(repo_path: str, artifacts_dir: str) -> dict: - path = _build_state_path(repo_path, artifacts_dir) - if not os.path.exists(path): - return {} - with open(path, encoding="utf-8") as fp: - return json.load(fp) - - -def _build_config_from_saved_state( - stored_config: dict, - overrides: dict | None = None, -) -> BuildConfig: - data = dict(stored_config or {}) - data.update(overrides or {}) - - # BuildConfig normalizes repo_url into repos, so model_dump() may contain - # both fields. Rehydrate through the canonical multi-repo form. - if data.get("repo_url") and data.get("repos"): - data.pop("repo_url") - - return BuildConfig(**data) if data else BuildConfig() - - -def _checkpoint_path(repo_path: str, artifacts_dir: str) -> str: - return os.path.join(_absolute_artifacts_dir(repo_path, artifacts_dir), "execution", "checkpoint.json") - - -def _load_execution_checkpoint(repo_path: str, artifacts_dir: str) -> dict: - path = _checkpoint_path(repo_path, artifacts_dir) - if not os.path.exists(path): - raise RuntimeError(f"No checkpoint found at {path}. Cannot resume.") - with open(path, encoding="utf-8") as fp: - return json.load(fp) - - -def _plan_result_from_checkpoint(checkpoint: dict) -> dict: - artifacts_dir = checkpoint.get("artifacts_dir", "") - return { - "prd": {}, - "architecture": {}, - "review": {}, - "issues": checkpoint.get("all_issues", []), - "levels": checkpoint.get("levels", []), - "file_conflicts": [], - "artifacts_dir": artifacts_dir, - "rationale": checkpoint.get("original_plan_summary", ""), - } - - -def _git_config_from_checkpoint(checkpoint: dict) -> dict | None: - integration_branch = checkpoint.get("git_integration_branch", "") - if not integration_branch: - return None - return { - "integration_branch": integration_branch, - "original_branch": checkpoint.get("git_original_branch", ""), - "initial_commit_sha": checkpoint.get("git_initial_commit", ""), - "mode": checkpoint.get("git_mode", ""), - "remote_url": checkpoint.get("git_remote_url", ""), - "remote_default_branch": checkpoint.get("git_remote_default_branch", ""), - } - - -def _read_plan_docs(plan_result: dict) -> tuple[str, str]: - plan_dir = os.path.join(plan_result.get("artifacts_dir", ""), "plan") - docs: dict[str, str] = {"prd.md": "", "architecture.md": ""} - for name in docs: - path = os.path.join(plan_dir, name) - if os.path.isfile(path): - try: - with open(path, encoding="utf-8") as fp: - docs[name] = fp.read() - except OSError: - pass - return docs["prd.md"], docs["architecture.md"] - - -def _resume_incomplete_summary(result: dict) -> str: - """Return a short failure summary when a resumed DAG is still incomplete.""" - failed = result.get("failed_issues", []) or [] - skipped = result.get("skipped_issues", []) or [] - if not failed and not skipped: - return "" - - failed_names = [ - item.get("issue_name", "") if isinstance(item, dict) else str(item) - for item in failed - ] - skipped_names = [ - item.get("issue_name", "") if isinstance(item, dict) else str(item) - for item in skipped - ] - completed = len(result.get("completed_issues", []) or []) - total = len(result.get("all_issues", []) or []) - parts = [f"Resume incomplete: {completed}/{total} issues completed"] - if failed_names: - parts.append(f"failed={failed_names}") - if skipped_names: - parts.append(f"skipped={skipped_names}") - return "; ".join(parts) - - -def _existing_pr_for_branch(repo_path: str, branch: str, base_branch: str) -> dict: - """Return an open PR for branch when one already exists, else empty dict.""" - if not branch: - return {} - cmd = [ - "gh", "pr", "list", - "--head", branch, - "--base", base_branch, - "--state", "open", - "--json", "url,number", - "--limit", "1", - ] - res = subprocess.run(cmd, cwd=repo_path, capture_output=True, text=True) - if res.returncode != 0: - return {} - try: - matches = json.loads(res.stdout or "[]") - except json.JSONDecodeError: - return {} - if not matches: - return {} - first = matches[0] - return { - "success": True, - "pr_url": first.get("url", ""), - "pr_number": first.get("number", 0), - } - - -def _push_existing_pr_branch(repo_path: str, branch: str) -> None: - if not branch: - return - subprocess.run( - ["git", "push", "origin", branch], - cwd=repo_path, - capture_output=True, - text=True, - ) - - -def _append_plan_docs_to_pr( - *, - repo_path: str, - pr_number: int, - prd_markdown: str, - architecture_markdown: str, -) -> None: - if not pr_number or not (prd_markdown or architecture_markdown): - return - current_body = subprocess.run( - ["gh", "pr", "view", str(pr_number), "--json", "body", "--jq", ".body"], - cwd=repo_path, - capture_output=True, - text=True, - check=True, - ).stdout.strip() - - if ( - "PRD (Product Requirements Document)" in current_body - or "Architecture" in current_body - ): - return - - plan_sections = "\n\n---\n" - if prd_markdown: - plan_sections += ( - "\n
📋 PRD (Product Requirements Document)" - "\n\n" - + prd_markdown - + "\n\n
\n" - ) - if architecture_markdown: - plan_sections += ( - "\n
🏗️ Architecture\n\n" - + architecture_markdown - + "\n\n
\n" - ) - - subprocess.run( - ["gh", "pr", "edit", str(pr_number), "--body", current_body + plan_sections], - cwd=repo_path, - capture_output=True, - text=True, - check=True, - ) - - -async def _continue_build_tail( - *, - goal: str, - repo_path: str, - artifacts_dir: str, - cfg: BuildConfig, - resolved: dict[str, str], - plan_result: dict, - dag_result: dict, - git_config: dict | None, - manifest: WorkspaceManifest | None, - ever_completed: int, - ever_merged: int, - build_state: dict | None = None, -) -> dict: - """Run the post-DAG build tail shared by build() and resume_build().""" - build_state = dict(build_state or {}) - if manifest and dag_result.get("workspace_manifest"): - manifest = WorkspaceManifest(**dag_result["workspace_manifest"]) - - exec_config = cfg.to_execution_config_dict() - verification = build_state.get("verification") - for cycle in range(cfg.max_verify_fix_cycles + 1): - app.note(f"Verification cycle {cycle}", tags=["build", "verify"]) - verification = _unwrap(await app.call( - f"{NODE_ID}.run_verifier", - prd=plan_result.get("prd", {}), - repo_path=repo_path, - artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), - completed_issues=[r for r in dag_result.get("completed_issues", [])], - failed_issues=[r for r in dag_result.get("failed_issues", [])], - skipped_issues=dag_result.get("skipped_issues", []), - model=resolved["verifier_model"], - permission_mode=cfg.permission_mode, - ai_provider=cfg.ai_provider, - workspace_manifest=manifest.model_dump() if manifest else None, - ), "run_verifier") - build_state["verification"] = verification - - if verification.get("passed", False) or cycle >= cfg.max_verify_fix_cycles: - break - - failed_criteria = [ - c for c in verification.get("criteria_results", []) - if not c.get("passed", True) - ] - if not failed_criteria: - app.note("Verification failed but no specific criteria failures found", tags=["build", "verify"]) - break - - app.note( - f"Verification failed ({len(failed_criteria)} criteria), " - f"{cfg.max_verify_fix_cycles - cycle} fix cycles remaining", - tags=["build", "verify", "retry"], - ) - fix_result = _unwrap(await app.call( - f"{NODE_ID}.generate_fix_issues", - failed_criteria=failed_criteria, - dag_state=dag_result, - prd=plan_result.get("prd", {}), - artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), - model=resolved["verifier_model"], - permission_mode=cfg.permission_mode, - ai_provider=cfg.ai_provider, - workspace_manifest=manifest.model_dump() if manifest else None, - ), "generate_fix_issues") - - for debt in fix_result.get("debt_items", []): - dag_result.setdefault("accumulated_debt", []).append({ - "type": "unmet_acceptance_criterion", - "criterion": debt.get("criterion", ""), - "reason": debt.get("reason", ""), - "severity": debt.get("severity", "high"), - }) - - fix_issues = fix_result.get("fix_issues", []) - if not fix_issues: - app.note("No fixable issues generated — accepting with debt", tags=["build", "verify"]) - break - - fix_plan = { - "prd": plan_result.get("prd", {}), - "architecture": plan_result.get("architecture", {}), - "review": plan_result.get("review", {}), - "issues": fix_issues, - "levels": [[fi.get("name", f"fix-{i}") for i, fi in enumerate(fix_issues)]], - "file_conflicts": [], - "artifacts_dir": plan_result.get("artifacts_dir", artifacts_dir), - "rationale": f"Fix issues for verification cycle {cycle + 1}", - } - dag_result = _unwrap(await app.call( - f"{NODE_ID}.execute", - plan_result=fix_plan, - repo_path=repo_path, - config=exec_config, - git_config=git_config, - workspace_manifest=manifest.model_dump() if manifest else None, - ), "execute_fixes") - ever_completed = max(ever_completed, len(dag_result.get("completed_issues", []) or [])) - ever_merged = max(ever_merged, len(dag_result.get("merged_branches", []) or [])) - build_state["dag_result"] = dag_result - - success = verification.get("passed", False) if verification else False - completed = len(dag_result.get("completed_issues", [])) - total = len(dag_result.get("all_issues", [])) - app.note( - f"Build {'succeeded' if success else 'completed with issues'}: " - f"{completed}/{total} issues, verification={'passed' if success else 'failed'}", - tags=["build", "complete"], - ) - - prd_markdown, architecture_markdown = _read_plan_docs(plan_result) - - if manifest and len(manifest.repos) > 1: - app.note( - f"Phase 3b: Multi-repo finalization ({len(manifest.repos)} repos)", - tags=["build", "finalize", "multi-repo"], - ) - for ws_repo in manifest.repos: - try: - finalize_result = _unwrap(await app.call( - f"{NODE_ID}.run_repo_finalize", - repo_path=ws_repo.absolute_path, - artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), - model=resolved["git_model"], - permission_mode=cfg.permission_mode, - ai_provider=cfg.ai_provider, - ), f"run_repo_finalize ({ws_repo.repo_name})") - if finalize_result.get("success"): - app.note( - f"Repo finalized ({ws_repo.repo_name}): {finalize_result.get('summary', '')}", - tags=["build", "finalize", "complete"], - ) - else: - app.note( - f"Repo finalize incomplete ({ws_repo.repo_name}): {finalize_result.get('summary', '')}", - tags=["build", "finalize", "warning"], - ) - except Exception as e: - app.note( - f"Repo finalize failed for {ws_repo.repo_name} (non-blocking): {e}", - tags=["build", "finalize", "error"], - ) - else: - app.note("Phase 3b: Repo finalization", tags=["build", "finalize"]) - try: - finalize_result = _unwrap(await app.call( - f"{NODE_ID}.run_repo_finalize", - repo_path=repo_path, - artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), - model=resolved["git_model"], - permission_mode=cfg.permission_mode, - ai_provider=cfg.ai_provider, - ), "run_repo_finalize") - if finalize_result.get("success"): - app.note( - f"Repo finalized: {finalize_result.get('summary', '')}", - tags=["build", "finalize", "complete"], - ) - else: - app.note( - f"Repo finalize incomplete: {finalize_result.get('summary', '')}", - tags=["build", "finalize", "warning"], - ) - except Exception as e: - app.note( - f"Repo finalize failed (non-blocking): {e}", - tags=["build", "finalize", "error"], - ) - - pr_results: list[RepoPRResult] = [] - ci_gate_results: list[dict] = [] - build_summary = ( - f"{'Success' if success else 'Partial'}: {completed}/{total} issues completed" - + (f", verification: {verification.get('summary', '')}" if verification else "") - ) - - if manifest and len(manifest.repos) > 1: - app.note("Phase 4: Multi-repo Push + PRs", tags=["build", "github_pr", "multi-repo"]) - for ws_repo in manifest.repos: - if not ws_repo.create_pr or not cfg.enable_github_pr: - continue - repo_git_init = ws_repo.git_init_result or {} - repo_remote_url = repo_git_init.get("remote_url", "") or ws_repo.repo_url - if not repo_remote_url: - continue - repo_integration_branch = repo_git_init.get("integration_branch", "") - if not repo_integration_branch: - continue - repo_base_branch = ( - cfg.github_pr_base - or repo_git_init.get("remote_default_branch", "") - or "main" - ) - try: - existing = _existing_pr_for_branch( - ws_repo.absolute_path, repo_integration_branch, repo_base_branch - ) - if existing: - _push_existing_pr_branch(ws_repo.absolute_path, repo_integration_branch) - pr_r = existing - else: - pr_r = _unwrap(await app.call( - f"{NODE_ID}.run_github_pr", - repo_path=ws_repo.absolute_path, - integration_branch=repo_integration_branch, - base_branch=repo_base_branch, - goal=goal, - build_summary=build_summary, - completed_issues=[ - r for r in dag_result.get("completed_issues", []) - if not r.get("repo_name") or r.get("repo_name") == ws_repo.repo_name - ], - accumulated_debt=dag_result.get("accumulated_debt", []), - artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), - model=resolved["git_model"], - permission_mode=cfg.permission_mode, - ai_provider=cfg.ai_provider, - ), "run_github_pr") - pr_results.append(RepoPRResult( - repo_name=ws_repo.repo_name, - repo_url=ws_repo.repo_url, - success=pr_r.get("success", False), - pr_url=pr_r.get("pr_url", ""), - pr_number=pr_r.get("pr_number", 0), - error_message=pr_r.get("error_message", ""), - )) - if pr_r.get("pr_url"): - app.note( - f"PR ready for {ws_repo.repo_name}: {pr_r.get('pr_url')}", - tags=["build", "github_pr", "complete"], - ) - if cfg.check_ci and pr_r.get("pr_number"): - gate = await _run_ci_gate( - repo_path=ws_repo.absolute_path, - pr_number=pr_r.get("pr_number", 0), - pr_url=pr_r.get("pr_url", ""), - integration_branch=repo_integration_branch, - base_branch=repo_base_branch, - cfg=cfg, - resolved_models=resolved, - goal=goal, - completed_issues=[ - r for r in dag_result.get("completed_issues", []) - if not r.get("repo_name") or r.get("repo_name") == ws_repo.repo_name - ], - ) - ci_gate_results.append({"repo_name": ws_repo.repo_name, **gate}) - except Exception as e: - pr_results.append(RepoPRResult( - repo_name=ws_repo.repo_name, - repo_url=ws_repo.repo_url, - success=False, - error_message=str(e), - )) - app.note( - f"PR creation failed for {ws_repo.repo_name}: {e}", - tags=["build", "github_pr", "error"], - ) - else: - remote_url = git_config.get("remote_url", "") if git_config else "" - if remote_url and cfg.enable_github_pr: - app.note("Phase 4: Push + PR", tags=["build", "github_pr"]) - base_branch = ( - cfg.github_pr_base - or (git_config.get("remote_default_branch") if git_config else "") - or "main" - ) - pr_url = "" - try: - existing = _existing_pr_for_branch( - repo_path, git_config["integration_branch"], base_branch - ) - pr_existed = bool(existing) - if existing: - _push_existing_pr_branch(repo_path, git_config["integration_branch"]) - pr_result = existing - else: - pr_result = _unwrap(await app.call( - f"{NODE_ID}.run_github_pr", - repo_path=repo_path, - integration_branch=git_config["integration_branch"], - base_branch=base_branch, - goal=goal, - build_summary=build_summary, - completed_issues=dag_result.get("completed_issues", []), - accumulated_debt=dag_result.get("accumulated_debt", []), - artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir), - model=resolved["git_model"], - permission_mode=cfg.permission_mode, - ai_provider=cfg.ai_provider, - ), "run_github_pr") - pr_url = pr_result.get("pr_url", "") - if pr_url: - app.note(f"PR ready: {pr_url}", tags=["build", "github_pr", "complete"]) - if not pr_existed: - try: - _append_plan_docs_to_pr( - repo_path=repo_path, - pr_number=pr_result.get("pr_number", 0), - prd_markdown=prd_markdown, - architecture_markdown=architecture_markdown, - ) - app.note( - "Plan docs appended to PR body", - tags=["build", "github_pr", "plan_docs"], - ) - except subprocess.CalledProcessError as e: - app.note( - f"Failed to append plan docs to PR (non-fatal): {e}", - tags=["build", "github_pr", "plan_docs", "warning"], - ) - else: - app.note( - f"PR creation failed: {pr_result.get('error_message', 'unknown')}", - tags=["build", "github_pr", "error"], - ) - if pr_url: - pr_results.append(RepoPRResult( - repo_name=_repo_name_from_url(cfg.repo_url) if cfg.repo_url else "repo", - repo_url=cfg.repo_url, - success=True, - pr_url=pr_url, - pr_number=pr_result.get("pr_number", 0), - )) - if cfg.check_ci and pr_result.get("pr_number"): - gate = await _run_ci_gate( - repo_path=repo_path, - pr_number=pr_result.get("pr_number", 0), - pr_url=pr_url, - integration_branch=git_config["integration_branch"], - base_branch=base_branch, - cfg=cfg, - resolved_models=resolved, - goal=goal, - completed_issues=dag_result.get("completed_issues", []), - ) - ci_gate_results.append({ - "repo_name": ( - _repo_name_from_url(cfg.repo_url) - if cfg.repo_url else "repo" - ), - **gate, - }) - except Exception as e: - app.note(f"PR creation failed: {e}", tags=["build", "github_pr", "error"]) - - if manifest and manifest.workspace_root: - try: - import shutil - shutil.rmtree(manifest.workspace_root, ignore_errors=True) - app.note( - f"Workspace cleaned up: {manifest.workspace_root}", - tags=["build", "cleanup"], - ) - except Exception: - pass - - build_result = BuildResult( - plan_result=plan_result, - dag_state=dag_result, - verification=verification, - success=success, - summary=f"{'Success' if success else 'Partial'}: {completed}/{total} issues completed" - + (f", verification: {verification.get('summary', '')}" if verification else ""), - pr_results=pr_results, - ci_gate_results=ci_gate_results, - ).model_dump() - - build_state.update({ - "goal": goal, - "repo_path": repo_path, - "repo_url": cfg.repo_url, - "artifacts_dir": artifacts_dir, - "config": cfg.model_dump(), - "plan_result": plan_result, - "dag_result": dag_result, - "git_config": git_config, - "workspace_manifest": manifest.model_dump() if manifest else None, - "verification": verification, - "pr_results": [r.model_dump() for r in pr_results], - "ci_gate_results": ci_gate_results, - "build_result": build_result, - }) - _save_build_state(repo_path, artifacts_dir, build_state) - - if _is_empty_build(success, ever_completed, ever_merged): - raise ReasonerFailed( - f"Build failed: 0/{total} issues completed, no branches merged", - result=build_result, - ) - - return build_result - - @app.reasoner() async def build( goal: str, @@ -1565,19 +961,6 @@ async def build( summary=f"Plan {approval_result.decision}: {reason}", ).model_dump() - build_state = { - "goal": goal, - "repo_path": repo_path, - "repo_url": cfg.repo_url, - "artifacts_dir": artifacts_dir, - "config": cfg.model_dump(), - "build_id": build_id, - "plan_result": plan_result, - "git_config": git_config, - "workspace_manifest": manifest.model_dump() if manifest else None, - } - _save_build_state(repo_path, artifacts_dir, build_state) - # 2. EXECUTE exec_config = cfg.to_execution_config_dict() @@ -1601,8 +984,6 @@ async def build( # genuinely shipped something. ever_completed = len(dag_result.get("completed_issues", []) or []) ever_merged = len(dag_result.get("merged_branches", []) or []) - build_state["dag_result"] = dag_result - _save_build_state(repo_path, artifacts_dir, build_state) # Refresh manifest with git_init_result populated by _init_all_repos() in # the DAG executor. Must happen before the verify/fix loop which can @@ -2005,14 +1386,6 @@ async def build( pr_results=pr_results, ci_gate_results=ci_gate_results, ).model_dump() - build_state.update({ - "dag_result": dag_result, - "verification": verification, - "pr_results": [r.model_dump() for r in pr_results], - "ci_gate_results": ci_gate_results, - "build_result": build_result, - }) - _save_build_state(repo_path, artifacts_dir, build_state) # An empty build — verification failed AND nothing was ever completed # or merged across the original run and every fix cycle — must not @@ -2687,109 +2060,60 @@ async def _post_thread_replies_and_resolve( return results -async def _resume_execute_impl( +@app.reasoner() +async def resume_build( repo_path: str, artifacts_dir: str = ".artifacts", config: dict | None = None, git_config: dict | None = None, ) -> dict: - checkpoint = _load_execution_checkpoint(repo_path, artifacts_dir) - build_state = _load_build_state(repo_path, artifacts_dir) - plan_result = build_state.get("plan_result") or _plan_result_from_checkpoint(checkpoint) - effective_git_config = git_config or build_state.get("git_config") or _git_config_from_checkpoint(checkpoint) + """Resume a crashed build from the last checkpoint. - app.note("Resuming DAG execution from checkpoint", tags=["execute", "resume"]) + Loads the plan result from artifacts and calls execute with resume=True. + """ + import json - result = _unwrap(await app.call( - f"{NODE_ID}.execute", - plan_result=plan_result, - repo_path=repo_path, - config=config, - git_config=effective_git_config, - resume=True, - ), "execute") + base = os.path.join(os.path.abspath(repo_path), artifacts_dir) - incomplete = _resume_incomplete_summary(result) - if incomplete: - raise ReasonerFailed(incomplete, result=result) + # Reconstruct plan_result from saved artifacts + plan_path = os.path.join(base, "execution", "checkpoint.json") + if not os.path.exists(plan_path): + raise RuntimeError( + f"No checkpoint found at {plan_path}. Cannot resume." + ) - return result + # Load the original plan artifacts to reconstruct plan_result + prd_path = os.path.join(base, "plan", "prd.md") + arch_path = os.path.join(base, "plan", "architecture.md") + rationale_path = os.path.join(base, "rationale.md") + # We need the plan_result dict — reconstruct from checkpoint's DAGState + with open(plan_path, "r") as f: + checkpoint = json.load(f) -@app.reasoner() -async def resume_execute( - repo_path: str, - artifacts_dir: str = ".artifacts", - config: dict | None = None, - git_config: dict | None = None, -) -> dict: - """Resume DAG execution only from ``.artifacts/execution/checkpoint.json``.""" - return await _resume_execute_impl( + plan_result = { + "prd": {}, # Not needed for resume — DAGState has summaries + "architecture": {}, + "review": {}, + "issues": checkpoint.get("all_issues", []), + "levels": checkpoint.get("levels", []), + "file_conflicts": [], + "artifacts_dir": checkpoint.get("artifacts_dir", base), + "rationale": checkpoint.get("original_plan_summary", ""), + } + + app.note("Resuming build from checkpoint", tags=["build", "resume"]) + + result = await app.call( + f"{NODE_ID}.execute", + plan_result=plan_result, repo_path=repo_path, - artifacts_dir=artifacts_dir, config=config, git_config=git_config, + resume=True, ) - -@app.reasoner() -async def resume_build( - repo_path: str, - artifacts_dir: str = ".artifacts", - config: dict | None = None, - git_config: dict | None = None, - goal: str = "", - repo_url: str = "", -) -> dict: - """Resume DAG execution, then continue verifier/finalize/PR/CI build tail.""" - checkpoint = _load_execution_checkpoint(repo_path, artifacts_dir) - build_state = _load_build_state(repo_path, artifacts_dir) - overrides = dict(config or {}) - if repo_url: - overrides["repo_url"] = repo_url - cfg = _build_config_from_saved_state(build_state.get("config") or {}, overrides) - - effective_goal = goal or build_state.get("goal", "") - plan_result = build_state.get("plan_result") or _plan_result_from_checkpoint(checkpoint) - effective_git_config = git_config or build_state.get("git_config") or _git_config_from_checkpoint(checkpoint) - manifest_data = build_state.get("workspace_manifest") or checkpoint.get("workspace_manifest") - manifest = WorkspaceManifest(**manifest_data) if manifest_data else None - - app.note("Resuming full build from checkpoint", tags=["build", "resume"]) - exec_config = cfg.to_execution_config_dict() - dag_result = await _resume_execute_impl( - repo_path=repo_path, - artifacts_dir=artifacts_dir, - config=exec_config, - git_config=effective_git_config, - ) - build_state.update({ - "goal": effective_goal, - "repo_path": repo_path, - "repo_url": cfg.repo_url, - "artifacts_dir": artifacts_dir, - "config": cfg.model_dump(), - "plan_result": plan_result, - "dag_result": dag_result, - "git_config": effective_git_config, - "workspace_manifest": manifest.model_dump() if manifest else None, - }) - _save_build_state(repo_path, artifacts_dir, build_state) - - return await _continue_build_tail( - goal=effective_goal, - repo_path=repo_path, - artifacts_dir=artifacts_dir, - cfg=cfg, - resolved=cfg.resolved_models(), - plan_result=plan_result, - dag_result=dag_result, - git_config=effective_git_config, - manifest=manifest, - ever_completed=len(dag_result.get("completed_issues", []) or []), - ever_merged=len(dag_result.get("merged_branches", []) or []), - build_state=build_state, - ) + return result def main(): diff --git a/tests/test_planner_execute.py b/tests/test_planner_execute.py index 6301a802..dd04fe46 100644 --- a/tests/test_planner_execute.py +++ b/tests/test_planner_execute.py @@ -11,9 +11,7 @@ from __future__ import annotations -import json -from pathlib import Path -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch, call as mock_call import pytest @@ -49,12 +47,6 @@ def _make_plan_result(issues: list[dict] | None = None) -> dict: } -def _write_checkpoint(repo_path: Path, checkpoint: dict) -> None: - checkpoint_path = repo_path / ".artifacts" / "execution" / "checkpoint.json" - checkpoint_path.parent.mkdir(parents=True, exist_ok=True) - checkpoint_path.write_text(json.dumps(checkpoint), encoding="utf-8") - - def _make_dag_state(completed: list[str], failed: list[str]) -> DAGState: """Build a DAGState with given completed / failed issue names.""" return DAGState( @@ -78,171 +70,6 @@ def _make_dag_state(completed: list[str], failed: list[str]) -> DAGState: ) -def test_resume_incomplete_summary_empty_for_finished_dag(): - import swe_af.app as app_module - - result = { - "all_issues": [{"name": "one"}], - "completed_issues": [{"issue_name": "one"}], - "failed_issues": [], - "skipped_issues": [], - } - - assert app_module._resume_incomplete_summary(result) == "" - - -def test_resume_incomplete_summary_flags_failed_and_skipped_dag(): - import swe_af.app as app_module - - result = { - "all_issues": [{"name": "one"}, {"name": "two"}, {"name": "three"}], - "completed_issues": [{"issue_name": "one"}], - "failed_issues": [{"issue_name": "two"}], - "skipped_issues": ["three"], - } - - summary = app_module._resume_incomplete_summary(result) - - assert "Resume incomplete: 1/3 issues completed" in summary - assert "failed=['two']" in summary - assert "skipped=['three']" in summary - - -def test_build_config_saved_state_round_trips_normalized_repo_url(): - import swe_af.app as app_module - from swe_af.execution.schemas import BuildConfig - - saved_config = BuildConfig( - repo_url="https://github.com/example/repo.git", - ).model_dump() - - cfg = app_module._build_config_from_saved_state(saved_config) - - assert cfg.repo_url == "https://github.com/example/repo.git" - assert len(cfg.repos) == 1 - - -@pytest.mark.asyncio -async def test_resume_execute_only_resumes_dag(tmp_path: Path): - import swe_af.app as app_module - - repo_path = tmp_path / "repo" - repo_path.mkdir() - plan_result = _make_plan_result() - completed_result = { - "all_issues": plan_result["issues"], - "completed_issues": [{"issue_name": "implement-feature"}], - "failed_issues": [], - "skipped_issues": [], - "merged_branches": ["feature-branch"], - } - _write_checkpoint( - repo_path, - { - "all_issues": plan_result["issues"], - "levels": plan_result["levels"], - "artifacts_dir": str(repo_path / ".artifacts"), - "git_integration_branch": "integration", - }, - ) - - async def fake_call(target: str, **kwargs): - assert target == f"{app_module.NODE_ID}.execute" - assert kwargs["resume"] is True - return completed_result - - with patch.object(app_module.app, "call", new=AsyncMock(side_effect=fake_call)): - result = await app_module.resume_execute(repo_path=str(repo_path)) - - assert result == completed_result - - -@pytest.mark.asyncio -async def test_resume_build_continues_full_tail(tmp_path: Path): - import swe_af.app as app_module - - repo_path = tmp_path / "repo" - repo_path.mkdir() - plan_result = _make_plan_result() - dag_result = { - "all_issues": plan_result["issues"], - "completed_issues": [{"issue_name": "implement-feature"}], - "failed_issues": [], - "skipped_issues": [], - "merged_branches": ["feature-branch"], - "accumulated_debt": [], - } - git_config = { - "integration_branch": "integration", - "original_branch": "main", - "initial_commit_sha": "abc123", - "mode": "existing", - "remote_url": "https://github.com/example/repo.git", - "remote_default_branch": "main", - } - _write_checkpoint( - repo_path, - { - "all_issues": plan_result["issues"], - "levels": plan_result["levels"], - "artifacts_dir": str(repo_path / ".artifacts"), - "git_integration_branch": "integration", - }, - ) - app_module._save_build_state( - str(repo_path), - ".artifacts", - { - "goal": "ship feature", - "repo_path": str(repo_path), - "repo_url": "https://github.com/example/repo.git", - "artifacts_dir": ".artifacts", - "config": { - "runtime": "codex", - "models": {"default": "gpt-5.5"}, - "enable_github_pr": True, - "check_ci": False, - }, - "plan_result": plan_result, - "git_config": git_config, - }, - ) - - calls: list[str] = [] - - async def fake_call(target: str, **kwargs): - calls.append(target) - if target == f"{app_module.NODE_ID}.execute": - assert kwargs["resume"] is True - return dag_result - if target == f"{app_module.NODE_ID}.run_verifier": - return {"passed": True, "summary": "ok", "criteria_results": []} - if target == f"{app_module.NODE_ID}.run_repo_finalize": - return {"success": True, "summary": "clean"} - if target == f"{app_module.NODE_ID}.run_github_pr": - return { - "success": True, - "pr_url": "https://github.com/example/repo/pull/1", - "pr_number": 1, - } - raise AssertionError(f"unexpected call: {target}") - - with ( - patch.object(app_module.app, "call", new=AsyncMock(side_effect=fake_call)), - patch.object(app_module, "_existing_pr_for_branch", return_value={}), - patch.object(app_module, "_append_plan_docs_to_pr"), - ): - result = await app_module.resume_build(repo_path=str(repo_path)) - - assert result["success"] is True - assert calls == [ - f"{app_module.NODE_ID}.execute", - f"{app_module.NODE_ID}.run_verifier", - f"{app_module.NODE_ID}.run_repo_finalize", - f"{app_module.NODE_ID}.run_github_pr", - ] - - # --------------------------------------------------------------------------- # test_execute_single_issue # ---------------------------------------------------------------------------