mirror of
https://github.com/openai/skills.git
synced 2026-09-18 15:06:29 +03:00
Codex readiness integration test: planning signal + verification rules, safer runner defaults (#52)
This commit is contained in:
Luke
@@ -9,10 +9,6 @@ metadata:
|
||||
|
||||
This skill runs a multi-stage integration test to validate agentic execution quality. It always runs in execute mode (no read-only mode).
|
||||
|
||||
## Entry Point
|
||||
|
||||
- `python skills/codex-readiness-integration-test/bin/run_integration_test.py`
|
||||
|
||||
## Outputs
|
||||
|
||||
Each run writes to `.codex-readiness-integration-test/<timestamp>/` and updates `.codex-readiness-integration-test/latest.json`.
|
||||
@@ -22,7 +18,7 @@ New outputs per run:
|
||||
- `llm_results.json` (automatic LLM evaluation)
|
||||
- `summary.txt` (human-readable summary)
|
||||
|
||||
## Pre-conditions
|
||||
## Pre-conditions (Required)
|
||||
|
||||
- Authenticate with the Codex CLI using the repo-local HOME before running the test.
|
||||
Run these in your own terminal (not via the integration test):
|
||||
@@ -35,18 +31,20 @@ New outputs per run:
|
||||
0) Ask the user how to source the task.
|
||||
- Offer two explicit options: (a) user provides a custom task/prompt, or (b) auto-generate a task.
|
||||
- Do not run the entry point until the user chooses one option.
|
||||
1) Generate or load `prompt.json`.
|
||||
1) Generate or load `{out_dir}/prompt.pending.json`.
|
||||
- Use the integration test's expected prompt path, not `prompt.json` at the repo root.
|
||||
- With the default out dir, this path is `.codex-readiness-integration-test/prompt.pending.json`.
|
||||
- If `--seed-task` is provided, it is used as the starting task.
|
||||
- If not provided, generate a task with `skills/codex-readiness-integration-test/references/generate_prompt.md` and save the JSON.
|
||||
- If not provided, generate a task with `skills/codex-readiness-integration-test/references/generate_prompt.md` and save the JSON to `{out_dir}/prompt.pending.json`.
|
||||
- The user must approve the prompt before execution (no auto-approve mode). Make sure to output a summary of the prompt when asking the user to approve.
|
||||
2) Execute the agentic loop via Codex CLI (uses `AGENTS.md` and `change_prompt`).
|
||||
3) Run build/test commands from the prompt plan via `skills/codex-readiness-integration-test/bin/run_plan.py`.
|
||||
3) Run build/test commands from the prompt plan via `skills/codex-readiness-integration-test/scripts/run_plan.py`.
|
||||
4) Collect evidence (`evidence.json`), deterministic checks, and run automatic LLM evals via Codex CLI.
|
||||
5) Score and write the report + summary output.
|
||||
|
||||
## Configuration
|
||||
|
||||
Optional fields in `prompt.json`:
|
||||
Optional fields in `{out_dir}/prompt.pending.json`:
|
||||
- `agentic_loop`: configure Codex CLI invocation for the agentic loop.
|
||||
- `llm_eval`: configure Codex CLI invocation for automatic evals.
|
||||
|
||||
@@ -55,7 +53,6 @@ If these fields are omitted, defaults are used.
|
||||
## Requirements
|
||||
|
||||
- The LLM evaluator must fail if evidence mentions the phrase `Context compaction enabled`.
|
||||
- The LLM evaluator must check that `AGENTS.md` was referenced.
|
||||
- Use qualitative context-usage evaluation (no strict thresholds).
|
||||
|
||||
|
||||
@@ -76,4 +73,5 @@ If these fields are omitted, defaults are used.
|
||||
|
||||
- The prompts in `skills/codex-readiness-integration-test/references/` expect strict JSON.
|
||||
- Use `skills/codex-readiness-integration-test/references/json_fix.md` to repair invalid JSON output.
|
||||
- This skill calls the `codex` CLI. Ensure it is installed and available on PATH, or override the command in `prompt.json`.
|
||||
- This skill calls the `codex` CLI. Ensure it is installed and available on PATH, or override the command in `{out_dir}/prompt.pending.json`.
|
||||
- If the agentic loop detects sandbox-blocked tool access, it now writes `requires_escalation: true` to `{run_dir}/agentic_summary.json` and exits with code `3`. Re-run the integration test with escalated permissions in that case.
|
||||
|
||||
@@ -7,9 +7,6 @@ You are evaluating whether the agentic loop executed successfully. You will be g
|
||||
|
||||
Hard fails:
|
||||
- If any evidence or logs mention the phrase "Context compaction enabled", status MUST be FAIL.
|
||||
- If AGENTS.md exists (evidence.agents_md.exists) but there is no evidence it was referenced (for example agentic_summary.cmd or logs mention the AGENTS.md path), status MUST be FAIL.
|
||||
|
||||
If AGENTS.md does not exist, record that fact and do not fail solely for missing reference.
|
||||
|
||||
Prefer PASS only if agentic_summary.status is PASS and exit_code is 0.
|
||||
|
||||
|
||||
@@ -19,8 +19,8 @@
|
||||
},
|
||||
{
|
||||
"id": "exec_plan_before_code_changes",
|
||||
"title": "P0: ExecPlan created before code changes",
|
||||
"description": "ExecPlan file update appears before any non-doc, non-.codex file changes.",
|
||||
"title": "P0: Planning signal before code changes",
|
||||
"description": "A planning signal (for example: update_plan or \"Plan:\") appears in agentic.log before any non-doc, non-.codex file changes.",
|
||||
"priority": 0,
|
||||
"weight": null,
|
||||
"type": "DETERMINISTIC",
|
||||
@@ -34,6 +34,23 @@
|
||||
},
|
||||
"enabled_by_default": true
|
||||
},
|
||||
{
|
||||
"id": "verification_after_code_changes",
|
||||
"title": "P0: Verification after code changes",
|
||||
"description": "A build, test, or lint verification command appears in agentic.log after the first non-doc, non-.codex file change.",
|
||||
"priority": 0,
|
||||
"weight": null,
|
||||
"type": "DETERMINISTIC",
|
||||
"scope": "run_dir",
|
||||
"execute_required": true,
|
||||
"evaluator_prompt_id": null,
|
||||
"deterministic_rule_id": "verification_after_code_changes",
|
||||
"deterministic_rule_params": {
|
||||
"prompt_path": "prompt.json",
|
||||
"agentic_log_path": "logs/agentic.log"
|
||||
},
|
||||
"enabled_by_default": true
|
||||
},
|
||||
{
|
||||
"id": "repo_root_only_changes",
|
||||
"title": "P1: Repo-root-only changes",
|
||||
@@ -109,4 +126,4 @@
|
||||
"enabled_by_default": true
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,7 +15,7 @@ Return strict JSON with this schema:
|
||||
"scoring_focus": ["correctness", "context_usage", "builds_tests_pass", "maintainability", "risk"],
|
||||
"agentic_loop": {
|
||||
"cmd": "codex",
|
||||
"args": ["exec", "--dangerously-bypass-approvals-and-sandbox", "-C", "{repo_root}", "{change_prompt}"],
|
||||
"args": ["exec", "--full-auto", "-C", "{repo_root}", "{change_prompt}"],
|
||||
"timeout_seconds": 1800
|
||||
},
|
||||
"llm_eval": {
|
||||
|
||||
@@ -56,18 +56,15 @@ def run_cmd_allow_failure(cmd: list[str]) -> str:
|
||||
|
||||
|
||||
def should_include_untracked(path: Path) -> bool:
|
||||
if path.name == '.DS_Store':
|
||||
if path.name == ".DS_Store":
|
||||
return False
|
||||
for part in path.parts:
|
||||
if part.startswith('.codex'):
|
||||
return False
|
||||
return True
|
||||
return all(not part.startswith(".codex") for part in path.parts)
|
||||
|
||||
|
||||
def build_untracked_diff() -> str:
|
||||
raw = run_cmd(['git', 'ls-files', '--others', '--exclude-standard'])
|
||||
if raw.startswith('<error>'):
|
||||
return ''
|
||||
raw = run_cmd(["git", "ls-files", "--others", "--exclude-standard"])
|
||||
if raw.startswith("<error>"):
|
||||
return ""
|
||||
diffs = []
|
||||
for line in raw.splitlines():
|
||||
line = line.strip()
|
||||
@@ -76,10 +73,10 @@ def build_untracked_diff() -> str:
|
||||
path = Path(line)
|
||||
if not should_include_untracked(path):
|
||||
continue
|
||||
diff = run_cmd_allow_failure(['git', 'diff', '--no-index', '/dev/null', line])
|
||||
if diff and not diff.startswith('<error>'):
|
||||
diff = run_cmd_allow_failure(["git", "diff", "--no-index", "/dev/null", line])
|
||||
if diff and not diff.startswith("<error>"):
|
||||
diffs.append(diff)
|
||||
return '\n'.join(diffs)
|
||||
return "\n".join(diffs)
|
||||
|
||||
|
||||
def load_json_if_exists(path: Path) -> dict | None:
|
||||
|
||||
+259
-142
@@ -2,11 +2,13 @@
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
VALID_STATUSES = {"PASS", "WARN", "FAIL", "NOT_RUN"}
|
||||
ANSI_ESCAPE_PATTERN = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
|
||||
ERROR_MARKERS = [
|
||||
"error:",
|
||||
"failed",
|
||||
@@ -14,6 +16,56 @@ ERROR_MARKERS = [
|
||||
"traceback",
|
||||
"segmentation fault",
|
||||
]
|
||||
PLANNING_SIGNAL_PATTERNS = [
|
||||
re.compile(r"\bupdate_plan\b", re.IGNORECASE),
|
||||
re.compile(r"^\s*plan\s*[:\-]", re.IGNORECASE),
|
||||
re.compile(r"^\s*plan\s+update\b", re.IGNORECASE),
|
||||
re.compile(r"^\s*\*\*planning\b", re.IGNORECASE),
|
||||
re.compile(r"^\s*steps?\s*[:\-]", re.IGNORECASE),
|
||||
re.compile(r"^\s*approach\s*[:\-]", re.IGNORECASE),
|
||||
re.compile(r"\bhere(?:'s| is)\s+(?:the\s+)?plan\b", re.IGNORECASE),
|
||||
]
|
||||
COMMAND_LINE_PATTERNS = [
|
||||
re.compile(r"^\s*\$\s+(.+)$"),
|
||||
re.compile(r"^\s*!\s*(.+)$"),
|
||||
re.compile(r"^\s*running(?: command)?\s*:\s+(.+)$", re.IGNORECASE),
|
||||
re.compile(r"^\s*cmd\s*:\s+(.+)$", re.IGNORECASE),
|
||||
]
|
||||
# Match shell "-lc '<cmd>'" forms even when prefixed by a path like /bin/zsh.
|
||||
SHELL_LC_PATTERN = re.compile(r"(?:^|\s)-lc\s+(?P<quote>['\"])(?P<cmd>.+?)(?P=quote)")
|
||||
VERIFICATION_KEYWORDS = [
|
||||
" test",
|
||||
"pytest",
|
||||
"npm test",
|
||||
"pnpm test",
|
||||
"yarn test",
|
||||
"node --test",
|
||||
"go test",
|
||||
"cargo test",
|
||||
"mvn test",
|
||||
"gradle test",
|
||||
"./gradlew test",
|
||||
"lint",
|
||||
"eslint",
|
||||
"ruff",
|
||||
"flake8",
|
||||
"black --check",
|
||||
"prettier --check",
|
||||
"typecheck",
|
||||
"tsc",
|
||||
" build",
|
||||
"compile",
|
||||
"mvn package",
|
||||
"gradle build",
|
||||
"./gradlew build",
|
||||
"go build",
|
||||
"cargo build",
|
||||
"make build",
|
||||
"make test",
|
||||
"make lint",
|
||||
"make verify",
|
||||
"verify",
|
||||
]
|
||||
|
||||
|
||||
def load_json(path: Path) -> dict:
|
||||
@@ -196,78 +248,138 @@ def resolve_repo_root() -> Path | None:
|
||||
return Path(repo_root_raw).resolve()
|
||||
|
||||
|
||||
def exec_plan_path_candidates(
|
||||
exec_plan_path: str, run_dir: Path, repo_root: Path | None
|
||||
) -> list[str]:
|
||||
candidates: set[str] = set()
|
||||
exec_plan_path = exec_plan_path.strip()
|
||||
if exec_plan_path:
|
||||
candidates.add(exec_plan_path)
|
||||
candidates.add(exec_plan_path.lstrip("./"))
|
||||
|
||||
plan_path = Path(exec_plan_path)
|
||||
if not plan_path.is_absolute():
|
||||
if repo_root:
|
||||
candidates.add(str((repo_root / plan_path).resolve()))
|
||||
candidates.add(str((run_dir / plan_path).resolve()))
|
||||
candidates.add(str((run_dir.parent / plan_path).resolve()))
|
||||
|
||||
return sorted(candidates)
|
||||
|
||||
|
||||
def find_exec_plan_command_index(lines: list[str], candidates: list[str]) -> int | None:
|
||||
patterns = []
|
||||
for path in candidates:
|
||||
if not path:
|
||||
continue
|
||||
escaped = re.escape(path)
|
||||
patterns.append(re.compile(rf">>?\s*{escaped}(?:$|\\s|\"|')"))
|
||||
patterns.append(re.compile(rf"\\btee\\b(?:\\s+-a)?\\s+{escaped}(?:$|\\s|\"|')"))
|
||||
|
||||
for idx, line in enumerate(lines):
|
||||
for pattern in patterns:
|
||||
if pattern.search(line):
|
||||
return idx
|
||||
return None
|
||||
|
||||
|
||||
def is_doc_path(path: str) -> bool:
|
||||
return path.lower().endswith(".md")
|
||||
|
||||
|
||||
def is_exec_plan_candidate(path: str) -> bool:
|
||||
if not is_doc_path(path):
|
||||
return False
|
||||
name = Path(path).name.lower()
|
||||
if name == "plans.md":
|
||||
return False
|
||||
return "plan" in name
|
||||
def strip_ansi(text: str) -> str:
|
||||
return ANSI_ESCAPE_PATTERN.sub("", text)
|
||||
|
||||
|
||||
def infer_exec_plan_paths(log_text: str) -> list[str]:
|
||||
candidates: set[str] = set()
|
||||
events = parse_agentic_file_update_events(log_text)
|
||||
def command_binary(cmd: str) -> str:
|
||||
try:
|
||||
parts = shlex.split(cmd)
|
||||
except ValueError:
|
||||
parts = cmd.split()
|
||||
if not parts:
|
||||
return ""
|
||||
return Path(parts[0]).name.lower()
|
||||
|
||||
|
||||
def is_codex_invocation(cmd: str) -> bool:
|
||||
binary = command_binary(cmd)
|
||||
return binary in {"codex", "codex.exe"}
|
||||
|
||||
|
||||
def extract_command_events(log_text: str) -> list[dict[str, Any]]:
|
||||
events: list[dict[str, Any]] = []
|
||||
for idx, raw_line in enumerate(log_text.splitlines()):
|
||||
clean_line = strip_ansi(raw_line).strip()
|
||||
if not clean_line:
|
||||
continue
|
||||
|
||||
# First handle common "command-like" prefixes such as "$ npm test".
|
||||
for pattern in COMMAND_LINE_PATTERNS:
|
||||
match = pattern.match(clean_line)
|
||||
if not match:
|
||||
continue
|
||||
cmd = match.group(1).strip()
|
||||
if not cmd:
|
||||
continue
|
||||
if is_codex_invocation(cmd):
|
||||
# Ignore runner-level codex invocations; they are not verification steps.
|
||||
continue
|
||||
events.append(
|
||||
{
|
||||
"line_index": idx,
|
||||
"cmd": cmd,
|
||||
"raw_line": clean_line,
|
||||
}
|
||||
)
|
||||
break
|
||||
|
||||
else:
|
||||
# Fall back to extracting the inner command from shell "-lc" invocations
|
||||
# such as: /bin/zsh -lc 'npm test' ... succeeded in 64ms
|
||||
lc_match = SHELL_LC_PATTERN.search(clean_line)
|
||||
if not lc_match:
|
||||
continue
|
||||
cmd = lc_match.group("cmd").strip()
|
||||
if not cmd:
|
||||
continue
|
||||
if is_codex_invocation(cmd):
|
||||
continue
|
||||
events.append(
|
||||
{
|
||||
"line_index": idx,
|
||||
"cmd": cmd,
|
||||
"raw_line": clean_line,
|
||||
}
|
||||
)
|
||||
return events
|
||||
|
||||
|
||||
def prompt_command_candidates(prompt: dict[str, Any]) -> list[str]:
|
||||
candidates: list[str] = []
|
||||
plan = prompt.get("build_test_plan")
|
||||
if not isinstance(plan, list):
|
||||
return candidates
|
||||
for entry in plan:
|
||||
if isinstance(entry, dict):
|
||||
cmd = entry.get("cmd")
|
||||
elif isinstance(entry, str):
|
||||
cmd = entry
|
||||
else:
|
||||
cmd = None
|
||||
if isinstance(cmd, str) and cmd.strip():
|
||||
candidates.append(cmd.strip().lower())
|
||||
return candidates
|
||||
|
||||
|
||||
def is_verification_command(cmd: str, prompt_cmds: list[str]) -> bool:
|
||||
lower = cmd.lower()
|
||||
if any(keyword in lower for keyword in VERIFICATION_KEYWORDS):
|
||||
return True
|
||||
return any(prompt_cmd and prompt_cmd in lower for prompt_cmd in prompt_cmds)
|
||||
|
||||
|
||||
def first_code_change_event(events: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||||
code_events: list[dict[str, Any]] = []
|
||||
for event in events:
|
||||
path = event.get("path")
|
||||
if path and is_exec_plan_candidate(path):
|
||||
candidates.add(path)
|
||||
|
||||
for match in re.finditer(r"([\w./-]*plan[\w./-]*\.md)", log_text, re.IGNORECASE):
|
||||
path = match.group(1)
|
||||
if is_exec_plan_candidate(path):
|
||||
candidates.add(path)
|
||||
|
||||
return sorted(candidates)
|
||||
|
||||
|
||||
def select_existing_plan_file(candidates: list[str]) -> Path | None:
|
||||
for candidate in candidates:
|
||||
path = str(event.get("path", ""))
|
||||
try:
|
||||
path = Path(candidate)
|
||||
line_index = int(event.get("line_index"))
|
||||
except Exception:
|
||||
continue
|
||||
if path.exists():
|
||||
return path
|
||||
if not path:
|
||||
continue
|
||||
if path.startswith(".codex/"):
|
||||
continue
|
||||
if is_doc_path(path):
|
||||
continue
|
||||
code_events.append({"path": path, "line_index": line_index})
|
||||
if not code_events:
|
||||
return None
|
||||
return min(code_events, key=lambda item: item["line_index"])
|
||||
|
||||
|
||||
def file_update_line_indexes(events: list[dict[str, Any]]) -> set[int]:
|
||||
indexes: set[int] = set()
|
||||
for event in events:
|
||||
try:
|
||||
indexes.add(int(event.get("line_index")))
|
||||
except Exception:
|
||||
continue
|
||||
return indexes
|
||||
|
||||
|
||||
def find_first_planning_signal_index(lines: list[str], skip_indexes: set[int]) -> int | None:
|
||||
for idx, line in enumerate(lines):
|
||||
if idx in skip_indexes:
|
||||
continue
|
||||
for pattern in PLANNING_SIGNAL_PATTERNS:
|
||||
if pattern.search(line):
|
||||
return idx
|
||||
return None
|
||||
|
||||
|
||||
@@ -276,7 +388,7 @@ def check_exec_plan_before_code_changes(run_dir: Path, params: dict[str, Any]) -
|
||||
if not prompt_path.exists():
|
||||
return result("FAIL", "prompt.json is missing.")
|
||||
try:
|
||||
prompt = load_json(prompt_path)
|
||||
_prompt = load_json(prompt_path)
|
||||
except Exception:
|
||||
return result("FAIL", "prompt.json could not be parsed.")
|
||||
|
||||
@@ -290,112 +402,116 @@ def check_exec_plan_before_code_changes(run_dir: Path, params: dict[str, Any]) -
|
||||
if not events:
|
||||
return result("WARN", "No file update entries found in agentic.log.")
|
||||
|
||||
repo_root = resolve_repo_root()
|
||||
exec_plan_path = prompt.get("exec_plan_path")
|
||||
candidates: list[str] = []
|
||||
if isinstance(exec_plan_path, str) and exec_plan_path.strip():
|
||||
exec_plan_path = exec_plan_path.strip()
|
||||
candidates.extend(exec_plan_path_candidates(exec_plan_path, run_dir, repo_root))
|
||||
else:
|
||||
inferred = infer_exec_plan_paths(log_text)
|
||||
for inferred_path in inferred:
|
||||
candidates.extend(exec_plan_path_candidates(inferred_path, run_dir, repo_root))
|
||||
|
||||
candidates = [path for path in candidates if path]
|
||||
if not candidates:
|
||||
return result(
|
||||
"FAIL",
|
||||
"ExecPlan path missing from prompt.json and could not be inferred from agentic.log.",
|
||||
[{"path": str(log_path), "quote": "exec plan not detected"}],
|
||||
["Create the ExecPlan and ensure it is referenced in logs before code changes."],
|
||||
)
|
||||
|
||||
plan_line = None
|
||||
code_line = None
|
||||
for event in events:
|
||||
path = event["path"]
|
||||
line_index = event["line_index"]
|
||||
if any(path == cand or path.endswith(cand) for cand in candidates):
|
||||
if plan_line is None or line_index < plan_line:
|
||||
plan_line = line_index
|
||||
continue
|
||||
if path.startswith(".codex/"):
|
||||
continue
|
||||
if is_doc_path(path):
|
||||
continue
|
||||
if code_line is None:
|
||||
code_line = line_index
|
||||
|
||||
command_line = find_exec_plan_command_index(lines, candidates)
|
||||
if command_line is not None and (plan_line is None or command_line < plan_line):
|
||||
plan_line = command_line
|
||||
|
||||
if plan_line is None:
|
||||
return result(
|
||||
"FAIL",
|
||||
"ExecPlan file was not created before code changes.",
|
||||
[{"path": str(log_path), "quote": "missing exec plan update"}],
|
||||
["Create the ExecPlan before making code changes."],
|
||||
)
|
||||
|
||||
if code_line is None:
|
||||
code_event = first_code_change_event(events)
|
||||
if code_event is None:
|
||||
return result(
|
||||
"WARN",
|
||||
"No non-doc, non-.codex file changes found; ordering not evaluated.",
|
||||
[{"path": str(log_path), "quote": "no code updates"}],
|
||||
)
|
||||
code_line = int(code_event["line_index"])
|
||||
|
||||
plan_file = select_existing_plan_file(candidates)
|
||||
if plan_file is None or not plan_file.exists():
|
||||
skip_indexes = file_update_line_indexes(events)
|
||||
plan_line = find_first_planning_signal_index(lines, skip_indexes)
|
||||
if plan_line is None:
|
||||
return result(
|
||||
"FAIL",
|
||||
"ExecPlan file is missing on disk.",
|
||||
[{"path": "candidate_paths", "quote": ", ".join(candidates)}],
|
||||
["Create the ExecPlan file before making code changes."],
|
||||
)
|
||||
|
||||
try:
|
||||
plan_text = plan_file.read_text(encoding="utf-8", errors="ignore")
|
||||
except Exception:
|
||||
plan_text = ""
|
||||
|
||||
required_headings = [
|
||||
"# ",
|
||||
"## Purpose / Big Picture",
|
||||
"## Progress",
|
||||
"## Decision Log",
|
||||
"## Outcomes & Retrospective",
|
||||
]
|
||||
missing = [heading for heading in required_headings if heading not in plan_text]
|
||||
if missing:
|
||||
return result(
|
||||
"FAIL",
|
||||
"ExecPlan file is missing required sections.",
|
||||
[{"path": str(plan_file), "quote": ", ".join(missing)}],
|
||||
["Ensure the ExecPlan follows PLANS.md headings."],
|
||||
"No planning signal detected in agentic.log before code changes.",
|
||||
[
|
||||
{"path": str(log_path), "quote": "planning signal not detected"},
|
||||
{
|
||||
"path": str(log_path),
|
||||
"quote": lines[code_line] if code_line < len(lines) else "",
|
||||
},
|
||||
],
|
||||
["Emit a short plan (for example: 'Plan:' or use update_plan) before code edits."],
|
||||
)
|
||||
|
||||
if plan_line <= code_line:
|
||||
return result(
|
||||
"PASS",
|
||||
"ExecPlan update appears before code changes.",
|
||||
"Planning signal appears before code changes.",
|
||||
[
|
||||
{
|
||||
"path": str(log_path),
|
||||
"quote": lines[plan_line] if plan_line < len(lines) else "",
|
||||
},
|
||||
{"path": str(plan_file), "quote": "exec plan used"},
|
||||
{
|
||||
"path": str(log_path),
|
||||
"quote": lines[code_line] if code_line < len(lines) else "",
|
||||
},
|
||||
],
|
||||
)
|
||||
|
||||
return result(
|
||||
"FAIL",
|
||||
"ExecPlan update appears after code changes.",
|
||||
"Planning signal appears after code changes.",
|
||||
[
|
||||
{"path": str(log_path), "quote": lines[code_line] if code_line < len(lines) else ""},
|
||||
{"path": str(log_path), "quote": lines[plan_line] if plan_line < len(lines) else ""},
|
||||
],
|
||||
["Create the ExecPlan before making code changes."],
|
||||
["Emit a short plan (for example: 'Plan:' or use update_plan) before code edits."],
|
||||
)
|
||||
|
||||
|
||||
def check_verification_after_code_changes(run_dir: Path, params: dict[str, Any]) -> dict:
|
||||
prompt_path = run_dir / params.get("prompt_path", "prompt.json")
|
||||
if not prompt_path.exists():
|
||||
return result("FAIL", "prompt.json is missing.")
|
||||
try:
|
||||
prompt = load_json(prompt_path)
|
||||
except Exception:
|
||||
return result("FAIL", "prompt.json could not be parsed.")
|
||||
|
||||
log_path = run_dir / params.get("agentic_log_path", "logs/agentic.log")
|
||||
if not log_path.exists():
|
||||
return result("FAIL", "agentic.log is missing.")
|
||||
|
||||
log_text = log_path.read_text(encoding="utf-8", errors="ignore")
|
||||
lines = [strip_ansi(line) for line in log_text.splitlines()]
|
||||
file_events = parse_agentic_file_update_events(log_text)
|
||||
if not file_events:
|
||||
return result("WARN", "No file update entries found in agentic.log.")
|
||||
|
||||
code_event = first_code_change_event(file_events)
|
||||
if code_event is None:
|
||||
return result(
|
||||
"WARN",
|
||||
"No non-doc, non-.codex file changes found; verification ordering not evaluated.",
|
||||
[{"path": str(log_path), "quote": "no code updates"}],
|
||||
)
|
||||
code_line = int(code_event["line_index"])
|
||||
|
||||
command_events = extract_command_events(log_text)
|
||||
prompt_cmds = prompt_command_candidates(prompt)
|
||||
verification_events = [
|
||||
event
|
||||
for event in command_events
|
||||
if int(event["line_index"]) > code_line
|
||||
and is_verification_command(str(event.get("cmd", "")), prompt_cmds)
|
||||
]
|
||||
|
||||
code_quote = lines[code_line] if code_line < len(lines) else str(code_event.get("path", ""))
|
||||
if not verification_events:
|
||||
return result(
|
||||
"FAIL",
|
||||
"No build/test/lint verification command detected after code changes in agentic.log.",
|
||||
[{"path": str(log_path), "quote": code_quote}],
|
||||
[
|
||||
"Run at least one build, test, or lint command after code changes within the agentic loop."
|
||||
],
|
||||
)
|
||||
|
||||
evidence: list[dict[str, str]] = [{"path": str(log_path), "quote": code_quote}]
|
||||
for event in verification_events[:2]:
|
||||
idx = int(event["line_index"])
|
||||
quote = lines[idx] if idx < len(lines) else str(event.get("raw_line", event.get("cmd", "")))
|
||||
evidence.append({"path": str(log_path), "quote": quote})
|
||||
|
||||
return result(
|
||||
"PASS",
|
||||
"Verification command(s) appear after code changes in agentic.log.",
|
||||
evidence,
|
||||
)
|
||||
|
||||
|
||||
@@ -469,6 +585,7 @@ RULES = {
|
||||
"execution_logs_no_errors": check_execution_logs_no_errors,
|
||||
"agentic_run_success": check_agentic_run_success,
|
||||
"exec_plan_before_code_changes": check_exec_plan_before_code_changes,
|
||||
"verification_after_code_changes": check_verification_after_code_changes,
|
||||
"repo_root_only_changes": check_repo_root_only_changes,
|
||||
}
|
||||
|
||||
|
||||
+100
-18
@@ -54,6 +54,13 @@ MAX_FOLLOWUP_ROUNDS = 5
|
||||
TAIL_LINE_LIMIT = 400
|
||||
QUESTION_TERMINATE_GRACE_SECONDS = 2.0
|
||||
|
||||
SANDBOX_BLOCK_SUBSTRINGS = [
|
||||
"sandbox-blocked",
|
||||
"shell tool is sandbox-blocked",
|
||||
"sandbox_apply: operation not permitted",
|
||||
"operation not permitted",
|
||||
]
|
||||
|
||||
|
||||
def now_iso() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
@@ -100,18 +107,42 @@ def substitute_args(args: list[str], mapping: dict[str, str]) -> list[str]:
|
||||
return resolved
|
||||
|
||||
|
||||
def sanitize_agentic_args(args: list[str]) -> list[str]:
|
||||
"""Remove unsafe or runner-managed flags from prompt-supplied args."""
|
||||
sanitized: list[str] = []
|
||||
skip_next = False
|
||||
# Flags that should not be controlled by the prompt in this runner.
|
||||
deny_flags_with_value = {"--sandbox", "--ask-for-approval", "-C", "--cd"}
|
||||
deny_flags = {
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--full-auto",
|
||||
"exec",
|
||||
"resume",
|
||||
"{change_prompt}",
|
||||
}
|
||||
for arg in args:
|
||||
if skip_next:
|
||||
skip_next = False
|
||||
continue
|
||||
if arg in deny_flags_with_value:
|
||||
skip_next = True
|
||||
continue
|
||||
if arg in deny_flags:
|
||||
continue
|
||||
sanitized.append(arg)
|
||||
return sanitized
|
||||
|
||||
|
||||
def build_command(
|
||||
prompt: dict[str, Any], agents_path: Path, prompt_path: Path, repo_root: Path
|
||||
) -> tuple[list[str], int]:
|
||||
agentic_config = prompt.get("agentic_loop")
|
||||
config: dict[str, Any] = agentic_config if isinstance(agentic_config, dict) else {}
|
||||
cmd = config.get("cmd") or "codex"
|
||||
raw_args = config.get("args") or [
|
||||
"exec",
|
||||
"-C",
|
||||
"{repo_root}",
|
||||
"{change_prompt}",
|
||||
]
|
||||
prompt_args = sanitize_agentic_args(normalize_args(config.get("args")))
|
||||
# Hardcode a safe, broadly supported permission model at the runner level,
|
||||
# while allowing other prompt-supplied flags (e.g., model selection).
|
||||
raw_args = ["exec", "--full-auto"] + prompt_args + ["-C", "{repo_root}", "{change_prompt}"]
|
||||
args = normalize_args(raw_args)
|
||||
change_prompt = str(prompt.get("change_prompt") or "").strip()
|
||||
plan_instruction = str(prompt.get("plan_instruction") or "").strip()
|
||||
@@ -158,7 +189,8 @@ def build_resume_command(
|
||||
agentic_config = prompt.get("agentic_loop")
|
||||
config: dict[str, Any] = agentic_config if isinstance(agentic_config, dict) else {}
|
||||
cmd = config.get("cmd") or "codex"
|
||||
raw_args = config.get("args") or []
|
||||
prompt_args = sanitize_agentic_args(normalize_args(config.get("args")))
|
||||
raw_args = ["exec", "--full-auto"] + prompt_args
|
||||
args = normalize_args(raw_args)
|
||||
mapping = {
|
||||
"{agents_path}": str(agents_path),
|
||||
@@ -321,6 +353,12 @@ def is_question_line(line: str) -> bool:
|
||||
lower = line.strip().lower()
|
||||
if not lower:
|
||||
return False
|
||||
# Codex often ends with friendly follow-up headings like 'what changed:'
|
||||
# or 'what i verified:'. Treat these as non-blocking in non-interactive runs.
|
||||
if lower.startswith("what changed"):
|
||||
return False
|
||||
if lower.startswith("what ") and lower.endswith(":"):
|
||||
return False
|
||||
if lower.endswith("?"):
|
||||
return True
|
||||
if lower.startswith(QUESTION_PREFIXES):
|
||||
@@ -351,6 +389,13 @@ def extract_clarifying_question(log_text: str) -> str | None:
|
||||
def prompt_for_answer(question: str) -> str:
|
||||
print("\nCodex asked:")
|
||||
print(question)
|
||||
if not sys.stdin.isatty():
|
||||
auto = os.environ.get(
|
||||
"CODEX_INTEGRATION_AUTOANSWER",
|
||||
"Proceed with best effort using the repository context. Do not ask follow-up questions.",
|
||||
).strip()
|
||||
print(f"Auto-answering (non-interactive): {auto}")
|
||||
return auto
|
||||
while True:
|
||||
answer = input("Answer: ").strip()
|
||||
if answer:
|
||||
@@ -358,6 +403,18 @@ def prompt_for_answer(question: str) -> str:
|
||||
print("Please provide an answer to continue.")
|
||||
|
||||
|
||||
def detect_sandbox_block(log_text: str) -> str | None:
|
||||
for raw_line in log_text.splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line:
|
||||
continue
|
||||
lower = line.lower()
|
||||
for marker in SANDBOX_BLOCK_SUBSTRINGS:
|
||||
if marker in lower:
|
||||
return line
|
||||
return None
|
||||
|
||||
|
||||
def append_tail_lines(tail_lines: list[str], line: str) -> None:
|
||||
tail_lines.append(line)
|
||||
if len(tail_lines) > TAIL_LINE_LIMIT:
|
||||
@@ -696,6 +753,8 @@ def main() -> int:
|
||||
resume_prompt: str | None = None
|
||||
append_log = False
|
||||
summary: dict | None = None
|
||||
auto_answer_count = 0
|
||||
last_auto_answer_text: str | None = None
|
||||
|
||||
for attempt in range(1, MAX_FOLLOWUP_ROUNDS + 1):
|
||||
if resume_prompt:
|
||||
@@ -714,7 +773,7 @@ def main() -> int:
|
||||
cmd, timeout = build_command(prompt, agents_path, prompt_path, repo_root)
|
||||
attempt_label = f"agentic-attempt-{attempt}"
|
||||
|
||||
if resume_prompt:
|
||||
if resume_prompt and sys.stdin.isatty():
|
||||
summary = run_safe_interactive(
|
||||
cmd,
|
||||
repo_root,
|
||||
@@ -747,6 +806,20 @@ def main() -> int:
|
||||
question = summary.get("question_detected") or extract_clarifying_question(attempt_log)
|
||||
if question:
|
||||
questions.append(question)
|
||||
if not sys.stdin.isatty():
|
||||
if session_id is None:
|
||||
summary["status"] = "FAIL"
|
||||
summary["error"] = "session id missing for auto-answer resume"
|
||||
break
|
||||
resume_prompt = prompt_for_answer(question)
|
||||
auto_answer_count += 1
|
||||
last_auto_answer_text = resume_prompt
|
||||
summary["question_detected"] = question
|
||||
summary["auto_answer_used"] = True
|
||||
summary["auto_answer_text"] = resume_prompt
|
||||
summary["auto_answer_count"] = auto_answer_count
|
||||
summary["non_interactive_question_ignored"] = False
|
||||
continue
|
||||
if session_id is None:
|
||||
summary["status"] = "FAIL"
|
||||
summary["error"] = "session id missing for resume"
|
||||
@@ -766,20 +839,29 @@ def main() -> int:
|
||||
"error": "agentic loop did not run",
|
||||
}
|
||||
|
||||
summary.update(
|
||||
{
|
||||
"agents_path": str(agents_path),
|
||||
"prompt_path": str(prompt_path),
|
||||
"repo_root": str(repo_root),
|
||||
"attempts": attempts,
|
||||
"clarifying_questions": questions,
|
||||
"clarifying_question_count": len(questions),
|
||||
}
|
||||
)
|
||||
log_text = log_path.read_text(encoding="utf-8", errors="ignore")
|
||||
sandbox_block_evidence = detect_sandbox_block(log_text)
|
||||
if sandbox_block_evidence:
|
||||
summary["status"] = "FAIL"
|
||||
summary["error"] = (
|
||||
"Codex tool access appears to be sandbox-blocked. "
|
||||
"Re-run the integration test with escalated permissions."
|
||||
)
|
||||
summary["sandbox_blocked"] = True
|
||||
summary["sandbox_block_evidence"] = sandbox_block_evidence
|
||||
summary["requires_escalation"] = True
|
||||
print("Detected sandbox-blocked tool access; escalate permissions and re-run.")
|
||||
|
||||
if auto_answer_count:
|
||||
summary["auto_answer_count"] = auto_answer_count
|
||||
if last_auto_answer_text:
|
||||
summary["auto_answer_text"] = last_auto_answer_text
|
||||
|
||||
summary_path = run_dir / "agentic_summary.json"
|
||||
summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8")
|
||||
print(str(summary_path))
|
||||
if summary.get("requires_escalation"):
|
||||
return 3
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
+22
-10
@@ -511,16 +511,28 @@ def main() -> int:
|
||||
plan_path = write_plan_json(run_dir, prompt, Path.cwd())
|
||||
|
||||
if not args.skip_agentic_loop:
|
||||
run_step(
|
||||
[
|
||||
sys.executable,
|
||||
str(Path(__file__).resolve().parent / "run_agentic_loop.py"),
|
||||
"--out-dir",
|
||||
str(base_dir),
|
||||
"--run-dir",
|
||||
str(run_dir),
|
||||
]
|
||||
)
|
||||
agentic_cmd = [
|
||||
sys.executable,
|
||||
str(Path(__file__).resolve().parent / "run_agentic_loop.py"),
|
||||
"--out-dir",
|
||||
str(base_dir),
|
||||
"--run-dir",
|
||||
str(run_dir),
|
||||
]
|
||||
agentic_status = run_step(agentic_cmd)
|
||||
if agentic_status != 0:
|
||||
print(f"Agentic loop exited with code {agentic_status}.")
|
||||
return agentic_status
|
||||
|
||||
agentic_summary_path = run_dir / "agentic_summary.json"
|
||||
if agentic_summary_path.exists():
|
||||
agentic_summary = load_json(agentic_summary_path)
|
||||
if agentic_summary.get("requires_escalation"):
|
||||
print(
|
||||
"Agentic loop indicates sandbox-blocked access. "
|
||||
"Re-run the integration test with escalated permissions."
|
||||
)
|
||||
return 3
|
||||
|
||||
run_plan_cmd = [
|
||||
sys.executable,
|
||||
|
||||
+100
@@ -0,0 +1,100 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def load_rules_module():
|
||||
module_path = Path(__file__).resolve().parents[1] / "scripts" / "deterministic_rules.py"
|
||||
spec = importlib.util.spec_from_file_location("deterministic_rules", module_path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError(f"Unable to load deterministic_rules module from {module_path}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
RULES = load_rules_module()
|
||||
|
||||
|
||||
def write_run_dir(run_dir: Path, log_text: str) -> None:
|
||||
(run_dir / "logs").mkdir(parents=True, exist_ok=True)
|
||||
(run_dir / "prompt.json").write_text(json.dumps({"change_prompt": "x"}), encoding="utf-8")
|
||||
(run_dir / "logs" / "agentic.log").write_text(log_text, encoding="utf-8")
|
||||
|
||||
|
||||
def check(run_dir: Path) -> dict:
|
||||
return RULES.check_exec_plan_before_code_changes(
|
||||
run_dir,
|
||||
{"prompt_path": "prompt.json", "agentic_log_path": "logs/agentic.log"},
|
||||
)
|
||||
|
||||
|
||||
def test_planning_signal_before_code_change_passes(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-pass"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"Plan: inspect code paths",
|
||||
"file update",
|
||||
"M src/app.py",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "PASS"
|
||||
assert "before code changes" in result["rationale"]
|
||||
|
||||
|
||||
def test_planning_signal_after_code_change_fails(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-fail-ordering"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"file update",
|
||||
"M src/app.py",
|
||||
"Plan: now I will describe the approach",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "FAIL"
|
||||
assert "after code changes" in result["rationale"]
|
||||
|
||||
|
||||
def test_plan_file_path_does_not_count_as_planning_signal(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-plan-path"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"file update",
|
||||
"M docs/exec-plan.md",
|
||||
"file update",
|
||||
"M src/app.py",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "FAIL"
|
||||
assert "No planning signal detected" in result["rationale"]
|
||||
|
||||
|
||||
def test_no_code_changes_warns(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-warn-no-code"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"Plan: investigate the issue",
|
||||
"file update",
|
||||
"M docs/notes.md",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "WARN"
|
||||
assert "ordering not evaluated" in result["rationale"]
|
||||
+120
@@ -0,0 +1,120 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def load_rules_module():
|
||||
module_path = Path(__file__).resolve().parents[1] / "scripts" / "deterministic_rules.py"
|
||||
spec = importlib.util.spec_from_file_location("deterministic_rules", module_path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError(f"Unable to load deterministic_rules module from {module_path}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
RULES = load_rules_module()
|
||||
|
||||
|
||||
def write_run_dir(run_dir: Path, log_text: str, build_test_plan: list[dict] | None = None) -> None:
|
||||
(run_dir / "logs").mkdir(parents=True, exist_ok=True)
|
||||
prompt = {
|
||||
"change_prompt": "x",
|
||||
"build_test_plan": build_test_plan or [],
|
||||
}
|
||||
(run_dir / "prompt.json").write_text(json.dumps(prompt), encoding="utf-8")
|
||||
(run_dir / "logs" / "agentic.log").write_text(log_text, encoding="utf-8")
|
||||
|
||||
|
||||
def check(run_dir: Path) -> dict:
|
||||
return RULES.check_verification_after_code_changes(
|
||||
run_dir,
|
||||
{"prompt_path": "prompt.json", "agentic_log_path": "logs/agentic.log"},
|
||||
)
|
||||
|
||||
|
||||
def test_verification_command_after_code_change_passes(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-pass"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"file update",
|
||||
"M src/app.py",
|
||||
"$ pytest -q",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "PASS"
|
||||
assert "after code changes" in result["rationale"]
|
||||
|
||||
|
||||
def test_verification_only_before_code_change_fails(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-fail-order"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"$ pytest -q",
|
||||
"file update",
|
||||
"M src/app.py",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "FAIL"
|
||||
assert "No build/test/lint verification command detected" in result["rationale"]
|
||||
|
||||
|
||||
def test_prompt_plan_command_counts_as_verification(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-pass-prompt-command"
|
||||
build_test_plan = [{"label": "verify", "cmd": "make check-all"}]
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"file update",
|
||||
"M src/core.py",
|
||||
"$ make check-all",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text, build_test_plan=build_test_plan)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "PASS"
|
||||
|
||||
|
||||
def test_codex_exec_lines_do_not_count_as_verification(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-fail-codex-line"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"file update",
|
||||
"M src/app.py",
|
||||
'$ codex exec -C /repo "please run tests after this change"',
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "FAIL"
|
||||
assert "No build/test/lint verification command detected" in result["rationale"]
|
||||
|
||||
|
||||
def test_no_code_changes_warns(tmp_path: Path) -> None:
|
||||
run_dir = tmp_path / "run-warn-no-code"
|
||||
log_text = "\n".join(
|
||||
[
|
||||
"file update",
|
||||
"M docs/notes.md",
|
||||
"$ pytest -q",
|
||||
]
|
||||
)
|
||||
write_run_dir(run_dir, log_text)
|
||||
|
||||
result = check(run_dir)
|
||||
|
||||
assert result["status"] == "WARN"
|
||||
assert "ordering not evaluated" in result["rationale"]
|
||||
Reference in New Issue
Block a user