diff --git a/internal/engine/deploy.go b/internal/engine/deploy.go
index 7274aa4..e67078c 100644
--- a/internal/engine/deploy.go
+++ b/internal/engine/deploy.go
@@ -81,6 +81,13 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri
 	}
 	stopHB := e.StartHeartbeat(ctx)
 	defer stopHB()
+	// A deploy rolls workloads and runs its own gate jobs; a job container still
+	// changing data underneath it is exactly the overlap the lock exists to
+	// prevent, and the lock alone does not catch it once its holder is gone.
+	// Read-only, so it runs before the plan-binding boundary below.
+	if err := e.refuseForeignJobContainers(ctx, releaseID, epoch); err != nil {
+		return err
+	}
 	// The plan binding is the mutation boundary. Check it under the application
 	// lock before converging even host-scoped support components; a stale plan
 	// must leave both the application and proxy untouched.
@@ -89,6 +96,16 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri
 			return fmt.Errorf("deploy precondition under lock: %w", err)
 		}
 	}
+	// Past the plan boundary, so a stale plan leaves the host untouched. One
+	// read of the journals serves both this and the rollback-debt scan below;
+	// they are the same bytes off a possibly high-latency host.
+	journalIDs, journalsByID, err := journal.Journals(ctx, e.T, e.names())
+	if err != nil {
+		return err
+	}
+	if err := e.closeInterruptedJobRuns(ctx, journalIDs, journalsByID); err != nil {
+		return err
+	}
 	pf := e.ui.Step("preflight", false)
 	if err := e.preflight(ctx, false); err != nil {
 		pf(err)
@@ -107,7 +124,7 @@ func (e *Engine) deployCore(ctx context.Context, releaseID, localStagingDir stri
 	}
 	rollbackDebt := false
 	if done == nil {
-		rollbackDebt, err = e.rollbackEffectDebt(ctx, prev)
+		rollbackDebt, err = e.rollbackEffectDebt(prev, journalIDs, journalsByID)
 		if err != nil {
 			return fmt.Errorf("rollback effect history: %w", err)
 		}
@@ -214,11 +231,7 @@ func (e *Engine) pinnedScheduleDeployConflict() string {
 // failed deploy can mutate data even though its runner exits cleanly and writes
 // finish:fail; a later successful activation/current release or an explicit
 // abort clears that historical debt.
-func (e *Engine) rollbackEffectDebt(ctx context.Context, current string) (bool, error) {
-	ids, byID, err := journal.Journals(ctx, e.T, e.names())
-	if err != nil {
-		return false, err
-	}
+func (e *Engine) rollbackEffectDebt(current string, ids []string, byID map[string][]journal.Record) (bool, error) {
 	debt := false
 	for _, id := range ids {
 		summary := journal.Summarize(byID[id])
diff --git a/internal/engine/job.go b/internal/engine/job.go
index 8ffd5de..3f99758 100644
--- a/internal/engine/job.go
+++ b/internal/engine/job.go
@@ -66,6 +66,13 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 	stopHeartbeat := e.StartHeartbeat(ctx)
 	defer stopHeartbeat()
 
+	// Under the lock, before anything mutates and before any host write: a job
+	// container from an earlier operation may still be running with no process
+	// owning it. Read-only, so it is safe on this side of the plan boundary.
+	if err := e.refuseForeignJobContainers(ctx, operationID, epoch); err != nil {
+		return operationID, nil, err
+	}
+
 	current, err := release.Current(ctx, e.T, e.names())
 	if err != nil {
 		return operationID, nil, err
@@ -86,6 +93,18 @@ func (e *Engine) RunJobWithJournalID(ctx context.Context, request JobRunRequest)
 		return operationID, nil, errors.New("job plan is stale: current release runtime changed — re-plan")
 	}
 
+	// Before this operation writes its own start record, and after the
+	// staleness checks above. Reconciling later would find this run's start
+	// with no finish yet and close the very run about to execute; reconciling
+	// earlier would let a plan that will not execute write to the host.
+	journalIDs, journalsByID, err := journal.Journals(ctx, e.T, e.names())
+	if err != nil {
+		return operationID, nil, err
+	}
+	if err := e.closeInterruptedJobRuns(ctx, journalIDs, journalsByID); err != nil {
+		return operationID, nil, err
+	}
+
 	writer := &journal.Writer{
 		T: e.T, Names: e.names(), DeployID: operationID, Epoch: epoch,
 		Operator: journal.DefaultOperator(), GitSHA: e.Opts.GitSHA, ConfigHash: e.Opts.ConfigHash,
diff --git a/internal/engine/job_reconcile.go b/internal/engine/job_reconcile.go
new file mode 100644
index 0000000..e865dce
--- /dev/null
+++ b/internal/engine/job_reconcile.go
@@ -0,0 +1,211 @@
+package engine
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/labstack/onebox/internal/journal"
+)
+
+// refuseForeignJobContainers refuses when a one-off job container from another
+// operation is still running on the host.
+//
+// The question "is a job still running" is asked of Docker, not of the journal.
+// A journal says what a client managed to write, and the whole failure mode
+// here is a client that did not write. An interrupted run that DID record its
+// interruption looks finished on paper while its container keeps changing data,
+// and a plan re-run appends a second invocation to the same journal — both are
+// invisible to any reduction over records, and both are one `docker ps` away.
+func (e *Engine) refuseForeignJobContainers(ctx context.Context, currentOperationID string, currentEpoch int) error {
+	containers, err := e.jobContainers(ctx)
+	if err != nil {
+		return err
+	}
+	currentEpochLabel := strconv.Itoa(currentEpoch)
+	for _, c := range containers {
+		// Operation AND epoch. A sealed job plan is re-runnable and carries one
+		// operation id for its whole life, and AcquireLock hands the lock
+		// straight back to a caller presenting the id already written in it. So
+		// a second run of one plan would reclaim the lock from a live first run
+		// and then exempt that run's container as its own — two concurrent
+		// data-changing containers, which is the single thing this prevents.
+		if c.operation == currentOperationID && c.epoch == currentEpochLabel {
+			continue
+		}
+		if c.operation == currentOperationID {
+			return fmt.Errorf(
+				"an earlier run of operation %s (epoch %s) left a job container running on this host (%.12s); "+
+					"wait for it to finish, or establish what it did and stop it with `docker rm -f %.12s`",
+				c.operation, labelOrUnknown(c.epoch), c.id, c.id)
+		}
+		if c.operation == "" {
+			// The label is present but carries no value, so the container
+			// cannot be attributed. Refuse anyway: an unattributable job
+			// container is exactly as dangerous as an attributable one.
+			return fmt.Errorf(
+				"a job container is running on this host (%.12s) with an empty %s label, so the operation that "+
+					"started it cannot be identified; establish what it did and stop it with `docker rm -f %.12s`",
+				c.id, JobOperationLabel, c.id)
+		}
+		return fmt.Errorf(
+			"a job container from operation %s is still running on this host (%.12s) with no process owning it; "+
+				"wait for it to finish, or establish what it did and stop it with `docker rm -f %.12s`",
+			c.operation, c.id, c.id)
+	}
+	return nil
+}
+
+type jobContainer struct {
+	id        string
+	operation string
+	epoch     string
+}
+
+func labelOrUnknown(value string) string {
+	if value == "" {
+		return "unknown"
+	}
+	return value
+}
+
+// jobContainers lists every running one-off job container, whichever operation
+// created it. The label is unvalued in the filter so this finds containers of
+// operations this process knows nothing about, which is the point.
+func (e *Engine) jobContainers(ctx context.Context) ([]jobContainer, error) {
+	res, err := e.T.Run(ctx,
+		"docker ps --filter label="+q(JobOperationLabel)+
+			" --format "+q("{{.ID}} {{.Label \""+JobOperationLabel+"\"}} {{.Label \""+JobEpochLabel+"\"}}"))
+	if err != nil {
+		return nil, err
+	}
+	if res.ExitCode != 0 {
+		return nil, fmt.Errorf("list running job containers (exit %d): %s", res.ExitCode, strings.TrimSpace(res.Stderr))
+	}
+	var out []jobContainer
+	for _, line := range strings.Split(res.Stdout, "\n") {
+		// Not TrimSpace before the cut: a container whose label carries no value
+		// prints "<id> " with nothing after the separator, and trimming the line
+		// first removes the separator itself — the container would then be
+		// skipped as unparseable, which is precisely the one that most needs
+		// refusing.
+		id, rest, _ := strings.Cut(strings.TrimRight(line, "\r\n"), " ")
+		operation, epoch, _ := strings.Cut(rest, " ")
+		id, operation, epoch = strings.TrimSpace(id), strings.TrimSpace(operation), strings.TrimSpace(epoch)
+		if id == "" {
+			continue
+		}
+		if !validID.MatchString(id) {
+			return nil, fmt.Errorf("suspicious container id %q from docker ps — refusing to reuse in a command", id)
+		}
+		out = append(out, jobContainer{id: id, operation: operation, epoch: epoch})
+	}
+	return out, nil
+}
+
+// closeInterruptedJobRuns writes the terminal record an interrupted client
+// could not, so an operation stops being incomplete forever.
+//
+// Housekeeping, not safety: the refusal above is what protects a live
+// container, and this runs only once nothing of the sort is running. Being
+// wrong here is therefore cheap, which is why a journal reduction is good
+// enough for it and is not good enough for the refusal.
+// The caller supplies the journals so a deploy, which also scans them for
+// rollback debt, reads them once.
+func (e *Engine) closeInterruptedJobRuns(ctx context.Context, ids []string, byID map[string][]journal.Record) error {
+	for _, id := range ids {
+		for _, run := range unfinishedJobRuns(byID[id]) {
+			if err := e.closeJobRun(ctx, id, run); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// unfinishedJobRun is one invocation that never reached a terminal record.
+type unfinishedJobRun struct {
+	Epoch int
+	Job   string
+	// ResultRecorded is true when the client journaled the job's own result
+	// before it went away. That is the only proof of the outcome that survives:
+	// the container is `--rm`, so nothing about it outlives its exit. A
+	// recorded failure is evidence exactly as much as a recorded success — only
+	// an absent result means the outcome is unknown.
+	ResultRecorded bool
+	ResultOK       bool
+}
+
+// unfinishedJobRuns groups a journal by epoch, because a journal holds one
+// invocation per epoch and a plan may legitimately be run more than once. A
+// finish in an earlier epoch says nothing about a later one.
+func unfinishedJobRuns(records []journal.Record) []unfinishedJobRun {
+	type state struct {
+		started, finished bool
+		run               unfinishedJobRun
+	}
+	byEpoch, order := map[int]*state{}, []int{}
+	for _, r := range records {
+		if r.Phase != "job" {
+			continue
+		}
+		s, seen := byEpoch[r.Epoch]
+		if !seen {
+			s = &state{run: unfinishedJobRun{Epoch: r.Epoch}}
+			byEpoch[r.Epoch], order = s, append(order, r.Epoch)
+		}
+		switch {
+		case r.Event == "start" && r.OperationKind == "job_run":
+			s.started, s.run.Job = true, r.Service
+		case r.Event == "finish":
+			s.finished = true
+		case r.Event == "result" && strings.HasPrefix(r.SubStep, "job:"):
+			// The result record is written by the shared job phase and carries
+			// no operation kind, so it is matched on its own shape.
+			s.run.ResultRecorded = true
+			s.run.ResultOK = r.Status == "ok"
+		}
+	}
+	var out []unfinishedJobRun
+	for _, epoch := range order {
+		if s := byEpoch[epoch]; s.started && !s.finished {
+			out = append(out, s.run)
+		}
+	}
+	return out
+}
+
+// closeJobRun records the outcome, never inventing one. Only a result the
+// client itself journaled proves the job succeeded; anything else is recorded
+// interrupted, which is what an unknown outcome actually is.
+func (e *Engine) closeJobRun(ctx context.Context, operationID string, run unfinishedJobRun) error {
+	record := journal.Record{
+		Phase: "job", Event: "finish", Status: "fail", ErrorCode: "interrupted",
+		OperationKind: "job_run", Service: run.Job,
+	}
+	if run.ResultRecorded {
+		// The outcome was observed and written down. Interrupted describes an
+		// unknown outcome, and saying it of a known failure hides that the job
+		// ran and failed on its own terms.
+		record.ErrorCode = ""
+		if run.ResultOK {
+			record.Status = "ok"
+		}
+	}
+	// The epoch is what groups a journal into invocations, so a terminal record
+	// written without it lands in an invocation of its own and leaves the one it
+	// was meant to close still open.
+	// No Operator: audit takes the last non-empty operator in an epoch group, so
+	// stamping the reconciling operator here would rewrite the interrupted run's
+	// row to name whoever happened to deploy next. The start record already
+	// carries who ran it.
+	writer := &journal.Writer{
+		T: e.T, Names: e.names(), DeployID: operationID, Epoch: run.Epoch,
+	}
+	if err := writer.Append(ctx, record); err != nil {
+		return fmt.Errorf("close interrupted job run %s: %w", operationID, err)
+	}
+	e.logf("closed interrupted job run %s (%s) as %s", operationID, run.Job, record.Status)
+	return nil
+}
diff --git a/internal/engine/job_reconcile_test.go b/internal/engine/job_reconcile_test.go
new file mode 100644
index 0000000..02104f4
--- /dev/null
+++ b/internal/engine/job_reconcile_test.go
@@ -0,0 +1,222 @@
+package engine
+
+import (
+	"bytes"
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/labstack/onebox/internal/journal"
+	"github.com/labstack/onebox/internal/transport"
+)
+
+// reconcileFake serves one journal listing and one running-container listing.
+// `running` is `<id> <operation>` lines, exactly as the label probe formats.
+func reconcileFake(journals string, running []string) *transport.Fake {
+	return &transport.Fake{Dynamic: func(cmd string) (transport.Result, bool) {
+		switch {
+		case strings.Contains(cmd, "for f in"):
+			return transport.Result{Stdout: journals}, true
+		case strings.Contains(cmd, "label='ob.operation'"):
+			return transport.Result{Stdout: strings.Join(running, "\n") + "\n"}, true
+		}
+		return transport.Result{}, false
+	}}
+}
+
+func reconcileEngine(t *testing.T, f *transport.Fake) *Engine {
+	t.Helper()
+	return New(testConfig(), testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
+}
+
+// closeAll reads the journals the way a caller does, then closes what it finds.
+func closeAll(t *testing.T, e *Engine) error {
+	t.Helper()
+	ids, byID, err := journal.Journals(context.Background(), e.T, e.names())
+	if err != nil {
+		t.Fatalf("read journals: %v", err)
+	}
+	return e.closeInterruptedJobRuns(context.Background(), ids, byID)
+}
+
+const startedJobJournal = journalMarkerLine + "J1.jsonl\n" +
+	`{"deploy_id":"J1","epoch":4,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t1"}` + "\n"
+
+// The safety crux, and the reason this asks Docker rather than the journal: a
+// container still changing data with no process owning it must stop the next
+// operation.
+func TestRefuseWhileAnotherOperationsJobContainerRuns(t *testing.T) {
+	f := reconcileFake("", []string{"abc123def456 J1 4"})
+	err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2", 4)
+	if err == nil {
+		t.Fatal("a live job container from another operation must refuse")
+	}
+	for _, want := range []string{"J1", "abc123def456", "still running"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Fatalf("refusal missing %q: %v", want, err)
+		}
+	}
+}
+
+// This operation's own container is not a reason to refuse itself — a deploy
+// runs gate jobs under its own id.
+func TestRefuseAllowsThisOperationsOwnContainer(t *testing.T) {
+	f := reconcileFake("", []string{"abc123def456 J1 4"})
+	if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1", 4); err != nil {
+		t.Fatalf("own container refused: %v", err)
+	}
+}
+
+// A run that recorded its own interruption still has a live container, and a
+// journal reduction would call it finished. That is the Ctrl-C case.
+func TestRefuseCatchesAnInterruptedRunThatRecordedItself(t *testing.T) {
+	journals := startedJobJournal +
+		`{"deploy_id":"J1","epoch":4,"phase":"job","event":"finish","status":"fail","error_code":"interrupted","operation_kind":"job_run","service":"catalog-refresh","ts":"t2"}` + "\n"
+	f := reconcileFake(journals, []string{"abc123def456 J1 4"})
+	if err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2", 4); err == nil {
+		t.Fatal("a recorded interruption must not hide a live container")
+	}
+}
+
+// Gone, and the client never journaled a result: the outcome is unknown.
+func TestCloseRecordsAnUnknownOutcomeAsInterrupted(t *testing.T) {
+	f := reconcileFake(startedJobJournal, nil)
+	if err := closeAll(t, reconcileEngine(t, f)); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	appended := strings.Join(f.Commands, "\n")
+	if !strings.Contains(appended, `"error_code":"interrupted"`) || !strings.Contains(appended, `"event":"finish"`) {
+		t.Fatalf("interrupted run was not recorded honestly:\n%s", appended)
+	}
+	// The epoch groups a journal into invocations. Written without it, the
+	// record lands in an invocation of its own and leaves this one open.
+	if !strings.Contains(appended, `"epoch":4`) {
+		t.Fatalf("terminal record did not join the invocation it closes:\n%s", appended)
+	}
+}
+
+// The one window where success is provable after the fact. The result record is
+// written by the shared job phase and carries no operation kind, so matching it
+// on that would silently never fire.
+func TestCloseRecordsAJournaledResultAsSuccess(t *testing.T) {
+	journals := startedJobJournal +
+		`{"deploy_id":"J1","epoch":4,"phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"ok","ts":"t2"}` + "\n"
+	f := reconcileFake(journals, nil)
+	if err := closeAll(t, reconcileEngine(t, f)); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	appended := strings.Join(f.Commands, "\n")
+	if !strings.Contains(appended, `"status":"ok"`) || strings.Contains(appended, "interrupted") {
+		t.Fatalf("a proven success was not closed as one:\n%s", appended)
+	}
+}
+
+// A plan may be run more than once, appending a second invocation to the same
+// journal. A finish in an earlier epoch says nothing about a later one.
+func TestCloseGroupsAJournalByInvocation(t *testing.T) {
+	journals := startedJobJournal +
+		`{"deploy_id":"J1","epoch":4,"phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t2"}` + "\n" +
+		`{"deploy_id":"J1","epoch":5,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"catalog-refresh","ts":"t3"}` + "\n"
+	f := reconcileFake(journals, nil)
+	if err := closeAll(t, reconcileEngine(t, f)); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	appended := strings.Join(f.Commands, "\n")
+	if !strings.Contains(appended, `"epoch":5`) {
+		t.Fatalf("the unfinished second invocation was not closed:\n%s", appended)
+	}
+	if strings.Count(appended, `"event":"finish"`) != 1 {
+		t.Fatalf("the finished invocation must not be closed again:\n%s", appended)
+	}
+}
+
+// Deploy journals and completed runs are not orphans.
+func TestCloseIgnoresDeploysAndFinishedRuns(t *testing.T) {
+	journals := journalMarkerLine + "R1.jsonl\n" +
+		`{"deploy_id":"R1","epoch":1,"phase":"deploy","event":"start","ts":"t"}` + "\n" +
+		journalMarkerLine + "J2.jsonl\n" +
+		`{"deploy_id":"J2","epoch":1,"phase":"job","event":"start","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n" +
+		`{"deploy_id":"J2","epoch":1,"phase":"job","event":"finish","status":"ok","operation_kind":"job_run","service":"chore","ts":"t"}` + "\n"
+	f := reconcileFake(journals, nil)
+	if err := closeAll(t, reconcileEngine(t, f)); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	if strings.Contains(strings.Join(f.Commands, "\n"), `"event":"finish"`) {
+		t.Fatalf("nothing was unfinished, so nothing should be written:\n%s", strings.Join(f.Commands, "\n"))
+	}
+}
+
+// A recorded failure is evidence of the outcome exactly as much as a recorded
+// success. Calling it interrupted would hide that the job ran and failed on its
+// own terms.
+func TestCloseKeepsARecordedFailureAsAFailure(t *testing.T) {
+	journals := startedJobJournal +
+		`{"deploy_id":"J1","epoch":4,"phase":"job","sub_step":"job:catalog-refresh","event":"result","status":"fail","ts":"t2"}` + "\n"
+	f := reconcileFake(journals, nil)
+	if err := closeAll(t, reconcileEngine(t, f)); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	appended := strings.Join(f.Commands, "\n")
+	if !strings.Contains(appended, `"status":"fail"`) {
+		t.Fatalf("a recorded failure was not closed as one:\n%s", appended)
+	}
+	if strings.Contains(appended, "interrupted") {
+		t.Fatalf("a known failure must not be reported as an unknown outcome:\n%s", appended)
+	}
+}
+
+// A container whose label carries no value prints "<id> " with nothing after
+// the separator. Trimming the line before the cut removes the separator, and
+// the container is skipped as unparseable — the one that most needs refusing.
+func TestRefuseCatchesAContainerWithAnEmptyOperationLabel(t *testing.T) {
+	f := reconcileFake("", []string{"abc123def456  "})
+	err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J2", 4)
+	if err == nil {
+		t.Fatal("an unattributable job container must refuse")
+	}
+	if !strings.Contains(err.Error(), "empty") || !strings.Contains(err.Error(), "abc123def456") {
+		t.Fatalf("refusal did not name the problem: %v", err)
+	}
+}
+
+// A sealed job plan carries one operation id for its whole life and is
+// re-runnable, and AcquireLock hands the lock straight back to a caller
+// presenting the id already written in it. Matching the operation alone would
+// let a second run exempt the container its own earlier run left behind.
+func TestRefuseCatchesAnEarlierRunOfTheSameOperation(t *testing.T) {
+	f := reconcileFake("", []string{"abc123def456 J1 4"})
+	err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1", 5)
+	if err == nil {
+		t.Fatal("an earlier invocation of the same plan must refuse")
+	}
+	for _, want := range []string{"earlier run", "J1", "epoch 4", "abc123def456"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Fatalf("refusal missing %q: %v", want, err)
+		}
+	}
+}
+
+// A container carrying no epoch label cannot be shown to belong to this
+// invocation, so it is not exempt from it either.
+func TestRefuseDoesNotExemptAContainerWithNoEpoch(t *testing.T) {
+	f := reconcileFake("", []string{"abc123def456 J1 "})
+	err := reconcileEngine(t, f).refuseForeignJobContainers(context.Background(), "J1", 4)
+	if err == nil || !strings.Contains(err.Error(), "epoch unknown") {
+		t.Fatalf("unlabelled epoch = %v, want a refusal naming it", err)
+	}
+}
+
+// The reconciling operator must not be recorded as the interrupted run's.
+// Audit takes the last non-empty operator in an epoch group, so stamping it
+// here rewrites the row to name whoever deployed next.
+func TestCloseDoesNotAttributeTheRunToTheReconciler(t *testing.T) {
+	f := reconcileFake(startedJobJournal, nil)
+	if err := closeAll(t, reconcileEngine(t, f)); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	for _, c := range f.Commands {
+		if strings.Contains(c, `"event":"finish"`) && strings.Contains(c, `"operator"`) {
+			t.Fatalf("terminal record claimed an operator:\n%s", c)
+		}
+	}
+}
diff --git a/internal/engine/job_test.go b/internal/engine/job_test.go
index 88cfd4b..f2056de 100644
--- a/internal/engine/job_test.go
+++ b/internal/engine/job_test.go
@@ -172,3 +172,76 @@ func TestInterruptedRunClassifiesTheRunNotTheClient(t *testing.T) {
 		t.Fatal("a job that failed on its own terms is not interrupted")
 	}
 }
+
+// Reconciliation runs before this operation writes its own start record. Run it
+// after, and the snapshot contains a start with no finish yet — this very run —
+// which the reconciler would close as interrupted before the job executes.
+func TestRunJobDoesNotCloseTheRunItIsAboutToStart(t *testing.T) {
+	const runtime = "services:\n  migrate:\n    image: ghcr.io/x/app@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"
+	target := currentJobFake(runtime)
+	// The journal listing reflects what this run has appended so far, so the
+	// snapshot's contents depend on when it is taken — which is the whole
+	// question.
+	inner := target.Dynamic
+	target.Dynamic = func(cmd string) (transport.Result, bool) {
+		if strings.Contains(cmd, "for f in") {
+			var lines []string
+			for _, c := range target.Commands {
+				start := strings.Index(c, `{"deploy_id":"op-job-run"`)
+				if start < 0 {
+					continue
+				}
+				if end := strings.LastIndex(c, "}"); end > start {
+					lines = append(lines, c[start:end+1])
+				}
+			}
+			if len(lines) == 0 {
+				return transport.Result{}, true
+			}
+			return transport.Result{Stdout: "@@ob-journal@@op-job-run.jsonl\n" + strings.Join(lines, "\n") + "\n"}, true
+		}
+		return inner(cmd)
+	}
+	engine := manualJobEngine(t, target)
+	request := JobRunRequest{
+		OperationID: "op-job-run", Job: "migrate", ExpectedRelease: engineTestPreviousReleaseID,
+		ExpectedRuntimeDigest: HashBytes([]byte(runtime)), ExpectedDataEffect: "none",
+	}
+	if _, _, err := engine.RunJobWithJournalID(context.Background(), request); err != nil {
+		t.Fatalf("run job: %v", err)
+	}
+	commands := strings.Join(target.Commands, "\n")
+	if strings.Contains(commands, `"error_code":"interrupted"`) {
+		t.Fatalf("the run closed itself as interrupted before executing:\n%s", commands)
+	}
+	// Exactly one terminal record: the real one, written after the job ran.
+	if got := strings.Count(commands, `"phase":"job","event":"finish"`); got != 1 {
+		t.Fatalf("terminal record count = %d, want 1:\n%s", got, commands)
+	}
+}
+
+// The refusal is only worth having if it is actually called. Deleting the call
+// site left every test green, so this drives the whole run against a host
+// reporting a foreign job container and requires it to stop.
+func TestRunJobRefusesWhileAForeignJobContainerRuns(t *testing.T) {
+	const runtime = "services:\n  migrate:\n    image: ghcr.io/x/app@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"
+	target := currentJobFake(runtime)
+	inner := target.Dynamic
+	target.Dynamic = func(cmd string) (transport.Result, bool) {
+		if strings.Contains(cmd, "label='ob.operation'") {
+			return transport.Result{Stdout: "abc123def456 other-op 2\n"}, true
+		}
+		return inner(cmd)
+	}
+	engine := manualJobEngine(t, target)
+	_, _, err := engine.RunJobWithJournalID(context.Background(), JobRunRequest{
+		OperationID: "op-job-run", Job: "migrate", ExpectedRelease: engineTestPreviousReleaseID,
+		ExpectedRuntimeDigest: HashBytes([]byte(runtime)), ExpectedDataEffect: "none",
+	})
+	if err == nil || !strings.Contains(err.Error(), "other-op") {
+		t.Fatalf("run job = %v, want a refusal naming the foreign operation", err)
+	}
+	if strings.Contains(strings.Join(target.Commands, "\n"), "ONEBOX_RESULT_FILE=") {
+		t.Fatalf("the job ran anyway:\n%s", strings.Join(target.Commands, "\n"))
+	}
+}
diff --git a/internal/onebox/operation_errors.go b/internal/onebox/operation_errors.go
index ce7ba7e..9370af4 100644
--- a/internal/onebox/operation_errors.go
+++ b/internal/onebox/operation_errors.go
@@ -123,7 +123,8 @@ var operationFailureDefinitions = map[string]OperationFailure{
 		// Not a failure of the job and not an incomplete operation: the client
 		// went away, and what the job did is unknown unless it journaled a
 		// result first. Written by the interrupted run itself, on a context of
-		// its own, whenever it can still reach the host.
+		// its own, whenever it can still reach the host — and otherwise by the
+		// next operation, which finds the run unfinished and closes it.
 		Message: "the operation's client went away before its outcome could be recorded",
 		Command: "ob audit --output json",
 	},
diff --git a/internal/onebox/service_test.go b/internal/onebox/service_test.go
index 113c25b..12aeecd 100644
--- a/internal/onebox/service_test.go
+++ b/internal/onebox/service_test.go
@@ -81,6 +81,9 @@ func serviceFake() *transport.Fake {
 				return transport.Result{Stdout: "demo\n"}, true
 			case strings.Contains(cmd, "readlink"):
 				return transport.Result{Stdout: "releases/R0\n"}, true
+			// Ahead of the project-container probe, which also uses --format.
+			case strings.Contains(cmd, "label='ob.operation'"):
+				return transport.Result{Stdout: "\n"}, true
 			case strings.Contains(cmd, "docker ps") && strings.Contains(cmd, "--format"):
 				return transport.Result{Stdout: "S1|web|R0|Up (healthy)\nPG1|database|R0|Up (healthy)\n"}, true
 			case strings.Contains(cmd, "for f in"):
@@ -267,6 +270,8 @@ deployment: {order: [web, worker]}
 			return transport.Result{Stdout: project(false)}, true
 		case strings.Contains(command, "cat ") && strings.Contains(command, "compose.yaml"):
 			return transport.Result{Stdout: liveCompose}, true
+		case strings.Contains(command, "label='ob.operation'"):
+			return transport.Result{Stdout: "\n"}, true
 		case strings.Contains(command, "docker ps") && strings.Contains(command, "--format"):
 			return transport.Result{Stdout: "S1|web|R0|Up\nW1|worker|R0|Up\n"}, true
 		default: