634525fa21
- Context isolation: fresh Pi process per stage (run_stage.py) - Gate enforcement: blocks close without approved gates - Auto commit/push on feature close (close_feature.py) - Write restrictions: only allowed directories (ALLOWED_WRITE_DIRS) - Pi extension: orquestra-status with /orquestra-stage command - Documentation: context-handoff.md, updated README - Scripts: agent_status.py, verify.sh, install.sh updated
120 lines
3.6 KiB
Python
Executable File
120 lines
3.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Close a feature: validates all gates are approved, updates backlog, and runs commit.
|
|
Usage: python3 scripts/close_feature.py <feature_id>
|
|
"""
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BACKLOG_PATH = ROOT / 'backlog' / 'features.json'
|
|
ARTIFACTS_DIR = ROOT / 'work' / 'artifacts'
|
|
COMMIT_SCRIPT = ROOT / 'scripts' / 'commit_feature.sh'
|
|
|
|
|
|
def now_iso():
|
|
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
|
|
|
|
|
|
def check_gate(feature_id, gate_file, expected_verdict='APPROVED'):
|
|
"""Check if a gate artifact exists and has the expected verdict."""
|
|
path = ARTIFACTS_DIR / feature_id / gate_file
|
|
if not path.exists():
|
|
return False, f"Missing {gate_file}"
|
|
try:
|
|
data = json.loads(path.read_text(encoding='utf-8'))
|
|
if data.get('verdict') != expected_verdict:
|
|
return False, f"{gate_file} verdict is {data.get('verdict')}, expected {expected_verdict}"
|
|
return True, None
|
|
except Exception as e:
|
|
return False, f"{gate_file} invalid: {e}"
|
|
|
|
|
|
def close_feature(feature_id):
|
|
"""Close a feature after validating all gates."""
|
|
print(f"Closing feature {feature_id}...")
|
|
|
|
# Check all required gates
|
|
gates = [
|
|
('reviewer.json', 'APPROVED'),
|
|
('security.json', 'APPROVED'),
|
|
('qa.json', 'APPROVED'),
|
|
('leader-close.json', None), # Just needs to exist
|
|
]
|
|
|
|
errors = []
|
|
for gate_file, verdict in gates:
|
|
if verdict:
|
|
ok, err = check_gate(feature_id, gate_file, verdict)
|
|
else:
|
|
path = ARTIFACTS_DIR / feature_id / gate_file
|
|
ok = path.exists()
|
|
err = f"Missing {gate_file}" if not ok else None
|
|
|
|
if not ok:
|
|
errors.append(err)
|
|
print(f" ❌ {err}")
|
|
else:
|
|
print(f" ✅ {gate_file}")
|
|
|
|
if errors:
|
|
print(f"\n[FAIL] Cannot close {feature_id}: {len(errors)} gate(s) missing or invalid")
|
|
return 1
|
|
|
|
# Update backlog
|
|
if not BACKLOG_PATH.exists():
|
|
print(f"[FAIL] Backlog not found: {BACKLOG_PATH}")
|
|
return 1
|
|
|
|
try:
|
|
backlog = json.loads(BACKLOG_PATH.read_text(encoding='utf-8'))
|
|
except Exception as e:
|
|
print(f"[FAIL] Invalid backlog: {e}")
|
|
return 1
|
|
|
|
feature_found = False
|
|
for feature in backlog.get('features', []):
|
|
if feature.get('id') == feature_id:
|
|
feature['status'] = 'done'
|
|
feature['completed_at'] = now_iso()
|
|
feature_found = True
|
|
print(f"\n[OK] Updated backlog: {feature_id} -> done")
|
|
break
|
|
|
|
if not feature_found:
|
|
print(f"[FAIL] Feature {feature_id} not found in backlog")
|
|
return 1
|
|
|
|
# Save backlog
|
|
try:
|
|
BACKLOG_PATH.write_text(json.dumps(backlog, indent=2, ensure_ascii=False) + '\n', encoding='utf-8')
|
|
except Exception as e:
|
|
print(f"[FAIL] Cannot save backlog: {e}")
|
|
return 1
|
|
|
|
# Run commit script
|
|
if COMMIT_SCRIPT.exists():
|
|
print(f"\nRunning commit script...")
|
|
result = subprocess.run([str(COMMIT_SCRIPT), feature_id], cwd=ROOT)
|
|
if result.returncode != 0:
|
|
print(f"[WARN] Commit script returned {result.returncode}")
|
|
|
|
print(f"\n✅ Feature {feature_id} closed successfully")
|
|
return 0
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print("Usage: python3 scripts/close_feature.py <feature_id>")
|
|
return 1
|
|
|
|
feature_id = sys.argv[1]
|
|
return close_feature(feature_id)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|