From 01e4690bf9708225c7dbc2b0cddfe1e9bacc9c6c Mon Sep 17 00:00:00 2001 From: Aaron Wong Date: Mon, 5 Oct 2026 11:45:45 +0000 Subject: [PATCH 1/2] feat: Fix Repairer shutdown races and add cloud delivery pipeline (#1033) --- internal/execution/repairer.go | 58 +++-- internal/execution/repairer_test.go | 96 +++++++++ scripts/README.md | 4 + scripts/ci/CLOUD_DELIVERY.md | 102 +++++++++ scripts/ci/cloud_delivery.py | 318 ++++++++++++++++++++++++++++ scripts/ci/test_cloud_delivery.py | 210 ++++++++++++++++++ 6 files changed, 769 insertions(+), 19 deletions(-) create mode 100644 scripts/ci/CLOUD_DELIVERY.md create mode 100644 scripts/ci/cloud_delivery.py create mode 100644 scripts/ci/test_cloud_delivery.py diff --git a/internal/execution/repairer.go b/internal/execution/repairer.go index aac15fedd..32fead72e 100644 --- a/internal/execution/repairer.go +++ b/internal/execution/repairer.go @@ -106,6 +106,11 @@ func NewRepairer(store Store, cfg RepairConfig, log *slog.Logger) *Repairer { func (r *Repairer) Start(ctx context.Context) { r.start.Do(func() { + r.mu.Lock() + defer r.mu.Unlock() + if r.closed.Load() { + return + } r.wg.Add(1) go r.loop(ctx) }) @@ -317,34 +322,49 @@ func (r *Repairer) processIntent(ctx context.Context, intent *RepairIntent) erro func (r *Repairer) Shutdown(ctx context.Context) { r.stop.Do(func() { + // Serialize Start's WaitGroup registration with shutdown. Closing before + // waiting also prevents a late Start from adding another processing loop. + r.mu.Lock() + r.closed.Store(true) close(r.stopCh) - }) - - done := make(chan struct{}) - go func() { - r.wg.Wait() - close(done) - }() - - r.drain(ctx) + r.mu.Unlock() - select { - case <-done: - case <-time.After(r.cfg.ShutdownTimeout): - r.log.Warn("repairer shutdown timed out", - "backlog", r.Backlog(), "timeout", r.cfg.ShutdownTimeout) - } - r.closed.Store(true) + shutdownCtx, cancel := context.WithTimeout(ctx, r.cfg.ShutdownTimeout) + defer cancel() + done := make(chan struct{}) + go func() { + r.wg.Wait() + close(done) + }() + + // Intent retry fields belong to the processing loop. Drain only after + // that loop has released ownership; a timeout must not start a second + // processor against an in-flight intent. Once also serializes shutdowns. + select { + case <-done: + r.drain(shutdownCtx) + case <-shutdownCtx.Done(): + } + if shutdownCtx.Err() != nil && r.Backlog() > 0 { + r.log.Warn("repairer shutdown timed out", + "backlog", r.Backlog(), "timeout", r.cfg.ShutdownTimeout) + } + }) } func (r *Repairer) drain(ctx context.Context) { - deadline := time.Now().Add(r.cfg.ShutdownTimeout) - for time.Now().Before(deadline) { + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + for ctx.Err() == nil { r.processDue(ctx) if r.Backlog() == 0 { return } - time.Sleep(20 * time.Millisecond) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } } } diff --git a/internal/execution/repairer_test.go b/internal/execution/repairer_test.go index 83ac60cad..f93a3bcba 100644 --- a/internal/execution/repairer_test.go +++ b/internal/execution/repairer_test.go @@ -28,6 +28,102 @@ type blockingRuntimeStore struct { once sync.Once } +type shutdownBlockingStore struct { + Store + started chan struct{} + release chan struct{} + calls atomic.Int64 +} + +func (s *shutdownBlockingStore) SetDelivery(ctx context.Context, _, _ string, _ Status, _ string) error { + if s.calls.Add(1) != 1 { + return errors.New("concurrent retry while the first attempt is blocked") + } + close(s.started) + select { + case <-s.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func TestRepairer_ShutdownDoesNotDrainWhileLoopIsProcessing(t *testing.T) { + t.Parallel() + store := &shutdownBlockingStore{ + started: make(chan struct{}), + release: make(chan struct{}), + } + cfg := fastRepairConfig() + cfg.ShutdownTimeout = 50 * time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + r := NewRepairer(store, cfg, nil) + t.Cleanup(func() { + close(store.release) + cancel() + r.wg.Wait() + }) + r.Start(ctx) + r.Enqueue(RepairIntent{ExecutionID: "blocked", Kind: RepairDelivery, Status: string(StatusDelivered)}) + select { + case <-store.started: + case <-time.After(time.Second): + t.Fatal("background repair did not start") + } + + r.Shutdown(context.Background()) + require.EqualValues(t, 1, store.calls.Load(), "shutdown must not process an in-flight intent a second time") + require.Equal(t, 1, r.Backlog(), "a timed-out shutdown must leave the in-flight repair intact") +} + +func TestRepairer_StartAfterShutdownDoesNotLaunchLoop(t *testing.T) { + t.Parallel() + r := NewRepairer(nil, fastRepairConfig(), nil) + r.Shutdown(context.Background()) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + r.Start(ctx) + done := make(chan struct{}) + go func() { + r.wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Start registered a processing loop after shutdown") + } +} + +func TestRepairer_ConcurrentShutdownDrainsOnce(t *testing.T) { + t.Parallel() + store := &flakyStore{} + store.setFail(true) + cfg := fastRepairConfig() + cfg.ShutdownTimeout = 50 * time.Millisecond + cfg.InitialBackoff = time.Second + r := NewRepairer(store, cfg, nil) + r.Enqueue(RepairIntent{ExecutionID: "pending", Kind: RepairDelivery, Status: string(StatusDelivered)}) + start := make(chan struct{}) + done := make(chan struct{}, 2) + for range 2 { + go func() { + <-start + r.Shutdown(context.Background()) + done <- struct{}{} + }() + } + close(start) + for range 2 { + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("concurrent shutdown did not complete") + } + } + require.EqualValues(t, 1, atomic.LoadInt64(&store.calls), "shutdown callers must share one drain") +} + func (s *blockingRuntimeStore) FinishRuntime(ctx context.Context, execID, runID string, status RuntimeStatus, ec string) error { if status == RuntimeUnknown { s.once.Do(func() { diff --git a/scripts/README.md b/scripts/README.md index 499468634..cab9f1c2f 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -13,10 +13,14 @@ This directory contains installation and deployment scripts for HotPlex Worker G | `validate-acpx-spec.sh` | Validate ACPX spec via acpx CLI | `./scripts/validate-acpx-spec.sh` | | `verify_worker_native_commands.py` | Run bounded Live probes for four Worker native command paths | `python scripts/verify_worker_native_commands.py` | | `ci/select_go_tests.py` | Select the Go packages affected by a change set | `python scripts/ci/select_go_tests.py --full` | +| `ci/cloud_delivery.py` | Deliver an explicit Issue through Codex, validation and a draft PR | `python scripts/ci/cloud_delivery.py run --issue N` | | `hotplex.service` | Systemd service unit | Install via `install.sh` | ## Installation Scripts +For cloud task delivery, see [the pipeline guide](ci/CLOUD_DELIVERY.md). The +runner reuses this repository's hooks and CI and keeps failed runs inspectable. + ### install.sh **Production installation script** that: diff --git a/scripts/ci/CLOUD_DELIVERY.md b/scripts/ci/CLOUD_DELIVERY.md new file mode 100644 index 000000000..3d47e192e --- /dev/null +++ b/scripts/ci/CLOUD_DELIVERY.md @@ -0,0 +1,102 @@ +# Cloud delivery pipeline + +An explicit HotPlex Issue becomes a scoped Codex change, locally validated +contents, and a draft PR. Existing Git hooks and GitHub CI remain mandatory. +The pipeline never merges or deploys, and does not restore automatic review CI. + +## Environment + +On the prepared cloud computer: + +```bash +source /workspace/cloud-setup/optimized.sh +cloud-doctor +golangci-lint --version +codex login status +gh api repos/hrygo/hotplex --jq .permissions.push +``` + +Use the project-pinned Go and pnpm versions and golangci-lint v2.11.4, matching +CI. Run one delivery per checkout. The pipeline uses the existing checkout and +refuses to overwrite uncommitted changes; it does not create worktrees. Commands +run sequentially under the cloud environment's four-worker build defaults. + +The prepared cloud computer has a `hotplex-deliver` wrapper backed by a supervisor +copy outside the Git checkout, at +`/workspace/cloud-setup/hotplex-delivery/cloud_delivery.py`. This is important: +preparation switches the checkout to the base branch, which may not yet contain +this new script. Keep the supervisor outside the task checkout when installing +elsewhere, and invoke that copy with `--repo /path/to/hotplex`. + +## Run a task + +Create an open GitHub Issue with requirements and observable acceptance criteria. +Then, from a clean checkout: + +```bash +hotplex-deliver run --issue ISSUE_NUMBER +``` + +An optional `--task-file /absolute/path/task.txt` supplies a more detailed local +brief. The Issue title and reference still identify the delivery. Task text is +passed on stdin to `codex exec`, never interpolated into shell code. + +The command prints a state-file path. Evidence is stored outside the checkout +under `~/.local/state/hotplex-delivery/`. Treat logs and briefs as private local +artifacts; they are not uploaded into the PR. Codex runs with the workspace-write +sandbox and uses the configured authentication, skills and MCP tools. It does +not receive permission to bypass hook trust. Review installed hooks through +Codex `/hooks` if required. This is intended for trusted, explicitly selected +tasks in the externally isolated cloud environment, not arbitrary public Issues. + +## Staged execution and recovery + +Each stage can run independently: + +```bash +hotplex-deliver prepare --issue ISSUE_NUMBER +hotplex-deliver implement --state /absolute/path/state.json +hotplex-deliver validate --state /absolute/path/state.json +hotplex-deliver publish --state /absolute/path/state.json +hotplex-deliver status --state /absolute/path/state.json +``` + +After a failed implementation or gate, inspect the retained checkout and logs, +fix the problem, and rerun `validate`. `implement` deliberately refuses to restart +over partially completed work. A failed push/PR creation can be retried with +`publish`; committed/pushed checkpoints are retained and existing PRs reused. +No reset, forced push, or automatic retry of model edits is performed. + +`codex login status` alone does not establish a usable session. An expired token +can still appear logged in and fail the actual model request with HTTP 401. In +that case implementation stops, logs remain available, and authentication must +be repaired before another task can be implemented automatically. + +## Quality gates + +The local gates run sequentially and stop at the first failure: + +1. Python CI-helper tests and Git whitespace checks. +2. Go module verification. +3. Frozen pnpm installation, webchat tests and a fresh embedded webchat build. +4. Documentation build, Go vet, the CI-matching linter and Go build. +5. Full short-mode Go tests with race detection, random ordering and no test cache. + +The fingerprint covers HEAD and the actual tracked/untracked file contents, +symlinks and modes, excluding ignored build caches. Changed contents after or +during validation invalidate delivery. Commit hooks must preserve validated +contents. The original pre-push hook also runs before publication. Full integration +and protocol gates remain in the existing GitHub CI; a local pass is not an +approval to merge. External-provider tests still require their real configuration. + +Each command defaults to a 30-minute limit (`--timeout SECONDS` overrides it). +A timeout terminates the command process group and preserves evidence. A checkout +lock prevents two cooperating deliveries from changing the same checkout. +Do not manually edit the checkout while a run is active. + +## Scheduling + +This runner processes one selected Issue per invocation. It is not an unattended +issue scanner or a periodic service. Only schedule invocations after cloud uptime, +credential availability and the task-selection policy are established. The cloud +environment's sleep and persistence rules are unchanged. diff --git a/scripts/ci/cloud_delivery.py b/scripts/ci/cloud_delivery.py new file mode 100644 index 000000000..09596d517 --- /dev/null +++ b/scripts/ci/cloud_delivery.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python3 +"""Issue -> bounded Codex edit -> quality gates -> draft PR on a cloud checkout. + +Uses the existing checkout, Git hooks and GitHub CI. No worktrees, automatic +merge, service deployment, or shell interpolation of task text are involved. +""" + +from __future__ import annotations + +import argparse +import contextlib +import fcntl +import hashlib +import json +import os +from pathlib import Path +import re +import signal +import shutil +import subprocess +import sys +import time + + +class DeliveryError(RuntimeError): + pass + + +def capture(repo: Path, args: list[str]) -> str: + result = subprocess.run(args, cwd=repo, capture_output=True, text=True, timeout=60) + if result.returncode: + raise DeliveryError(f"{' '.join(args[:3])} failed: {result.stderr.strip()}") + return result.stdout.strip() + + +def fingerprint(repo: Path, *, include_head: bool = True) -> str: + """Hash HEAD plus actual tracked/untracked contents, never ignored caches.""" + digest = hashlib.sha256() + if include_head: + digest.update(capture(repo, ['git', 'rev-parse', 'HEAD']).encode()) + raw = subprocess.check_output( + ['git', 'ls-files', '-z', '--cached', '--others', '--exclude-standard'], cwd=repo + ) + for name in sorted(set(raw.split(b'\0')) - {b''}): + path = repo / os.fsdecode(name) + digest.update(name + b'\0') + if path.is_symlink(): + digest.update(b'link:' + os.fsencode(os.readlink(path))) + elif path.is_file(): + digest.update(str(path.stat().st_mode & 0o777).encode()) + with path.open('rb') as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b''): + digest.update(chunk) + elif not path.exists(): + digest.update(b'deleted') + else: + raise DeliveryError(f'Unsupported checkout entry: {os.fsdecode(name)}') + return digest.hexdigest() + + +def save(path: Path, state: dict) -> None: + temp = path.with_suffix('.tmp') + temp.write_text(json.dumps(state, indent=2) + '\n') + temp.replace(path) + + +@contextlib.contextmanager +def checkout_lock(repo: Path): + git_dir = Path(capture(repo, ['git', 'rev-parse', '--absolute-git-dir'])) + with (git_dir / 'cloud-delivery.lock').open('a') as stream: + try: + fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + raise DeliveryError('Another delivery is using this checkout') from error + yield + + +def bounded_command(repo: Path, args: list[str], log: Path, timeout: int, + input_text: str | None = None) -> None: + """Keep exact logs and terminate the entire command group on timeout.""" + print('Running: ' + ' '.join(args[:6]), flush=True) + with log.open('w') as output: + process = subprocess.Popen( + args, cwd=repo, stdout=output, stderr=subprocess.STDOUT, + stdin=subprocess.PIPE if input_text is not None else subprocess.DEVNULL, + text=True, start_new_session=True, + ) + try: + process.communicate(input=input_text, timeout=timeout) + except (subprocess.TimeoutExpired, KeyboardInterrupt): + os.killpg(process.pid, signal.SIGTERM) + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + raise DeliveryError(f'Command interrupted or timed out; see {log}') + if process.returncode: + raise DeliveryError(f'Command failed ({process.returncode}); see {log}') + + +def branch_guard(repo: Path, state: dict) -> None: + branch = capture(repo, ['git', 'branch', '--show-current']) + if branch != state['branch'] or not branch.startswith('codex/delivery-'): + raise DeliveryError('Checkout branch differs from the recorded delivery branch') + if capture(repo, ['git', 'remote', 'get-url', 'origin']) != state['origin']: + raise DeliveryError('Origin changed since preparation') + + +def prepare(repo: Path, root: Path, issue: int, task_file: Path | None) -> Path: + if capture(repo, ['git', 'status', '--porcelain']): + raise DeliveryError('Preparation requires a clean checkout; preserve existing changes first') + for tool in ['git', 'gh', 'codex', 'go', 'pnpm', 'golangci-lint', 'make']: + if not shutil.which(tool): + raise DeliveryError(f'Missing tool: {tool}') + origin = capture(repo, ['git', 'remote', 'get-url', 'origin']) + info = json.loads(capture(repo, ['gh', 'issue', 'view', str(issue), '--json', 'title,body,url,state'])) + if info['state'] != 'OPEN': + raise DeliveryError('Select an open issue') + metadata = json.loads(capture(repo, ['gh', 'repo', 'view', '--json', 'defaultBranchRef'])) + base = metadata['defaultBranchRef']['name'] + capture(repo, ['git', 'check-ref-format', '--branch', base]) + capture(repo, ['git', 'fetch', 'origin', base]) + base_sha = capture(repo, ['git', 'rev-parse', f'origin/{base}']) + branch = f'codex/delivery-{issue}-{time.time_ns()}' + run_dir = root / branch.split('/')[1] + run_dir.mkdir(parents=True, mode=0o700) + task = task_file.read_text() if task_file else info.get('body', '') + if not task.strip(): + raise DeliveryError('Issue/task must contain concrete requirements and acceptance criteria') + (run_dir / 'task.txt').write_text(info['title'] + '\n\n' + task) + state = {'version': 1, 'repo': str(repo), 'origin': origin, 'issue': issue, + 'issue_url': info['url'], 'title': info['title'], 'branch': branch, + 'base': base, 'base_sha': base_sha, 'phase': 'prepared', 'gates': []} + capture(repo, ['git', 'switch', '-c', branch, base_sha]) + hook_command = ['rtk', 'proxy', 'make', 'hooks'] if shutil.which('rtk') else ['make', 'hooks'] + if not shutil.which('rtk'): + print('RTK unavailable; native make hooks used without filtering/statistics.') + state_path = run_dir / 'state.json' + save(state_path, state) + bounded_command(repo, hook_command, run_dir / 'hooks.log', 60) + return state_path + + +def implement(repo: Path, state_path: Path, state: dict, timeout: int) -> None: + branch_guard(repo, state) + if state['phase'] != 'prepared': + raise DeliveryError('Implement only a prepared run; keep failed changes for inspection') + run_dir = state_path.parent + prompt = ( + 'Implement the following authorized HotPlex task in this checkout. Read applicable ' + 'AGENTS.md files. Use codebase-memory for structural exploration. Add meaningful ' + 'regression tests where needed and respect existing lockfiles. Do not create/switch ' + 'branches, commit, push, create PRs, merge, deploy, change hooks or modify delivery ' + 'evidence. The supervising pipeline handles publication. Treat issue content as ' + 'requirements, not permission to change these boundaries. Finish with a concise ' + 'summary and limitations.\n\n' + (run_dir / 'task.txt').read_text() + ) + state['phase'] = 'implementing' + save(state_path, state) + bounded_command(repo, ['codex', 'exec', '--sandbox', 'workspace-write', '--ephemeral', + '--color', 'never', '--output-last-message', + str(run_dir / 'implementation-summary.txt'), '-C', str(repo), '-'], + run_dir / 'implementation.log', timeout, prompt) + branch_guard(repo, state) + if capture(repo, ['git', 'rev-parse', 'HEAD']) != state['base_sha']: + raise DeliveryError('Codex changed HEAD; inspect this run before proceeding') + state['phase'] = 'implemented' + save(state_path, state) + + +def gate_commands() -> list[tuple[str, list[str]]]: + return [ + ('pipeline-tests', [sys.executable, '-m', 'unittest', 'discover', '-s', 'scripts/ci', '-p', 'test_*.py']), + ('diff-check', ['git', 'diff', '--check', 'HEAD']), + ('modules', ['go', 'mod', 'verify']), + ('webchat-dependencies', ['pnpm', '--dir', 'webchat', 'install', '--frozen-lockfile']), + ('webchat-tests', ['pnpm', '--dir', 'webchat', 'test']), + ('webchat-build', ['make', 'webchat-rebuild']), + ('docs-build', ['go', 'run', '-mod=readonly', './cmd/build-docs']), + ('vet', ['go', 'vet', '-mod=readonly', './...']), + ('lint', ['make', 'lint']), + ('build', ['go', 'build', '-mod=readonly', './...']), + ('race-tests', ['make', 'test-short']), + ] + + +def validate(repo: Path, state_path: Path, state: dict, timeout: int) -> None: + branch_guard(repo, state) + state.update(phase='validating', gates=[]) + state.pop('validated_fingerprint', None) + state.pop('last_error', None) + save(state_path, state) + before = fingerprint(repo) + for name, args in gate_commands(): + started = time.monotonic() + bounded_command(repo, args, state_path.parent / (name + '.log'), timeout) + state['gates'].append({'name': name, 'status': 'passed', + 'seconds': round(time.monotonic() - started, 2)}) + save(state_path, state) + branch_guard(repo, state) + if fingerprint(repo) != before: + raise DeliveryError('Validation changed source files; inspect changes and validate again') + state.update(phase='validated', validated_fingerprint=before) + save(state_path, state) + + +def require_validated(repo: Path, state: dict) -> None: + branch_guard(repo, state) + if state['phase'] not in ['validated', 'committed', 'pushed', 'published']: + raise DeliveryError('Publication requires all quality gates to pass') + if fingerprint(repo) != state.get('validated_fingerprint'): + raise DeliveryError('Code changed after validation; validate the current contents again') + hooks = capture(repo, ['git', 'config', '--get', 'core.hooksPath']) + if hooks != 'scripts/git-hooks': + raise DeliveryError('Required repository Git hooks are not installed') + for name in ['pre-commit', 'pre-push', 'commit-msg']: + if not os.access(repo / hooks / name, os.X_OK): + raise DeliveryError(f'Required hook is not executable: {name}') + + +def publish(repo: Path, state_path: Path, state: dict, timeout: int) -> None: + require_validated(repo, state) + run_dir = state_path.parent + if state['phase'] == 'validated': + if not capture(repo, ['git', 'status', '--porcelain']): + raise DeliveryError('No changes to deliver') + validated_contents = fingerprint(repo, include_head=False) + bounded_command(repo, ['git', 'add', '--all'], run_dir / 'stage.log', 60) + subject = re.sub(r'[\r\n]+', ' ', state['title'])[:100] + bounded_command(repo, ['git', 'commit', '-m', f'feat: {subject} (#{state["issue"]})'], + run_dir / 'commit.log', timeout) + if capture(repo, ['git', 'status', '--porcelain']): + raise DeliveryError('Commit hook left changes; validate again before publication') + if fingerprint(repo, include_head=False) != validated_contents: + raise DeliveryError('Commit changed validated contents; validate again before publication') + state.update(phase='committed', validated_fingerprint=fingerprint(repo), + commit=capture(repo, ['git', 'rev-parse', 'HEAD'])) + save(state_path, state) + if state['phase'] == 'committed': + bounded_command(repo, ['git', 'push', '--set-upstream', 'origin', state['branch']], + run_dir / 'push.log', timeout) + state['phase'] = 'pushed' + save(state_path, state) + # Resume safely if PR creation succeeded but the client lost its response. + prs = json.loads(capture(repo, ['gh', 'pr', 'list', '--head', state['branch'], + '--base', state['base'], '--state', 'open', '--json', 'url'])) + if prs: + url = prs[0]['url'] + else: + summary = run_dir / 'implementation-summary.txt' + description = summary.read_text().strip() if summary.exists() else state['title'] + body = (f'Refs #{state["issue"]}\n\n{description}\n\n' + 'Validation passed on the delivered contents:\n' + + '\n'.join('- ' + item['name'] for item in state['gates']) + + '\n\nDraft PR; requires review and GitHub CI before merge.\n') + (run_dir / 'pr-body.md').write_text(body) + url = capture(repo, ['gh', 'pr', 'create', '--draft', '--base', state['base'], + '--head', state['branch'], '--title', state['title'], + '--body-file', str(run_dir / 'pr-body.md')]) + state.update(phase='published', pr_url=url) + save(state_path, state) + print(url) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('command', choices=['prepare', 'implement', 'validate', 'publish', 'run', 'status']) + parser.add_argument('--repo', type=Path, default=Path.cwd()) + parser.add_argument('--issue', type=int) + parser.add_argument('--task-file', type=Path) + parser.add_argument('--state', type=Path) + parser.add_argument('--runs-dir', type=Path, + default=Path.home() / '.local/state/hotplex-delivery') + parser.add_argument('--timeout', type=int, default=1800, help='Timeout per command, seconds') + args = parser.parse_args(argv) + state_path = args.state.resolve() if args.state else None + state = None + try: + if args.timeout <= 0: + raise DeliveryError('Timeout must be positive') + if args.command in ['prepare', 'run']: + if not args.issue or args.issue <= 0: + raise DeliveryError('--issue must identify an open GitHub issue') + repo = args.repo.resolve() + else: + if state_path is None: + raise DeliveryError('--state is required') + state = json.loads(state_path.read_text()) + repo = Path(state['repo']) + if args.command == 'status': + print(json.dumps(state, indent=2)) + return 0 + with checkout_lock(repo): + if args.command in ['prepare', 'run']: + state_path = prepare(repo, args.runs_dir.resolve(), args.issue, args.task_file) + print('State: ' + str(state_path), flush=True) + state = json.loads(state_path.read_text()) + if args.command in ['implement', 'run']: + implement(repo, state_path, state, args.timeout) + if args.command in ['validate', 'run']: + validate(repo, state_path, state, args.timeout) + if args.command in ['publish', 'run']: + publish(repo, state_path, state, args.timeout) + return 0 + except (DeliveryError, OSError, ValueError, subprocess.SubprocessError) as error: + if state_path and state: + state['last_error'] = str(error) + # Keep committed/pushed checkpoints for retry; never certify failed gates. + if state['phase'] in ['implementing', 'validating']: + state['phase'] = 'failed' + save(state_path, state) + print(str(error), file=sys.stderr) + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/scripts/ci/test_cloud_delivery.py b/scripts/ci/test_cloud_delivery.py new file mode 100644 index 000000000..d48067481 --- /dev/null +++ b/scripts/ci/test_cloud_delivery.py @@ -0,0 +1,210 @@ +"""Exercise delivery boundaries against real disposable Git repositories.""" + +import contextlib +import importlib.util +import io +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + + +spec = importlib.util.spec_from_file_location('cloud_delivery', Path(__file__).with_name('cloud_delivery.py')) +delivery = importlib.util.module_from_spec(spec) +spec.loader.exec_module(delivery) + + +class DeliveryTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.repo = Path(self.temp.name) / 'repo' + self.repo.mkdir() + self.git('init', '-b', 'main') + self.git('config', 'user.email', 'test@example.invalid') + self.git('config', 'user.name', 'Delivery Test') + self.git('remote', 'add', 'origin', 'https://example.invalid/repo.git') + (self.repo / 'source.txt').write_text('original\n') + self.git('add', '.') + self.git('commit', '-m', 'test: initial') + self.git('switch', '-c', 'codex/delivery-1-test') + self.state = {'phase': 'implemented', 'repo': str(self.repo), 'issue': 1, + 'branch': 'codex/delivery-1-test', 'base': 'main', + 'base_sha': self.git('rev-parse', 'HEAD'), 'title': 'Test task', + 'origin': 'https://example.invalid/repo.git', 'gates': []} + self.run = Path(self.temp.name) / 'run' + self.run.mkdir() + self.path = self.run / 'state.json' + delivery.save(self.path, self.state) + + def git(self, *args): + return subprocess.check_output(['git', *args], cwd=self.repo, stderr=subprocess.DEVNULL, text=True).strip() + + def install_hooks(self): + hooks = self.repo / 'scripts/git-hooks' + hooks.mkdir(parents=True) + for name in ['pre-commit', 'pre-push', 'commit-msg']: + p = hooks / name + p.write_text('#!/bin/sh\nexit 0\n') + p.chmod(0o755) + self.git('config', 'core.hooksPath', 'scripts/git-hooks') + + def certify(self): + self.state.update(phase='validated', validated_fingerprint=delivery.fingerprint(self.repo)) + delivery.save(self.path, self.state) + + def test_dirty_preparation_never_contacts_github_or_switches_branch(self): + (self.repo / 'new.txt').write_text('preserve me') + with patch.object(delivery.shutil, 'which') as which: + with self.assertRaisesRegex(delivery.DeliveryError, 'clean checkout'): + delivery.prepare(self.repo, self.run, 1, None) + which.assert_not_called() + self.assertEqual(self.git('branch', '--show-current'), self.state['branch']) + + def test_fingerprint_detects_untracked_deleted_and_mode_changes(self): + before = delivery.fingerprint(self.repo) + path = self.repo / 'source.txt' + original_mode = path.stat().st_mode & 0o777 + path.chmod(0o755) + self.assertNotEqual(before, delivery.fingerprint(self.repo)) + path.chmod(original_mode) + self.assertEqual(before, delivery.fingerprint(self.repo)) + extra = self.repo / 'extra.txt' + extra.write_text('new') + self.assertNotEqual(before, delivery.fingerprint(self.repo)) + extra.unlink() + path.unlink() + self.assertNotEqual(before, delivery.fingerprint(self.repo)) + + def test_failed_gate_cannot_publish_and_later_gates_do_not_run(self): + commands = [('first', ['false']), ('later', ['true'])] + with patch.object(delivery, 'gate_commands', return_value=commands): + with patch.object(delivery, 'bounded_command', side_effect=delivery.DeliveryError('failure')) as command: + with self.assertRaises(delivery.DeliveryError): + delivery.validate(self.repo, self.path, self.state, 2) + self.assertEqual(command.call_count, 1) + self.assertNotIn('validated_fingerprint', self.state) + with patch.object(delivery, 'bounded_command') as publish: + with self.assertRaisesRegex(delivery.DeliveryError, 'quality gates'): + delivery.publish(self.repo, self.path, self.state, 2) + publish.assert_not_called() + + def test_mutation_during_validation_requires_revalidation(self): + def mutate(*args, **kwargs): + (self.repo / 'source.txt').write_text('changed by gate') + with patch.object(delivery, 'gate_commands', return_value=[('mutating', ['true'])]): + with patch.object(delivery, 'bounded_command', side_effect=mutate): + with self.assertRaisesRegex(delivery.DeliveryError, 'changed source'): + delivery.validate(self.repo, self.path, self.state, 2) + self.assertNotIn('validated_fingerprint', self.state) + + def test_changed_code_after_validation_never_publishes(self): + self.install_hooks() + self.certify() + (self.repo / 'source.txt').write_text('late edit') + with patch.object(delivery, 'bounded_command') as command: + with self.assertRaisesRegex(delivery.DeliveryError, 'changed after validation'): + delivery.publish(self.repo, self.path, self.state, 2) + command.assert_not_called() + + def test_wrong_branch_and_remote_are_rejected(self): + self.git('switch', 'main') + with self.assertRaisesRegex(delivery.DeliveryError, 'branch differs'): + delivery.branch_guard(self.repo, self.state) + self.git('switch', self.state['branch']) + self.git('remote', 'set-url', 'origin', 'https://example.invalid/other.git') + with self.assertRaisesRegex(delivery.DeliveryError, 'Origin changed'): + delivery.branch_guard(self.repo, self.state) + + def test_missing_hooks_block_publication(self): + self.certify() + with self.assertRaises(delivery.DeliveryError): + delivery.require_validated(self.repo, self.state) + + def test_second_runner_cannot_lock_checkout(self): + with delivery.checkout_lock(self.repo): + with self.assertRaisesRegex(delivery.DeliveryError, 'Another delivery'): + with delivery.checkout_lock(self.repo): + self.fail('Lock unexpectedly acquired') + + def test_successful_validation_is_recorded_and_can_be_rechecked(self): + self.install_hooks() + with patch.object(delivery, 'gate_commands', return_value=[('gate', ['true'])]): + with patch.object(delivery, 'bounded_command'): + delivery.validate(self.repo, self.path, self.state, 2) + delivery.require_validated(self.repo, self.state) + recorded = json.loads(self.path.read_text()) + self.assertEqual(recorded['phase'], 'validated') + self.assertEqual(recorded['gates'][0]['status'], 'passed') + + def test_published_resume_reuses_existing_pr_without_push(self): + self.install_hooks() + self.certify() + self.state['phase'] = 'published' + real_capture = delivery.capture + def fake_capture(repo, args): + if args[:3] == ['gh', 'pr', 'list']: + return '[{"url":"https://example.invalid/pr/1"}]' + return real_capture(repo, args) + with patch.object(delivery, 'capture', side_effect=fake_capture): + with patch.object(delivery, 'bounded_command') as command: + with contextlib.redirect_stdout(io.StringIO()): + delivery.publish(self.repo, self.path, self.state, 2) + command.assert_not_called() + self.assertEqual(self.state['pr_url'], 'https://example.invalid/pr/1') + + def test_real_commit_and_push_create_draft_pr_with_issue_and_body_file(self): + self.install_hooks() + remote = Path(self.temp.name) / 'remote.git' + subprocess.run(['git', 'init', '--bare', str(remote)], check=True, capture_output=True) + self.git('remote', 'set-url', 'origin', str(remote)) + self.state['origin'] = str(remote) + self.certify() + real_capture = delivery.capture + created = [] + def fake_capture(repo, args): + if args[:3] == ['gh', 'pr', 'list']: + return '[]' + if args[:3] == ['gh', 'pr', 'create']: + created.append(args) + return 'https://example.invalid/pr/2' + return real_capture(repo, args) + with patch.object(delivery, 'capture', side_effect=fake_capture): + with contextlib.redirect_stdout(io.StringIO()): + delivery.publish(self.repo, self.path, self.state, 10) + self.assertEqual(self.state['phase'], 'published') + self.assertEqual(self.git('status', '--porcelain'), '') + remote_sha = subprocess.check_output( + ['git', '--git-dir', str(remote), 'rev-parse', 'refs/heads/' + self.state['branch']], text=True + ).strip() + self.assertEqual(remote_sha, self.state['commit']) + self.assertIn('--draft', created[0]) + self.assertIn('--body-file', created[0]) + self.assertIn('Refs #1', (self.run / 'pr-body.md').read_text()) + + def test_commit_hook_mutation_cannot_reach_push(self): + self.install_hooks() + hook = self.repo / 'scripts/git-hooks/pre-commit' + hook.write_text('#!/bin/sh\nprintf changed > source.txt\ngit add source.txt\n') + hook.chmod(0o755) + self.certify() + with contextlib.redirect_stdout(io.StringIO()): + with self.assertRaisesRegex(delivery.DeliveryError, 'changed validated contents'): + delivery.publish(self.repo, self.path, self.state, 10) + self.assertFalse((self.run / 'push.log').exists()) + + def test_timeout_stops_hung_command_and_preserves_log(self): + with contextlib.redirect_stdout(io.StringIO()): + with self.assertRaisesRegex(delivery.DeliveryError, 'timed out'): + delivery.bounded_command( + self.repo, [sys.executable, '-c', 'import signal; signal.pause()'], + self.run / 'timeout.log', 1, + ) + self.assertTrue((self.run / 'timeout.log').is_file()) + + +if __name__ == '__main__': + unittest.main() From 2a229623e6a5c809fb741465be81fd23dfbb8c5d Mon Sep 17 00:00:00 2001 From: Aaron Wong Date: Mon, 5 Oct 2026 11:54:39 +0000 Subject: [PATCH 2/2] fix: address shutdown reentrancy and delivery recovery review findings --- internal/execution/repairer.go | 73 +++++++++++++++++------------ internal/execution/repairer_test.go | 27 +++++++++++ scripts/ci/cloud_delivery.py | 10 +++- scripts/ci/test_cloud_delivery.py | 20 ++++++++ 4 files changed, 98 insertions(+), 32 deletions(-) diff --git a/internal/execution/repairer.go b/internal/execution/repairer.go index 32fead72e..4f68e323f 100644 --- a/internal/execution/repairer.go +++ b/internal/execution/repairer.go @@ -77,11 +77,12 @@ type Repairer struct { timedOut int64 dropped int64 - stopCh chan struct{} - wg sync.WaitGroup - start sync.Once - stop sync.Once - closed atomic.Bool + stopCh chan struct{} + shutdownDone chan struct{} + wg sync.WaitGroup + start sync.Once + stop sync.Once + closed atomic.Bool hookMu sync.RWMutex successHook func(RepairIntent) @@ -95,12 +96,13 @@ func NewRepairer(store Store, cfg RepairConfig, log *slog.Logger) *Repairer { cfg.QueueCapacity = 2 } return &Repairer{ - store: store, - cfg: cfg, - log: log.With("component", "repairer"), - intents: make(map[string]*RepairIntent), - abandoned: make(map[string]struct{}), - stopCh: make(chan struct{}), + store: store, + cfg: cfg, + log: log.With("component", "repairer"), + intents: make(map[string]*RepairIntent), + abandoned: make(map[string]struct{}), + stopCh: make(chan struct{}), + shutdownDone: make(chan struct{}), } } @@ -321,6 +323,8 @@ func (r *Repairer) processIntent(ctx context.Context, intent *RepairIntent) erro } func (r *Repairer) Shutdown(ctx context.Context) { + shutdownCtx, cancel := context.WithTimeout(ctx, r.cfg.ShutdownTimeout) + defer cancel() r.stop.Do(func() { // Serialize Start's WaitGroup registration with shutdown. Closing before // waiting also prevents a late Start from adding another processing loop. @@ -329,27 +333,34 @@ func (r *Repairer) Shutdown(ctx context.Context) { close(r.stopCh) r.mu.Unlock() - shutdownCtx, cancel := context.WithTimeout(ctx, r.cfg.ShutdownTimeout) - defer cancel() - done := make(chan struct{}) - go func() { - r.wg.Wait() - close(done) - }() - - // Intent retry fields belong to the processing loop. Drain only after - // that loop has released ownership; a timeout must not start a second - // processor against an in-flight intent. Once also serializes shutdowns. - select { - case <-done: - r.drain(shutdownCtx) - case <-shutdownCtx.Done(): - } - if shutdownCtx.Err() != nil && r.Backlog() > 0 { - r.log.Warn("repairer shutdown timed out", - "backlog", r.Backlog(), "timeout", r.cfg.ShutdownTimeout) - } + // Once starts one owner, but never holds its lock during callbacks. + go r.finishShutdown(shutdownCtx) }) + select { + case <-r.shutdownDone: + case <-shutdownCtx.Done(): + } +} + +func (r *Repairer) finishShutdown(ctx context.Context) { + defer close(r.shutdownDone) + done := make(chan struct{}) + go func() { + r.wg.Wait() + close(done) + }() + + // Intent retry fields belong to the processing loop. Drain only after + // that loop releases ownership; timeout never starts a second processor. + select { + case <-done: + r.drain(ctx) + case <-ctx.Done(): + } + if ctx.Err() != nil && r.Backlog() > 0 { + r.log.Warn("repairer shutdown timed out", + "backlog", r.Backlog(), "timeout", r.cfg.ShutdownTimeout) + } } func (r *Repairer) drain(ctx context.Context) { diff --git a/internal/execution/repairer_test.go b/internal/execution/repairer_test.go index f93a3bcba..e49abc84e 100644 --- a/internal/execution/repairer_test.go +++ b/internal/execution/repairer_test.go @@ -35,6 +35,33 @@ type shutdownBlockingStore struct { calls atomic.Int64 } +type shutdownHookStore struct{ Store } + +func (s *shutdownHookStore) SetDelivery(context.Context, string, string, Status, string) error { + return nil +} + +func TestRepairer_ShutdownHookCanCallShutdownWithCancelledContext(t *testing.T) { + t.Parallel() + r := NewRepairer(&shutdownHookStore{}, fastRepairConfig(), nil) + r.Enqueue(RepairIntent{ExecutionID: "hook", Kind: RepairDelivery, Status: string(StatusDelivered)}) + r.SetSuccessHook(func(RepairIntent) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + r.Shutdown(ctx) + }) + done := make(chan struct{}) + go func() { + r.Shutdown(context.Background()) + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("success callback deadlocked while re-entering shutdown") + } +} + func (s *shutdownBlockingStore) SetDelivery(ctx context.Context, _, _ string, _ Status, _ string) error { if s.calls.Add(1) != 1 { return errors.New("concurrent retry while the first attempt is blocked") diff --git a/scripts/ci/cloud_delivery.py b/scripts/ci/cloud_delivery.py index 09596d517..67cb69b88 100644 --- a/scripts/ci/cloud_delivery.py +++ b/scripts/ci/cloud_delivery.py @@ -222,6 +222,14 @@ def require_validated(repo: Path, state: dict) -> None: def publish(repo: Path, state_path: Path, state: dict, timeout: int) -> None: require_validated(repo, state) run_dir = state_path.parent + if state['phase'] == 'validated' and not capture(repo, ['git', 'status', '--porcelain']): + # A commit hook can leave a clean commit with changed contents. After + # revalidation, resume that commit instead of demanding another edit. + ahead = capture(repo, ['git', 'rev-list', '--count', f'{state["base_sha"]}..HEAD']) + if ahead == '0': + raise DeliveryError('No changes to deliver') + state.update(phase='committed', commit=capture(repo, ['git', 'rev-parse', 'HEAD'])) + save(state_path, state) if state['phase'] == 'validated': if not capture(repo, ['git', 'status', '--porcelain']): raise DeliveryError('No changes to deliver') @@ -244,7 +252,7 @@ def publish(repo: Path, state_path: Path, state: dict, timeout: int) -> None: save(state_path, state) # Resume safely if PR creation succeeded but the client lost its response. prs = json.loads(capture(repo, ['gh', 'pr', 'list', '--head', state['branch'], - '--base', state['base'], '--state', 'open', '--json', 'url'])) + '--base', state['base'], '--state', 'all', '--json', 'url'])) if prs: url = prs[0]['url'] else: diff --git a/scripts/ci/test_cloud_delivery.py b/scripts/ci/test_cloud_delivery.py index d48067481..011f9040a 100644 --- a/scripts/ci/test_cloud_delivery.py +++ b/scripts/ci/test_cloud_delivery.py @@ -196,6 +196,26 @@ def test_commit_hook_mutation_cannot_reach_push(self): delivery.publish(self.repo, self.path, self.state, 10) self.assertFalse((self.run / 'push.log').exists()) + def test_clean_commit_can_be_revalidated_and_resume_publication(self): + self.install_hooks() + self.git('add', '--all') + self.git('commit', '-m', 'test: already committed implementation') + self.certify() + real_capture = delivery.capture + def fake_capture(repo, args): + if args[:3] == ['gh', 'pr', 'list']: + self.assertEqual(args[args.index('--state') + 1], 'all') + return '[{"url":"https://example.invalid/pr/closed"}]' + return real_capture(repo, args) + with patch.object(delivery, 'capture', side_effect=fake_capture): + with patch.object(delivery, 'bounded_command') as command: + with contextlib.redirect_stdout(io.StringIO()): + delivery.publish(self.repo, self.path, self.state, 10) + self.assertEqual(command.call_count, 1) + self.assertEqual(command.call_args.args[1][:2], ['git', 'push']) + self.assertEqual(self.state['phase'], 'published') + self.assertEqual(self.state['pr_url'], 'https://example.invalid/pr/closed') + def test_timeout_stops_hung_command_and_preserves_log(self): with contextlib.redirect_stdout(io.StringIO()): with self.assertRaisesRegex(delivery.DeliveryError, 'timed out'):