claudeandClaude Sonnet 5
70b7a8bbcd
Inspired by openai/codex#26966 (closed not planned). codex_goal.py is a standalone CLI that stores one versioned, project-keyed goal per repo outside Codex's own sandbox, and injects it as plain user context on every new or resumed `codex exec` run. Writes are atomic and locked, runs never mutate the goal, Codex is spawned via argv (no shell), and signals are forwarded for normal cancellation behavior. 16 unit tests plus a recorded two-run + crash-recovery demo, verified on macOS and in the project's Colima Linux container. Launch page at ventures/codex-goal/site/index.html uses a self-contained scoreboard/arcade identity (vendored OFL DSEG7 font, no external requests) distinct from the other ventures. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
249 lines
9.4 KiB
Python
249 lines
9.4 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, ROOT)
|
|
FAKE_CODEX = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fake_codex.py")
|
|
|
|
import codex_goal # noqa: E402
|
|
|
|
|
|
class TempHome(unittest.TestCase):
|
|
def setUp(self):
|
|
self._home_dir = tempfile.TemporaryDirectory()
|
|
self._project_dir = tempfile.TemporaryDirectory()
|
|
os.environ["CODEX_GOAL_HOME"] = self._home_dir.name
|
|
os.chmod(FAKE_CODEX, 0o755)
|
|
|
|
def tearDown(self):
|
|
os.environ.pop("CODEX_GOAL_HOME", None)
|
|
self._home_dir.cleanup()
|
|
self._project_dir.cleanup()
|
|
|
|
def run_cli(self, *argv):
|
|
return codex_goal.main(list(argv))
|
|
|
|
@property
|
|
def project(self):
|
|
return self._project_dir.name
|
|
|
|
|
|
class SetShowTests(TempHome):
|
|
def test_set_then_show_roundtrip(self):
|
|
rc = self.run_cli("set", "-C", self.project, "Ship", "the", "thing")
|
|
self.assertEqual(rc, 0)
|
|
|
|
key = codex_goal.project_key(self.project)
|
|
state = codex_goal.load_state(codex_goal.goal_path(codex_goal.state_home(), key))
|
|
self.assertEqual(state["current"]["text"], "Ship the thing")
|
|
self.assertEqual(state["history"], [])
|
|
|
|
def test_show_without_goal_fails_clearly(self):
|
|
rc = self.run_cli("show", "-C", self.project)
|
|
self.assertEqual(rc, 1)
|
|
|
|
def test_second_set_pushes_history(self):
|
|
self.run_cli("set", "-C", self.project, "first goal")
|
|
self.run_cli("set", "-C", self.project, "second goal")
|
|
|
|
key = codex_goal.project_key(self.project)
|
|
state = codex_goal.load_state(codex_goal.goal_path(codex_goal.state_home(), key))
|
|
self.assertEqual(state["current"]["text"], "second goal")
|
|
self.assertEqual(len(state["history"]), 1)
|
|
self.assertEqual(state["history"][0]["text"], "first goal")
|
|
|
|
def test_empty_goal_rejected(self):
|
|
rc = self.run_cli("set", "-C", self.project, " ")
|
|
self.assertEqual(rc, 2)
|
|
|
|
def test_two_projects_get_independent_state(self):
|
|
with tempfile.TemporaryDirectory() as other:
|
|
self.run_cli("set", "-C", self.project, "goal A")
|
|
self.run_cli("set", "-C", other, "goal B")
|
|
|
|
key_a = codex_goal.project_key(self.project)
|
|
key_b = codex_goal.project_key(other)
|
|
self.assertNotEqual(key_a, key_b)
|
|
|
|
state_a = codex_goal.load_state(codex_goal.goal_path(codex_goal.state_home(), key_a))
|
|
state_b = codex_goal.load_state(codex_goal.goal_path(codex_goal.state_home(), key_b))
|
|
self.assertEqual(state_a["current"]["text"], "goal A")
|
|
self.assertEqual(state_b["current"]["text"], "goal B")
|
|
|
|
def test_state_lives_outside_project_dir(self):
|
|
self.run_cli("set", "-C", self.project, "stay external")
|
|
key = codex_goal.project_key(self.project)
|
|
path = codex_goal.goal_path(codex_goal.state_home(), key)
|
|
self.assertFalse(path.startswith(os.path.realpath(self.project)))
|
|
self.assertTrue(path.startswith(self._home_dir.name))
|
|
|
|
|
|
class RunDryRunTests(TempHome):
|
|
def test_run_without_goal_errors_and_does_not_spawn(self):
|
|
rc = self.run_cli("run", "-C", self.project, "--dry-run")
|
|
self.assertEqual(rc, 2)
|
|
|
|
def test_dry_run_new_session_injects_goal(self):
|
|
self.run_cli("set", "-C", self.project, "win the game")
|
|
rc = self.run_cli("run", "-C", self.project, "--codex-bin", "codex", "--dry-run")
|
|
self.assertEqual(rc, 0)
|
|
|
|
def test_dry_run_resume_last_builds_resume_argv(self):
|
|
self.run_cli("set", "-C", self.project, "resume goal")
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = self.run_cli("run", "-C", self.project, "--last", "--dry-run")
|
|
self.assertEqual(rc, 0)
|
|
payload = json.loads(buf.getvalue())
|
|
self.assertEqual(payload["mode"], "resume")
|
|
self.assertIn("resume", payload["argv"])
|
|
self.assertIn("--last", payload["argv"])
|
|
|
|
def test_shell_metacharacters_survive_as_one_argv_item(self):
|
|
goal = "ship it; rm -rf / #not a shell command"
|
|
self.run_cli("set", "-C", self.project, goal)
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
self.run_cli("run", "-C", self.project, "--dry-run")
|
|
payload = json.loads(buf.getvalue())
|
|
self.assertIn(goal, payload["argv"][-1])
|
|
# the dangerous text is one argv element, never concatenated into a command string
|
|
self.assertEqual(sum(goal in item for item in payload["argv"]), 1)
|
|
|
|
|
|
class RunSpawnTests(TempHome):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self._record = tempfile.NamedTemporaryFile(delete=False, suffix=".json")
|
|
self._record.close()
|
|
os.environ["FAKE_CODEX_RECORD"] = self._record.name
|
|
|
|
def tearDown(self):
|
|
os.environ.pop("FAKE_CODEX_RECORD", None)
|
|
os.environ.pop("FAKE_CODEX_EXIT", None)
|
|
os.environ.pop("FAKE_CODEX_SLEEP", None)
|
|
with contextlib.suppress(OSError):
|
|
os.remove(self._record.name)
|
|
super().tearDown()
|
|
|
|
def test_run_spawns_fake_codex_injects_goal_and_logs(self):
|
|
self.run_cli("set", "-C", self.project, "beat the deadline")
|
|
os.environ["FAKE_CODEX_EXIT"] = "0"
|
|
rc = self.run_cli("run", "-C", self.project, "--codex-bin", FAKE_CODEX)
|
|
self.assertEqual(rc, 0)
|
|
|
|
with open(self._record.name) as fh:
|
|
recorded = json.load(fh)
|
|
self.assertEqual(recorded["argv"][0], "exec")
|
|
self.assertIn("beat the deadline", recorded["argv"][-1])
|
|
self.assertIn(codex_goal.INJECT_HEADER, recorded["argv"][-1])
|
|
|
|
key = codex_goal.project_key(self.project)
|
|
log_path = codex_goal.runlog_path(codex_goal.state_home(), key)
|
|
with open(log_path) as fh:
|
|
lines = [json.loads(l) for l in fh if l.strip()]
|
|
self.assertEqual(len(lines), 1)
|
|
self.assertEqual(lines[0]["mode"], "new")
|
|
self.assertEqual(lines[0]["exit_code"], 0)
|
|
|
|
def test_run_preserves_nonzero_exit_status(self):
|
|
self.run_cli("set", "-C", self.project, "goal")
|
|
os.environ["FAKE_CODEX_EXIT"] = "17"
|
|
rc = self.run_cli("run", "-C", self.project, "--codex-bin", FAKE_CODEX)
|
|
self.assertEqual(rc, 17)
|
|
|
|
def test_two_runs_both_see_the_same_persisted_goal(self):
|
|
self.run_cli("set", "-C", self.project, "persist across sessions")
|
|
os.environ["FAKE_CODEX_EXIT"] = "0"
|
|
|
|
self.run_cli("run", "-C", self.project, "--codex-bin", FAKE_CODEX)
|
|
with open(self._record.name) as fh:
|
|
first = json.load(fh)
|
|
|
|
self.run_cli("run", "-C", self.project, "--last", "--codex-bin", FAKE_CODEX)
|
|
with open(self._record.name) as fh:
|
|
second = json.load(fh)
|
|
|
|
self.assertIn("persist across sessions", first["argv"][-1])
|
|
self.assertIn("persist across sessions", second["argv"][-1])
|
|
self.assertIn("resume", second["argv"])
|
|
|
|
key = codex_goal.project_key(self.project)
|
|
log_path = codex_goal.runlog_path(codex_goal.state_home(), key)
|
|
with open(log_path) as fh:
|
|
lines = [json.loads(l) for l in fh if l.strip()]
|
|
self.assertEqual(len(lines), 2)
|
|
self.assertEqual(lines[0]["mode"], "new")
|
|
self.assertEqual(lines[1]["mode"], "resume")
|
|
|
|
def test_failed_run_does_not_touch_the_goal(self):
|
|
self.run_cli("set", "-C", self.project, "do not corrupt me")
|
|
os.environ["FAKE_CODEX_EXIT"] = "1"
|
|
self.run_cli("run", "-C", self.project, "--codex-bin", FAKE_CODEX)
|
|
|
|
key = codex_goal.project_key(self.project)
|
|
state = codex_goal.load_state(codex_goal.goal_path(codex_goal.state_home(), key))
|
|
self.assertEqual(state["current"]["text"], "do not corrupt me")
|
|
|
|
def test_sigint_is_forwarded_and_cancels_child(self):
|
|
self.run_cli("set", "-C", self.project, "long running")
|
|
os.environ["FAKE_CODEX_SLEEP"] = "5"
|
|
|
|
child_argv = [
|
|
sys.executable,
|
|
os.path.join(ROOT, "codex_goal.py"),
|
|
"run",
|
|
"-C",
|
|
self.project,
|
|
"--codex-bin",
|
|
FAKE_CODEX,
|
|
]
|
|
env = dict(os.environ)
|
|
proc = subprocess.Popen(child_argv, env=env)
|
|
time.sleep(0.5)
|
|
proc.send_signal(signal.SIGINT)
|
|
rc = proc.wait(timeout=5)
|
|
self.assertEqual(rc, 130)
|
|
|
|
|
|
class AtomicWriteTests(TempHome):
|
|
def test_failed_write_leaves_existing_state_intact(self):
|
|
self.run_cli("set", "-C", self.project, "original goal")
|
|
key = codex_goal.project_key(self.project)
|
|
path = codex_goal.goal_path(codex_goal.state_home(), key)
|
|
with open(path) as fh:
|
|
before = fh.read()
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
with codex_goal.locked(codex_goal.state_home(), key):
|
|
state = codex_goal.load_state(path)
|
|
state["current"]["text"] = "corrupted"
|
|
|
|
real_dump = json.dump
|
|
|
|
def exploding_dump(*a, **k):
|
|
raise RuntimeError("simulated crash mid-write")
|
|
|
|
json.dump = exploding_dump
|
|
try:
|
|
codex_goal.atomic_write_json(path, state)
|
|
finally:
|
|
json.dump = real_dump
|
|
|
|
with open(path) as fh:
|
|
after = fh.read()
|
|
self.assertEqual(before, after)
|
|
self.assertNotIn("corrupted", after)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|