chore(harness): isolate pi stage execution

2026-08-15 09:51:25 +02:00
parent 546971280f
commit cf1c69fc8b
9 changed files with 205 additions and 24 deletions
--- a/scripts/run_stage.py
+++ b/scripts/run_stage.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+import argparse
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKFLOW = ROOT / 'harness' / 'workflow.stages.yml'
+STATUS_EXT = ROOT / '.pi' / 'extensions' / 'orquestra-status'
+WEB_FETCH_EXT = ROOT / '.pi' / 'extensions' / 'orquestra-web-fetch.ts'
+ENGRAM_EXT = Path.home() / '.pi' / 'agent' / 'npm' / 'node_modules' / 'gentle-engram' / 'index.ts'
+
+
+def parse_stages():
+    stages = {}
+    current = None
+    section = None
+    for raw in WORKFLOW.read_text(encoding='utf-8').splitlines():
+        line = raw.rstrip()
+        if line.startswith('  - name:'):
+            current = {'input': [], 'output': []}
+            current['name'] = line.split(':', 1)[1].strip()
+            stages[current['name']] = current
+            section = None
+            continue
+        if current is None:
+            continue
+        if line.startswith('    owner:'):
+            current['owner'] = line.split(':', 1)[1].strip()
+        elif line.startswith('    input:'):
+            section = 'input'
+        elif line.startswith('    output:'):
+            section = 'output'
+        elif section and line.startswith('      - '):
+            current[section].append(line.split('- ', 1)[1].strip())
+        elif line.startswith('    '):
+            section = None
+    return stages
+
+
+def expand(paths, feature_id):
+    return [p.replace('<feature_id>', feature_id) for p in paths]
+
+
+def prompt_for(stage, feature_id):
+    inputs = expand(stage.get('input', []), feature_id)
+    outputs = expand(stage.get('output', []), feature_id)
+    owner = stage.get('owner', 'leader')
+    return f"""You are the Orquestra stage agent '{owner}' for feature '{feature_id}'.
+
+Fresh-process rule: do not rely on previous chat/session context. Use only the files listed here, explicit Engram memories you choose after a narrow search, and repository evidence you read yourself.
+
+First read these harness rules:
+- AGENTS.md
+- harness/agents.matrix.yml
+- harness/workflow.stages.yml
+- harness/contracts/handoff.md
+
+Stage input paths:
+{chr(10).join(f'- {p}' for p in inputs) or '- none'}
+
+Stage output paths:
+{chr(10).join(f'- {p}' for p in outputs) or '- none'}
+
+Rules:
+- Before work, run: python3 scripts/agent_status.py set --feature-id {feature_id} --stage {stage['name']} --agent {owner} --state running --action "Running {stage['name']}"
+- Read only the stage inputs that exist. If a required input is missing, write a blocked artifact when possible and answer blocked -> <path>.
+- Do not carry or summarize previous chat. Previous stage context is on disk in the declared input artifacts.
+- Write evidence to the declared output path(s). Do not return code in chat.
+- Finish by updating runtime status to done or blocked.
+- Final response must be exactly: done -> <path> or blocked -> <path>.
+"""
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Run one Orquestra stage in a fresh Pi process with minimal declared context.')
+    parser.add_argument('stage')
+    parser.add_argument('--feature-id', required=True)
+    parser.add_argument('--dry-run', action='store_true')
+    args = parser.parse_args()
+
+    stages = parse_stages()
+    if args.stage not in stages:
+        print(f"[FAIL] Unknown stage: {args.stage}", file=sys.stderr)
+        print(f"Allowed: {', '.join(stages)}", file=sys.stderr)
+        return 2
+
+    cmd = [
+        'pi', '--no-session', '--no-context-files', '--no-extensions', '--no-skills',
+        '-e', str(ENGRAM_EXT), '-e', str(STATUS_EXT), '-e', str(WEB_FETCH_EXT),
+        '-p', prompt_for(stages[args.stage], args.feature_id),
+    ]
+    if args.dry_run:
+        print(' '.join(cmd[:-1]))
+        print('\n--- prompt ---')
+        print(cmd[-1])
+        return 0
+
+    missing = []
+    if not ENGRAM_EXT.is_file():
+        missing.append(str(ENGRAM_EXT))
+    if not (STATUS_EXT / 'index.ts').is_file():
+        missing.append(str(STATUS_EXT / 'index.ts'))
+    if not WEB_FETCH_EXT.is_file():
+        missing.append(str(WEB_FETCH_EXT))
+    if missing:
+        print('[FAIL] Missing Pi runtime extension(s):', file=sys.stderr)
+        for path in missing:
+            print(f'  - {path}', file=sys.stderr)
+        print('Run scripts/install.sh for this project and install gentle-engram.', file=sys.stderr)
+        return 1
+    return subprocess.run(cmd, cwd=ROOT, env=os.environ.copy()).returncode
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())