Spaces:
Running
Running
File size: 3,227 Bytes
82372e5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | 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", [])),
)
|