Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| import os | |
| from pathlib import Path | |
| from src.models.provider import generate_with_hosted_model | |
| from src.schemas import ExternalAgentResult, SanitizedTask, SourceCandidate | |
| def run_codex_or_mock(task: SanitizedTask, base_dir: str | Path = "data") -> ExternalAgentResult: | |
| base = Path(base_dir) | |
| workspace = base / "runs" / task.run_id / "codex_workspace" | |
| workspace.mkdir(parents=True, exist_ok=True) | |
| prompt_path = workspace / "sanitized_task.txt" | |
| prompt_path.write_text(task.sanitized_query, encoding="utf-8") | |
| mock_mode = os.getenv("MOCK_MODE", "false").lower() == "true" | |
| if mock_mode: | |
| return _mock_result(task, base) | |
| hackathon_mode = os.getenv("HACKATHON_COMPLIANT_MODE", "false").lower() == "true" | |
| codex_runtime = os.getenv("CODEX_RUNTIME_MODE", "true").lower() == "true" | |
| if hackathon_mode or not codex_runtime: | |
| content, error = generate_with_hosted_model(task.sanitized_query) | |
| if content: | |
| return ExternalAgentResult( | |
| route="codex_big_agent", | |
| summary="Hosted under-32B model result:\n" + content, | |
| sources=[], | |
| artifacts=[], | |
| errors=[], | |
| ) | |
| return ExternalAgentResult( | |
| route="codex_big_agent", | |
| summary="No live hosted under-32B model result was available.", | |
| sources=[], | |
| artifacts=[], | |
| errors=[error or "Hosted model unavailable. No mock fallback was used because MOCK_MODE=false."], | |
| ) | |
| try: | |
| import openai_codex # type: ignore # noqa: F401 | |
| except Exception: | |
| return ExternalAgentResult( | |
| route="codex_big_agent", | |
| summary="Codex runtime is unavailable.", | |
| sources=[], | |
| artifacts=[], | |
| errors=["openai-codex is not installed or Codex auth is unavailable. No mock fallback was used because MOCK_MODE=false."], | |
| ) | |
| return ExternalAgentResult( | |
| route="codex_big_agent", | |
| summary="Codex SDK is installed, but the live adapter is not implemented in this prototype.", | |
| sources=[], | |
| artifacts=[], | |
| errors=["No mock fallback was used because MOCK_MODE=false."], | |
| ) | |
| def _mock_result(task: SanitizedTask, base: Path) -> ExternalAgentResult: | |
| fixture = json.loads((base / "fixtures" / "codex_mock_result.json").read_text(encoding="utf-8")) | |
| workspace = base / "runs" / task.run_id / "codex_workspace" | |
| artifact = workspace / "release_monitor_plan.md" | |
| artifact.write_text( | |
| "# Release Monitor Plan\n\n" | |
| f"Sanitized task: {task.sanitized_query}\n\n" | |
| "1. Store tool metadata in JSON.\n" | |
| "2. Poll release feeds or public APIs.\n" | |
| "3. Check docs URLs for freshness signals.\n" | |
| "4. Write a small report with changes and unknowns.\n", | |
| encoding="utf-8", | |
| ) | |
| sources = [SourceCandidate.from_dict(item) for item in fixture.get("sources", [])] | |
| return ExternalAgentResult( | |
| route="codex_big_agent", | |
| summary=fixture["summary"], | |
| sources=sources, | |
| artifacts=[str(artifact)], | |
| errors=list(fixture.get("errors", [])), | |
| ) | |