File size: 2,817 Bytes
d34961f | 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 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | #!/usr/bin/env python3
"""
harness-state skill script.
Operations: get_state, set_state
State is stored in .codex-plugin/harness_state.json relative to the repo root.
Falls back to the current working directory if the .codex-plugin dir is not found.
"""
import json
import os
import sys
from pathlib import Path
STATE_FILENAME = "harness_state.json"
DEFAULT_STATE = {
"active": False,
"phase": 0,
"phase_name": "",
}
def find_state_dir() -> Path:
"""Walk up from cwd looking for .codex-plugin/. Fall back to cwd."""
cwd = Path.cwd()
for parent in [cwd, *cwd.parents]:
candidate = parent / ".codex-plugin"
if candidate.is_dir():
return candidate
# Fallback: use cwd itself
return cwd
def state_path() -> Path:
return find_state_dir() / STATE_FILENAME
def read_state() -> dict:
path = state_path()
if not path.exists():
return dict(DEFAULT_STATE)
try:
with open(path) as f:
data = json.load(f)
# Fill missing keys with defaults
return {**DEFAULT_STATE, **data}
except (json.JSONDecodeError, OSError):
return dict(DEFAULT_STATE)
def write_state(active: bool, phase: int, phase_name: str) -> dict:
state = {"active": active, "phase": phase, "phase_name": phase_name}
path = state_path()
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(state, f, indent=2)
return state
def main():
if len(sys.argv) < 2:
print(json.dumps({"error": "Usage: state.py <json_input>"}))
sys.exit(1)
try:
args = json.loads(sys.argv[1])
except json.JSONDecodeError as e:
print(json.dumps({"error": f"Invalid JSON input: {e}"}))
sys.exit(1)
operation = args.get("operation")
if operation == "get_state":
result = read_state()
print(json.dumps(result))
elif operation == "set_state":
active = args.get("active")
phase = args.get("phase")
phase_name = args.get("phase_name", "")
if active is None or phase is None:
print(json.dumps({"error": "set_state requires 'active' (bool) and 'phase' (int)"}))
sys.exit(1)
if not isinstance(active, bool):
print(json.dumps({"error": "'active' must be a boolean"}))
sys.exit(1)
if not isinstance(phase, int):
print(json.dumps({"error": "'phase' must be an integer"}))
sys.exit(1)
result = write_state(active=active, phase=phase, phase_name=str(phase_name))
print(json.dumps({"ok": True, "state": result}))
else:
print(json.dumps({"error": f"Unknown operation '{operation}'. Valid: get_state, set_state"}))
sys.exit(1)
if __name__ == "__main__":
main() |