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/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..53a5bafe 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,48 @@ 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) + + +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, @@ -110,6 +178,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 +205,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 +226,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: @@ -162,28 +242,27 @@ 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 - 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 +312,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 +323,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 +383,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..7a00878e 100644 --- a/tests/test_codex_harness_patch.py +++ b/tests/test_codex_harness_patch.py @@ -1,8 +1,13 @@ from __future__ import annotations +import shutil + 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, apply_codex_harness_patch, ) @@ -50,6 +55,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 +89,83 @@ 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( + [ + { + "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 +174,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