feat(jobs): tie a one-off container to its operation, and record an interruption

internal/engine/audit.go

@@ -170,6 +170,12 @@ func auditRows(recs []journal.Record) []auditRow {
 				row.action = auditAction(r.Phase)
 			}
 			switch {
+			// Ahead of the failure arm, which this record also matches: an
+			// interrupted run is not a failure of the job. The client went away
+			// and the outcome is unknown, which is a different thing to tell an
+			// operator than "it failed" or "it never finished".
+			case r.ErrorCode == "interrupted":
+				row.outcome = "interrupted"
 			case r.Event == "abort":
 				row.outcome = "aborted"
 			case r.Status == "fail":

internal/engine/audit_test.go

@@ -129,3 +129,29 @@ func TestAuditReportsFailedJobRun(t *testing.T) {
 		t.Fatalf("failed job run audit = %+v", records)
 	}
 }
+
+// An interrupted run is neither a job failure nor incomplete-forever: the
+// client went away and the outcome is unknown. The record carries Status fail,
+// so it also matches the failure arm and the ordering is what distinguishes it.
+func TestAuditDistinguishesAnInterruptedJobRun(t *testing.T) {
+	f := &transport.Fake{Dynamic: func(cmd string) (transport.Result, bool) {
+		switch {
+		case strings.Contains(cmd, "ls -1"):
+			return transport.Result{Stdout: "job-3.jsonl\n"}, true
+		case strings.Contains(cmd, "job-3.jsonl"):
+			return transport.Result{Stdout: journalLines(
+				journal.Record{DeployID: "job-3", Phase: "job", Event: "start", Status: "ok", OperationKind: "job_run", Service: "catalog-refresh", Operator: "v@mac", TS: "t1"},
+				journal.Record{DeployID: "job-3", Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted", OperationKind: "job_run", Service: "catalog-refresh"},
+			)}, true
+		}
+		return transport.Result{}, false
+	}}
+	e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
+	records, err := e.AuditSnapshot(context.Background(), 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(records) != 1 || records[0].Outcome != "interrupted" || records[0].Service != "catalog-refresh" {
+		t.Fatalf("interrupted job audit = %+v", records)
+	}
+}

internal/engine/gate.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/labstack/onebox/internal/app"
@@ -73,7 +74,7 @@ func (e *Engine) runJobPhase(ctx context.Context, jw *journal.Writer, done map[s
 			return fmt.Errorf("journal %s intent: %w", key, err)
 		}
 		st := e.ui.Step("job "+job, true)
-		safe, detail, err := e.runOneJob(ctx, job, remoteDir, remoteCompose)
+		safe, detail, err := e.runOneJob(ctx, jw.DeployID, jw.Epoch, job, remoteDir, remoteCompose)
 		if err == nil {
 			e.logf("job %s: %s", job, detail)
 		}
@@ -113,7 +114,7 @@ func (e *Engine) runJobPhase(ctx context.Context, jw *journal.Writer, done map[s
 
 // runOneJob runs a single gate step and reports whether it declared itself
 // rollback-safe (changed=false). Returns (safe, detail, err).
-func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose string) (bool, string, error) {
+func (e *Engine) runOneJob(ctx context.Context, operationID string, epoch int, job, remoteDir, remoteCompose string) (bool, string, error) {
 	safeByDeclaration := e.jobDataEffect(job) == app.DataEffectNone
 	if !safeByDeclaration {
 		res, err := e.mutate(ctx, invalidateExecutionCommand(e.names().AppDir()))
@@ -128,7 +129,7 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st
 	resultFile := resultDir + "/result"
 	const containerResultFile = "/run/onebox/job-result"
 	containerized := true
-	runCmd := e.composeCmd(remoteCompose) + " run --rm --no-deps" +
+	runCmd := e.composeCmd(remoteCompose) + " run --rm --no-deps" + jobRunLabels(operationID, epoch) +
 		" -e ONEBOX_RESULT_FILE=" + containerResultFile +
 		" -v " + q(resultFile+":"+containerResultFile+":rw") + " " + job
 	if h, ok := e.Spec.Hooks[job]; ok && h.Run != "" {
@@ -147,6 +148,13 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st
 		var injected bool
 		runCmd, injected = injectComposeJobResult(runCmd, resultFile, containerResultFile)
 		containerized = injected
+		// A hook that runs its own compose command produces a container this
+		// operation owns just as much as the generated one, so it carries the
+		// same identity. A hook that is not a compose run gets no label because
+		// there is no container to put one on; that hook is also not
+		// containerized, so its result is unresolvable and reported as such
+		// below.
+		runCmd, _ = injectComposeJobLabels(runCmd, operationID, epoch)
 	}
 	e.ui.Cmd("job", runCmd) // verbose only — the plan lists it
 	resultMode := "600"
@@ -200,7 +208,43 @@ func (e *Engine) runOneJob(ctx context.Context, job, remoteDir, remoteCompose st
 	return !evidence.Changed, jobResultDetail(evidence), nil
 }
 
+// jobRunLabels ties a one-off container back to the operation that started it.
+// `compose run` names nothing and inherits no operation identity, so without
+// these an interrupted run leaves a container on the host that nothing can
+// match to a journal — every refusal and every reconciliation keys on them.
+func jobRunLabels(operationID string, epoch int) string {
+	if operationID == "" {
+		return ""
+	}
+	return " --label " + q(JobOperationLabel+"="+operationID) +
+		" --label " + q(JobEpochLabel+"="+strconv.Itoa(epoch))
+}
+
+const (
+	// JobOperationLabel carries the operation id of the run that created a
+	// one-off job container.
+	JobOperationLabel = "ob.operation"
+	// JobEpochLabel carries the lock epoch that run held.
+	JobEpochLabel = "ob.epoch"
+)
+
+func injectComposeJobLabels(command, operationID string, epoch int) (string, bool) {
+	labels := jobRunLabels(operationID, epoch)
+	if labels == "" {
+		return command, false
+	}
+	return injectComposeRunFlags(command, labels+" ")
+}
+
 func injectComposeJobResult(command, hostResultFile, containerResultFile string) (string, bool) {
+	return injectComposeRunFlags(command,
+		" -e "+"ONEBOX_RESULT_FILE="+containerResultFile+
+			" -v "+q(hostResultFile+":"+containerResultFile+":rw")+" ")
+}
+
+// injectComposeRunFlags splices flags into a hook's own `docker compose run`.
+// Anything that is not a compose run is left alone and reported as such.
+func injectComposeRunFlags(command, flags string) (string, bool) {
 	runIndex := strings.Index(command, " run ")
 	if runIndex < 0 {
 		return command, false
@@ -209,9 +253,7 @@ func injectComposeJobResult(command, hostResultFile, containerResultFile string)
 	if !strings.Contains(prefix, "docker compose") && !strings.Contains(prefix, "docker-compose") {
 		return command, false
 	}
-	flags := " run -e ONEBOX_RESULT_FILE=" + containerResultFile +
-		" -v " + q(hostResultFile+":"+containerResultFile+":rw") + " "
-	return prefix + flags + command[runIndex+len(" run "):], true
+	return prefix + " run" + flags + command[runIndex+len(" run "):], true
 }
 
 func (e *Engine) unknownJobResult(job, reason string) (bool, string, error) {

internal/engine/gate_test.go

@@ -142,7 +142,7 @@ func TestJobAutoRunsWithoutHook(t *testing.T) {
 		t.Fatalf("deploy: %v", err)
 	}
 	seq := strings.Join(f.Commands, "\n")
-	if !strings.Contains(seq, "run --rm --no-deps -e ONEBOX_RESULT_FILE=/run/onebox/job-result") {
+	if !strings.Contains(seq, "run --rm --no-deps") || !strings.Contains(seq, "-e ONEBOX_RESULT_FILE=/run/onebox/job-result") {
 		t.Fatalf("a job without a hook must auto-run compose run:\n%s", seq)
 	}
 	// gate protocol still applies to the auto-run job.
@@ -266,7 +266,7 @@ func TestUnknownJobMessagesExplainRollbackConsequence(t *testing.T) {
 	t.Run("no result declaration", func(t *testing.T) {
 		f := happyFake()
 		e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
-		safe, detail, err := e.runOneJob(context.Background(), "migrate", "/remote", "/remote/compose.yaml")
+		safe, detail, err := e.runOneJob(context.Background(), "op-1", 1, "migrate", "/remote", "/remote/compose.yaml")
 		if err != nil {
 			t.Fatalf("run job: %v", err)
 		}
@@ -282,7 +282,7 @@ func TestUnknownJobMessagesExplainRollbackConsequence(t *testing.T) {
 		e := New(cfg, testProject(t), happyFake(), Options{
 			Out: &bytes.Buffer{}, Sleep: noSleep, LocalDir: t.TempDir(),
 		})
-		safe, detail, err := e.runOneJob(context.Background(), "migrate", "/remote", "/remote/compose.yaml")
+		safe, detail, err := e.runOneJob(context.Background(), "op-1", 1, "migrate", "/remote", "/remote/compose.yaml")
 		if err != nil {
 			t.Fatalf("run local job: %v", err)
 		}
@@ -419,7 +419,7 @@ func TestMigrateComposeJobGetsPrivateWritableBoundResultFile(t *testing.T) {
 		mount := strings.Index(c, "-v '"+resultFile+":/run/onebox/job-result:rw'")
 		sealedFile := strings.Index(c, "chmod 600 '"+resultFile+"'")
 		if strings.Contains(c, "rm -rf '"+resultDir+"'") &&
-			strings.Contains(c, "run --rm --no-deps -e ONEBOX_RESULT_FILE=/run/onebox/job-result") &&
+			strings.Contains(c, "run --rm --no-deps") && strings.Contains(c, "-e ONEBOX_RESULT_FILE=/run/onebox/job-result") &&
 			privateDir >= 0 && privateDir < writableFile && writableFile < mount && mount < sealedFile {
 			found = true
 		}
@@ -428,3 +428,38 @@ func TestMigrateComposeJobGetsPrivateWritableBoundResultFile(t *testing.T) {
 		t.Fatalf("migrate container must receive a privately staged, writable, subsequently sealed result file:\n%s", strings.Join(f.Commands, "\n"))
 	}
 }
+
+// Without an operation label nothing on the host ties a running one-off
+// container back to the journal that started it, so an interrupted run cannot
+// be refused or reconciled — only guessed at.
+func TestJobContainerCarriesItsOperationIdentity(t *testing.T) {
+	f := happyFake()
+	e := New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
+	if _, _, err := e.runOneJob(context.Background(), "20260909-053225-abc-job_run-deadbeef", 7, "migrate", "/remote", "/remote/compose.yaml"); err != nil {
+		t.Fatal(err)
+	}
+	seq := strings.Join(f.Commands, "\n")
+	for _, want := range []string{
+		"--label 'ob.operation=20260909-053225-abc-job_run-deadbeef'",
+		"--label 'ob.epoch=7'",
+	} {
+		if !strings.Contains(seq, want) {
+			t.Fatalf("job container missing %s:\n%s", want, seq)
+		}
+	}
+}
+
+func TestInjectComposeJobLabelsOnlyTouchesAComposeRun(t *testing.T) {
+	got, ok := injectComposeJobLabels("docker compose -f x.yml run --rm migrate", "op-1", 2)
+	if !ok || !strings.Contains(got, "--label 'ob.operation=op-1'") || !strings.Contains(got, "--label 'ob.epoch=2'") {
+		t.Fatalf("compose run = %q ok=%v", got, ok)
+	}
+	// A hook that is not a compose run has no container to label.
+	if got, ok := injectComposeJobLabels("/usr/local/bin/migrate.sh", "op-1", 2); ok || got != "/usr/local/bin/migrate.sh" {
+		t.Fatalf("non-compose hook = %q ok=%v", got, ok)
+	}
+	// No operation identity, nothing to add.
+	if got, ok := injectComposeJobLabels("docker compose run --rm migrate", "", 0); ok || got != "docker compose run --rm migrate" {
+		t.Fatalf("empty operation = %q ok=%v", got, ok)
+	}
+}

internal/engine/job.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/labstack/onebox/internal/app"
 	"github.com/labstack/onebox/internal/journal"
@@ -43,7 +44,22 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 	if err != nil {
 		return operationID, nil, err
 	}
-	defer e.ReleaseLock(ctx)
+	// Released unless an interrupted run left this operation's container alive.
+	// ReleaseLock runs on its own background context, so on Ctrl-C it succeeds
+	// while the terminal journal append — which uses the cancelled one — does
+	// not: ownership would be dropped, immediately and silently, over a
+	// container still changing data.
+	holdLockForLiveContainer := false
+	defer func() {
+		if holdLockForLiveContainer {
+			e.warnf("operation %s was interrupted while its container is still running; "+
+				"keeping the application lock so nothing else mutates alongside it. "+
+				"Inspect with `docker ps --filter label=%s=%s`; the lock expires on its own after %s",
+				operationID, JobOperationLabel, operationID, e.lockTTL())
+			return
+		}
+		e.ReleaseLock(ctx)
+	}()
 	if err := e.WriteFence(ctx, operationID, epoch); err != nil {
 		return operationID, nil, err
 	}
@@ -93,7 +109,42 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 			record.Status = "fail"
 			record.Detail = runErr.Error()
 		}
-		if journalErr := writer.Append(ctx, record); journalErr != nil {
+		// A cancelled context is exactly when the terminal record matters most,
+		// and exactly when appending on that context cannot work. `ob exec`
+		// already writes its own on a bounded background context; without the
+		// same here an interrupted job stays INCOMPLETE in `ob audit` forever,
+		// with no record that it was ever interrupted. Append redacts Detail on
+		// a failure, so the reason has to ride on ErrorCode.
+		// Two independent questions. WHERE to append: a cancelled caller context
+		// cannot carry the write, whatever the run did, so a job that finished
+		// cleanly a moment before Ctrl-C still records its success. WHAT to
+		// record: only a run that ended because the client went away is
+		// interrupted — an outcome the job itself produced is its own.
+		journalContext := ctx
+		if ctx.Err() != nil {
+			var cancel context.CancelFunc
+			journalContext, cancel = context.WithTimeout(context.Background(), journalCleanupTimeout)
+			defer cancel()
+		}
+		if interruptedRun(ctx, runErr) {
+			record = journal.Record{
+				Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted",
+				OperationKind: "job_run", Service: job,
+			}
+		}
+		journalErr := writer.Append(journalContext, record)
+		if journalErr != nil && journalContext == ctx && ctx.Err() != nil {
+			// Cancellation can land during the write as easily as before it, and
+			// the check above only sees a context that was already gone. Retried
+			// only when the context died in the meantime: any other failure —
+			// a full disk, a refused write — may have landed on the host after
+			// reporting an error, and appending a second terminal record is
+			// worse than reporting the first failure.
+			retryContext, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout)
+			defer cancel()
+			journalErr = writer.Append(retryContext, record)
+		}
+		if journalErr != nil {
 			return errors.Join(runErr, fmt.Errorf("journal job finish: %w", journalErr))
 		}
 		return runErr
@@ -121,10 +172,52 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 	e.gateOpen = true
 	e.rollbackCovered = true
 	runErr := e.runJobPhase(ctx, writer, nil, remoteDir, remoteCompose, "job", []string{job})
+	if interruptedRun(ctx, runErr) {
+		// Cancelling the client kills at most the wrapper shell; the container
+		// belongs to the daemon and keeps running.
+		holdLockForLiveContainer = e.jobContainerRunning(operationID)
+	}
 	var result *journal.JobResultEvidence
 	if evidence, ok := e.jobResults[job]; ok {
 		resultCopy := evidence
 		result = &resultCopy
 	}
 	return operationID, result, finish(runErr)
 }
+
+const journalCleanupTimeout = 5 * time.Second
+
+// interruptedRun reports a run that ended because the client went away rather
+// than because the job finished.
+//
+// A run that produced no error finished, whatever became of the client
+// afterwards — its outcome is its own and must be recorded as such. Only once
+// the run failed does a gone context mean the client is why. The transport does
+// not always surface a cancelled context as context.Canceled, so a failure
+// under a cancelled context counts even when the error says something else.
+func interruptedRun(ctx context.Context, runErr error) bool {
+	if runErr == nil {
+		return false
+	}
+	return ctx.Err() != nil ||
+		errors.Is(runErr, context.Canceled) ||
+		errors.Is(runErr, context.DeadlineExceeded)
+}
+
+// jobContainerRunning answers whether this operation's one-off container is
+// still alive, on a context of its own because the caller's is already gone.
+// An unreadable answer is reported as running: keeping the lock over a
+// container that has in fact exited costs an operator one `--break-lock`, while
+// releasing it over one that has not costs them concurrent writers.
+func (e *Engine) jobContainerRunning(operationID string) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), journalCleanupTimeout)
+	defer cancel()
+	res, err := e.T.Run(ctx, "docker ps -q --filter label="+q(JobOperationLabel+"="+operationID))
+	if err != nil {
+		return true
+	}
+	if res.ExitCode != 0 {
+		return true
+	}
+	return strings.TrimSpace(res.Stdout) != ""
+}

internal/engine/job_test.go

@@ -3,6 +3,7 @@ package engine
 import (
 	"bytes"
 	"context"
+	"errors"
 	"strings"
 	"testing"
 
@@ -151,3 +152,23 @@ func TestRunJobUsesPlanIdentityAndJournalsAuthorization(t *testing.T) {
 		}
 	}
 }
+
+// A job that finished cleanly is not interrupted, whatever became of the client
+// in the window before its terminal record. Classification follows the run;
+// only the choice of append context follows the caller's context.
+func TestInterruptedRunClassifiesTheRunNotTheClient(t *testing.T) {
+	cancelled, cancel := context.WithCancel(context.Background())
+	cancel()
+	if interruptedRun(cancelled, nil) {
+		t.Fatal("a run that produced no error must never be recorded interrupted")
+	}
+	if !interruptedRun(cancelled, errors.New("ssh: session closed")) {
+		t.Fatal("a failed run under a cancelled context is interrupted, whatever the transport called it")
+	}
+	if !interruptedRun(context.Background(), context.Canceled) {
+		t.Fatal("a cancellation surfaced by the run itself is interrupted")
+	}
+	if interruptedRun(context.Background(), errors.New("migrate: exit 1")) {
+		t.Fatal("a job that failed on its own terms is not interrupted")
+	}
+}

internal/engine/schedule_execution_test.go

@@ -150,7 +150,7 @@ func TestDurableCompatibilityInvalidationMustSucceedBeforeDataChangingJob(t *tes
 				return base(command)
 			}
 			e := New(cfg, testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
-			_, _, err := e.runOneJob(context.Background(), "change", "/release", "/release/compose.yaml")
+			_, _, err := e.runOneJob(context.Background(), "op-1", 1, "change", "/release", "/release/compose.yaml")
 			if !invalidated || err == nil || !strings.Contains(err.Error(), "invalidate durable execution compatibility") {
 				t.Fatalf("failed invalidation did not stop data-changing job: invalidated=%t err=%v", invalidated, err)
 			}