fix(job): supervise scheduled manual runs on host

cmd/ob/job.go

@@ -34,25 +34,28 @@ func addJobCommand(root *cobra.Command, g *globalFlags) {
 	plan.Flags().StringVar(&backupReportOut, "backup-report-out", "", "write a plan-bound backup report template when migration backup is required")
 
 	var planPath, approvalPath, backupReportPath, overrideReason string
-	var breakLock bool
+	var breakLock, detach bool
 	run := &cobra.Command{
 		Use:   "run [id]",
 		Short: "run one manual job from an inline or saved sealed plan",
-		Long:  "Run one manual job through the canonical lock, fence, local-confirmation and journal boundary.\n\nHumans may pass an id and confirm interactively. Automation should supply a\nsaved --plan and its separately created local-confirmation artifact through\n--approval; migration plans may also require the exact plan-bound --backup-report.",
-		Args:  cobra.MaximumNArgs(1),
+		Long: "Run one manual job through the canonical lock, fence, local-confirmation and journal boundary.\n\n" +
+			"A job with schedule configured runs under its installed systemd unit and is\nfollowed by default; Ctrl-C stops following, not the host job. --detach returns\nafter that unit accepts the run. Unscheduled and migration jobs stay attached.\n\n" +
+			"Humans may pass an id and confirm interactively. Automation should supply a\nsaved --plan and its separately created local-confirmation artifact through\n--approval; migration plans may also require the exact plan-bound --backup-report.",
+		Args: cobra.MaximumNArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			jobID := ""
 			if len(args) == 1 {
 				jobID = args[0]
 			}
-			return runJob(cmd, g, jobID, planPath, approvalPath, backupReportPath, overrideReason, breakLock)
+			return runJob(cmd, g, jobID, planPath, approvalPath, backupReportPath, overrideReason, breakLock, detach)
 		},
 	}
 	run.Flags().StringVar(&planPath, "plan", "", "apply a saved job plan artifact")
 	run.Flags().StringVar(&approvalPath, "approval", "", "apply a plan-bound local confirmation artifact")
 	run.Flags().StringVar(&backupReportPath, "backup-report", "", "apply the backup report bound into the local confirmation")
 	run.Flags().StringVar(&overrideReason, "override-migration-backup", "", "audited break-glass reason (requires --approval)")
 	run.Flags().BoolVar(&breakLock, "break-lock", false, "break a stale operation lock after inspecting its holder")
+	run.Flags().BoolVar(&detach, "detach", false, "return after the installed host unit accepts the job")
 
 	group.AddCommand(plan, run)
 	root.AddCommand(group)
@@ -163,7 +166,7 @@ func renderJobPlan(cmd *cobra.Command, plan *onebox.JobPlan) {
 	}
 }
 
-func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, backupReportPath, overrideReason string, breakLock bool) error {
+func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, backupReportPath, overrideReason string, breakLock, detach bool) error {
 	if planPath != "" && jobID != "" {
 		return writeEarlyOperationFailure(cmd, g, errors.New("supply either a job id or --plan, not both"))
 	}
@@ -217,7 +220,7 @@ func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, b
 		}
 		return runMutation(cmd, g, onebox.ExecuteRequest{
 			Kind: onebox.KindJobRun, JobPlan: plan, Approval: approval, BreakLock: breakLock,
-			BackupReport: backupReport, MigrationBackupOverride: override,
+			BackupReport: backupReport, MigrationBackupOverride: override, Detach: detach,
 		}, "job run")
 	}
 
@@ -245,7 +248,7 @@ func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, b
 		}
 		approval = &grant
 	}
