Fix Repairer shutdown races and add cloud delivery pipeline

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,17 +96,23 @@ 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{}),
 	}
 }
 
 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)
 	})
@@ -316,36 +323,47 @@ 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.
+		r.mu.Lock()
+		r.closed.Store(true)
 		close(r.stopCh)
+		r.mu.Unlock()
 
-		done := make(chan struct{})
-		go func() {
-			r.wg.Wait()
-			close(done)
-		}()
-
-		timer := time.NewTimer(r.cfg.ShutdownTimeout)
-		defer timer.Stop()
-		select {
-		case <-done:
-			// The loop owns mutable intent state until it exits. Draining
-			// earlier would process the same intent concurrently.
-			r.drain(ctx)
-		case <-ctx.Done():
-			r.log.Warn("repairer shutdown cancelled",
-				"backlog", r.Backlog(), "err", ctx.Err())
-		case <-timer.C:
-			r.log.Warn("repairer shutdown timed out",
-				"backlog", r.Backlog(), "timeout", r.cfg.ShutdownTimeout)
-		}
-		r.closed.Store(true)
+		// 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) {
-	ctx, cancel := context.WithTimeout(ctx, r.cfg.ShutdownTimeout)
-	defer cancel()
 	ticker := time.NewTicker(20 * time.Millisecond)
 	defer ticker.Stop()
 	for ctx.Err() == nil {

internal/execution/repairer_test.go

@@ -28,6 +28,129 @@ type blockingRuntimeStore struct {
 	once    sync.Once
 }
 
+type shutdownBlockingStore struct {
+	Store
+	started chan struct{}
+	release chan 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")
+	}
+	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() {

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:

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.