-	return runMutation(cmd, g, onebox.ExecuteRequest{Kind: onebox.KindJobRun, JobPlan: &plan, Approval: approval, BreakLock: breakLock}, "job run")
+	return runMutation(cmd, g, onebox.ExecuteRequest{Kind: onebox.KindJobRun, JobPlan: &plan, Approval: approval, BreakLock: breakLock, Detach: detach}, "job run")
 }
 
 func loadJobMigrationOverride(

e2e/server_test.go

@@ -401,6 +401,21 @@ HTTPServer(("127.0.0.1", 18080), Handler).handle_request()
 		if err != nil {
 			t.Fatalf("job run failed: %v\n%s", err, out)
 		}
+		for _, want := range []string{"host run ", "⟳ job chore", "✓ job chore", "ob schedule history chore"} {
+			if !strings.Contains(out, want) {
+				t.Fatalf("host-owned job output is missing %q:\n%s", want, out)
+			}
+		}
+
+		detachedPlan := filepath.Join(dir, "ob-detached-job-plan.json")
+		s.mustOb(t, dir, "job", "plan", "chore", "-o", detachedPlan)
+		out, err = s.obInput(t, dir, s.obHome(t), "y\n", "job", "run", "--plan", detachedPlan, "--detach")
+		if err != nil {
+			t.Fatalf("detached job run failed: %v\n%s", err, out)
+		}
+		if !strings.Contains(out, "job chore accepted as ") || !strings.Contains(out, "ob schedule history chore") {
+			t.Fatalf("detached job did not report durable acceptance:\n%s", out)
+		}
 	})
 
 	t.Run("doctor", func(t *testing.T) {

internal/engine/schedule.go

@@ -265,6 +265,7 @@ func scheduleRunnerScript(application string, job app.ScheduledJob, names app.Na
 		"trap 'exit 143' 15",
 	)
 	lines = append(lines, scheduleRunPreamble(triggerUnit)...)
+	lines = append(lines, schedulePlannedBindingLines()...)
 	if job.DataEffect != app.DataEffectNone {
 		lines = append(lines, invalidateExecutionCommand(names.AppDir()))
 	}
@@ -310,6 +311,7 @@ func pinnedScheduleRunnerScript(application string, job app.ScheduledJob, names
 		"trap 'exit 143' 15",
 	)
 	lines = append(lines, scheduleRunPreamble(triggerUnit)...)
+	lines = append(lines, schedulePlannedBindingLines()...)
 	lines = append(lines, scheduleAttemptLoop(job, compose, container)...)
 	lines = append(lines, "")
 	return strings.Join(lines, "\n")
@@ -518,13 +520,17 @@ func scheduleInputsLines(inputsPath string) []string {
 	return []string{
 		"operation=''",
 		"execution=''",
+		"expected_release=''",
+		"expected_runtime=''",
 		"inputs_json=''",
 		"inputs_file=" + q(inputsPath),
 		"if [ -z \"${TRIGGER_UNIT:-}\" ] && [ -f \"$inputs_file\" ]; then",
 		"  while IFS= read -r line || [ -n \"$line\" ]; do",
 		"    case \"$line\" in",
 		"      ONEBOX_OPERATION=*) operation=${line#ONEBOX_OPERATION=} ;;",
 		"      ONEBOX_EXECUTION=*) execution=${line#ONEBOX_EXECUTION=} ;;",
+		"      ONEBOX_EXPECTED_RELEASE=*) expected_release=${line#ONEBOX_EXPECTED_RELEASE=} ;;",
+		"      ONEBOX_EXPECTED_RUNTIME=*) expected_runtime=${line#ONEBOX_EXPECTED_RUNTIME=} ;;",
 		"      [A-Z]*=*) set -- \"$@\" -e \"$line\"; key=${line%%=*}; value=${line#*=}; inputs_json=\"${inputs_json:+$inputs_json,}\\\"$key\\\":\\\"$value\\\"\" ;;",
 		"    esac",
 		"  done <\"$inputs_file\"",
@@ -533,6 +539,24 @@ func scheduleInputsLines(inputsPath string) []string {
 	}
 }
 
+// schedulePlannedBindingLines makes a sealed manual job plan authoritative at
+// the point that owns execution: after the host runner has acquired its locks,
+// immediately before it can start the container. Timer firings carry no
+// expected binding and pass through unchanged.
+const sealedManualJobBindingMarker = "Sealed manual job binding protocol v1."
+
+func schedulePlannedBindingLines() []string {
+	return []string{
+		"# " + sealedManualJobBindingMarker,
+		"if [ -n \"$expected_release\" ] && [ \"$release\" != \"$expected_release\" ]; then write_state 0; echo 'onebox: serving release changed after job approval' >&2; exit 74; fi",
+		"if [ -n \"$expected_runtime\" ]; then",
+		"  runtime_hash=$(sha256sum \"$release_dir/compose.yaml\") || { write_state 0; echo 'onebox: cannot hash the approved job runtime' >&2; exit 74; }",
+		"  runtime_digest=sha256:${runtime_hash%% *}",
+		"  if [ \"$runtime_digest\" != \"$expected_runtime\" ]; then write_state 0; echo 'onebox: serving runtime changed after job approval' >&2; exit 74; fi",
+		"fi",
+	}
+}
+
 // systemdVersion reads the leading number from `systemd 255 (255.4-1ubuntu8)`.
 func systemdVersion(firstLine string) (int, bool) {
 	fields := strings.Fields(firstLine)

internal/engine/schedule_execution.go

@@ -199,7 +199,7 @@ func (e *Engine) ExecutionResume(ctx context.Context, operation, id string, wait
 		return ScheduleRunResult{}, fmt.Errorf("resume requires the original current release")
 	}
 	// Host runner rechecks the saved definition and compatibility under locks.
-	return e.scheduleRun(ctx, operation, job, nil, wait, id)
+	return e.scheduleRun(ctx, operation, job, nil, wait, id, nil)
 }
 
 func (e *Engine) ExecutionAbandon(ctx context.Context, operation, id string) (err error) {

internal/engine/schedule_execution_test.go

@@ -167,7 +167,7 @@ func TestDurableResumeRefusesLegacyRunnerBeforePublishingRequest(t *testing.T) {
 		return base(command)
 	}
 	e := New(cfg, testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep})
-	_, err := e.scheduleRun(context.Background(), "resume-op", "refresh", nil, false, strings.Repeat("a", 32))
+	_, err := e.scheduleRun(context.Background(), "resume-op", "refresh", nil, false, strings.Repeat("a", 32), nil)
 	if err == nil || !strings.Contains(err.Error(), "ob schedule apply") {
 		t.Fatalf("legacy runner accepted: %v", err)
 	}

internal/engine/schedule_run.go

@@ -35,10 +35,29 @@ type ScheduleRunResult struct {
 // the inputs is the case the sealed plan of `ob job run` exists for, and a
 // migration or destructive job keeps that path.
 func (e *Engine) ScheduleRun(ctx context.Context, operationID, name string, inputs map[string]string, wait bool) (_ ScheduleRunResult, err error) {
-	return e.scheduleRun(ctx, operationID, name, inputs, wait, "")
+	return e.scheduleRun(ctx, operationID, name, inputs, wait, "", nil)
 }
 
-func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inputs map[string]string, wait bool, execution string) (_ ScheduleRunResult, err error) {
+// PlannedJobRun submits a sealed manual job plan to the job's installed
+// systemd unit. The runner checks the plan's release and runtime digest after
+// taking the host-side application exclusion, so releasing the admission lock
+// before systemd schedules the unit cannot move the job onto different bytes.
+func (e *Engine) PlannedJobRun(ctx context.Context, operationID, name, expectedRelease, expectedRuntime string, wait bool) (_ ScheduleRunResult, err error) {
+	if expectedRelease == "" || expectedRuntime == "" {
+		return ScheduleRunResult{}, errors.New("planned job run requires an expected release and runtime digest")
+	}
+	return e.scheduleRun(ctx, operationID, name, nil, wait, "", &plannedJobBinding{
+		release: expectedRelease,
+		runtime: expectedRuntime,
+	})
+}
+
+type plannedJobBinding struct {
+	release string
+	runtime string
+}
+
+func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inputs map[string]string, wait bool, execution string, planned *plannedJobBinding) (_ ScheduleRunResult, err error) {
 	result := ScheduleRunResult{Job: name, Operation: operationID, Inputs: inputs}
 	result.Execution = execution
 	if strings.TrimSpace(operationID) == "" {
@@ -54,10 +73,13 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 	if execution != "" && (workload.Execution == nil || !scheduleRunID.MatchString(execution) || len(inputs) != 0) {
 		return result, errors.New("resume requires a durable job, a valid execution ID, and no input overrides")
 	}
-	if workload.DataEffect != app.DataEffectNone {
+	if workload.DataEffect != app.DataEffectNone && planned == nil {
 		return result, fmt.Errorf("job %s declares data_effect %q; operator-initiated runs of it go through the sealed plan: ob job plan %s, then ob job run",
 			name, workload.DataEffect, name)
 	}
+	if planned != nil && execution != "" {
+		return result, errors.New("a planned job run cannot resume a durable execution")
+	}
 	if err := app.ValidateJobInputValues(workload, inputs); err != nil {
 		return result, err
 	}
@@ -70,6 +92,15 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 	}
 	unit := e.names().ScheduledJobUnit(name)
 	result.Unit = unit
+	if planned != nil {
+		res, err := e.T.Run(ctx, "grep -Fq "+q(sealedManualJobBindingMarker)+" "+q("/etc/systemd/system/"+unit+".run"))
+		if err != nil {
+			return result, err
+		}
+		if res.ExitCode != 0 {
+			return result, fmt.Errorf("installed job runner does not support sealed manual-job binding; run `ob schedule apply` before running %s", name)
+		}
+	}
 
 	// Starting an active unit is a no-op to systemd and would consume the
 	// inputs file for a run that never happens; say so instead.
@@ -113,7 +144,7 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 	path := e.names().ScheduledJobRunInputs(name)
 	create := "if [ -e " + q(path) + " ]; then exit 73; fi; " +
 		"umask 077 && install -d -m 700 " + q(e.names().AppDir()+"/schedule") + " && set -C && cat > " + q(path)
-	payload := scheduleInputsFile(operationID, inputs)
+	payload := scheduleInputsFile(operationID, inputs, planned)
 	if execution != "" {
 		payload += "ONEBOX_EXECUTION=" + execution + "\n"
 	}
@@ -139,6 +170,8 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 	writer := &journal.Writer{
 		T: e.T, Names: e.names(), DeployID: operationID, Epoch: epoch, Operator: journal.DefaultOperator(),
 		GitSHA: e.Opts.GitSHA, ConfigHash: e.Opts.ConfigHash, Runner: &e.Opts.Runner,
+		ApprovalDigest: e.Opts.ApprovalDigest, ApprovalClass: e.Opts.ApprovalClass,
+		ApprovedBy: e.Opts.ApprovedBy, ApprovalSource: e.Opts.ApprovalSource,
 	}
 	detail := "inputs: defaults"
 	if execution != "" {
@@ -148,6 +181,15 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 		detail = "inputs: " + scheduleInputsDetail(inputs)
 	}
 	record := journal.Record{Phase: "schedule-run", Event: "start", Status: "ok", Target: name, TargetKind: "job", Detail: detail}
+	if planned != nil {
+		// This journal describes host-unit admission, not the job's eventual
+		// outcome. Keep schedule-run's truthful "started" audit semantics while
+		// retaining the sealed job plan's authorization evidence.
+		record.ApprovalDigest = e.Opts.ApprovalDigest
+		record.ApprovalClass = e.Opts.ApprovalClass
+		record.ApprovedBy = e.Opts.ApprovedBy
+		record.ApprovalSource = e.Opts.ApprovalSource
+	}
 	if err := writer.Append(ctx, record); err != nil {
 		return result, fmt.Errorf("journal schedule run start: %w", err)
 	}
@@ -176,6 +218,14 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 	if wait {
 		start = "systemctl start " + q(unit+".service")
 	}
+	finishStep := func(error) {}
+	if wait {
+		if planned != nil {
+			e.ui.Infof("host run %s; Ctrl-C detaches; inspect with `ob schedule history %s`", operationID, name)
+		}
+		finishStep = e.ui.Step("job "+name, true)
+		defer func() { finishStep(err) }()
+	}
 	res, err = e.mutate(ctx, start)
 	if err != nil {
 		return result, err
@@ -187,7 +237,7 @@ func (e *Engine) scheduleRun(ctx context.Context, operationID, name string, inpu
 		// The unit is queued, so the runner owns the file now.
 		pending = false
 		result.Started = true
-		e.logf("schedule: %s started as %s; ob schedule history %s shows the outcome", name, operationID, name)
+		e.ui.Successf("job %s accepted as %s; inspect with `ob schedule history %s`", name, operationID, name)
 		return result, nil
 	}
 	// A blocking start that exits non-zero may mean the job failed, which is
@@ -263,8 +313,14 @@ func (e *Engine) discardInputs(ctx context.Context, path string) {
 // id on its reserved line, then one declared override per line. Values were
 // validated against a charset that has no newline or quote, so the format
 // needs no escaping.
-func scheduleInputsFile(operationID string, inputs map[string]string) string {
+func scheduleInputsFile(operationID string, inputs map[string]string, planned *plannedJobBinding) string {
 	lines := []string{app.ReservedInputPrefix + "OPERATION=" + operationID}
+	if planned != nil {
+		lines = append(lines,
+			app.ReservedInputPrefix+"EXPECTED_RELEASE="+planned.release,
+			app.ReservedInputPrefix+"EXPECTED_RUNTIME="+planned.runtime,
+		)
+	}
 	for _, name := range sortedInputNames(inputs) {
 		lines = append(lines, name+"="+inputs[name])
 	}

internal/engine/schedule_run_test.go

@@ -72,6 +72,68 @@ func TestScheduleRunWritesInputsJournalsThenStartsAfterReleasingTheLock(t *testi
 	}
 }
 
+func TestPlannedJobRunStagesItsExactBindingAndDetachesToSystemd(t *testing.T) {
+	cfg := testConfig()
+	cfg.Workloads["refresh"] = app.Workload{
+		Role: app.RoleJob, When: "manual", DataEffect: "destructive",
+		Schedule: &app.JobSchedule{Cron: "0 4 * * 1", Timezone: "UTC", Timeout: "8h"},
+	}
+	f := happyFake()
+	base := f.Dynamic
+	f.Dynamic = func(cmd string) (transport.Result, bool) {
+		switch {
+		case strings.Contains(cmd, "command -v flock"):
+			return transport.Result{Stdout: "ok\n"}, true
+		case strings.Contains(cmd, "systemctl --version"):
+			return transport.Result{Stdout: "systemd 255 (255.4-1ubuntu8)\n"}, true
+		case strings.Contains(cmd, "systemctl is-active"):
+			return transport.Result{Stdout: "inactive\n"}, true
+		case strings.Contains(cmd, "systemctl start"):
+			return transport.Result{}, true
+		}
+		return base(cmd)
+	}
+	e := New(cfg, testProject(t), f, Options{
+		Out: &bytes.Buffer{}, Sleep: noSleep,
+		ApprovalDigest: "approval-digest", ApprovalClass: "strong",
+		ApprovedBy: "operator", ApprovalSource: "local_confirmation",
+	})
+	const (
+		operation = "20260914-191943-job_run-9503bc4cfa47"
+		release   = "20260914-190602-deploy-731e31b2d992"
+		runtime   = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	)
+	result, err := e.PlannedJobRun(context.Background(), operation, "refresh", release, runtime, false)
+	if err != nil {
+		t.Fatalf("planned job run: %v\n%s", err, strings.Join(f.Commands, "\n"))
+	}
+	if !result.Started || result.Operation != operation || result.Unit != "ob-sample-refresh" {
+		t.Fatalf("result = %#v", result)
+	}
+	written := strings.Join(f.Inputs, "\n")
+	for _, want := range []string{
+		"ONEBOX_OPERATION=" + operation,
+		"ONEBOX_EXPECTED_RELEASE=" + release,
+		"ONEBOX_EXPECTED_RUNTIME=" + runtime,
+	} {
+		if !strings.Contains(written, want) {
+			t.Fatalf("planned activation omitted %q:\n%s", want, written)
+		}
+	}
+	commands := strings.Join(f.Commands, "\n")
+	if !strings.Contains(commands, "grep -Fq '"+sealedManualJobBindingMarker+"' '/etc/systemd/system/ob-sample-refresh.run'") {
+		t.Fatalf("planned job did not verify the installed runner protocol:\n%s", commands)
+	}
+	if !strings.Contains(commands, "systemctl start --no-block 'ob-sample-refresh.service'") {
+		t.Fatalf("planned job was not detached to systemd:\n%s", commands)
+	}
+	for _, want := range []string{`"approval_digest":"approval-digest"`, `"approval_class":"strong"`} {
+		if !strings.Contains(commands, want) {
+			t.Fatalf("planned job journal omitted %q:\n%s", want, commands)
+		}
+	}
+}
+
 func TestScheduleRunRefusals(t *testing.T) {
 	cfg := testConfig()
 	cfg.Workloads["sync"] = app.Workload{

internal/engine/schedule_test.go

@@ -201,12 +201,24 @@ func TestScheduledJobUnitContract(t *testing.T) {
 		"compose.yaml",
 		`run --rm --no-deps "$@" --name 'sample-nightly-1'`,
 		"docker rm -f 'sample-nightly-1'",
+		"ONEBOX_EXPECTED_RELEASE",
+		"ONEBOX_EXPECTED_RUNTIME",
+		sealedManualJobBindingMarker,
+		"sha256sum",
+		"serving release changed after job approval",
+		"serving runtime changed after job approval",
 		"nightly",
 	} {
 		if !strings.Contains(runner, want) {
 			t.Errorf("runner is missing %q:\n%s", want, runner)
 		}
 	}
+	locked := strings.Index(runner, "flock --exclusive --timeout 10")
+	bound := strings.Index(runner, "serving release changed after job approval")
+	run := strings.Index(runner, "docker compose")
+	if locked < 0 || bound < 0 || run < 0 || !(locked < bound && bound < run) {
+		t.Fatalf("planned binding must be checked under the app lock before the container runs:\n%s", runner)
+	}
 	command := exec.CommandContext(context.Background(), "sh", "-n")
 	command.Stdin = strings.NewReader(runner)
 	if output, err := command.CombinedOutput(); err != nil {