fix: reconcile schedules after runner upgrades

cmd/ob/ops.go

@@ -64,6 +64,29 @@ func addOpsCommands(root *cobra.Command, g *globalFlags) {
 	proxyCmd.AddCommand(proxyApplyCmd)
 	root.AddCommand(proxyCmd)
 
+	// schedule apply — explicitly reconcile runner-owned host timers after an
+	// Onebox upgrade, without coupling package installation to remote mutation.
+	scheduleCmd := &cobra.Command{Use: "schedule", Short: "manage host timers for scheduled jobs",
+		Long: "Manage the systemd timers generated for scheduled jobs.\n\n" +
+			"Timers outlive the Onebox process and the package installed on the operator\n" +
+			"workstation. `apply` explicitly reconciles their units after a runner or\n" +
+			"configuration change without deploying a release.",
+		Args: cobra.NoArgs, RunE: showCommandHelp}
+	var scheduleBreakLock bool
+	scheduleApplyCmd := &cobra.Command{
+		Use:   "apply",
+		Short: "reconcile scheduled-job units without deploying a release",
+		Long:  "Converge every declared scheduled-job timer, service, runner, and failure notifier to what the current Onebox runner generates.\n\nTaken under the application lock and fence so a deploy or host-fired job cannot modify the same runtime concurrently. This is the explicit post-upgrade path; upgrading the local package never mutates a remote host by itself.",
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			return runMutation(cmd, g, onebox.ExecuteRequest{
+				Kind: onebox.KindScheduleApply, BreakLock: scheduleBreakLock,
+			}, "schedule apply")
+		},
+	}
+	scheduleApplyCmd.Flags().BoolVar(&scheduleBreakLock, "break-lock", false, "break a stale operation lock after inspecting its holder")
+	scheduleCmd.AddCommand(scheduleApplyCmd)
+	root.AddCommand(scheduleCmd)
+
 	// secrets list | edit | push
 	secretsCmd := &cobra.Command{Use: "secrets", Short: "SOPS-encrypted secrets",
 		Long: "SOPS-encrypted secrets for this project.\n\n" +

cmd/ob/output.go

@@ -104,6 +104,7 @@ var cliOutputMatrix = map[string]cliOutputClass{
 	"ob proxy apply":    {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
 	"ob resume":         {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
 	"ob rollback":       {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
+	"ob schedule apply": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
 	"ob schema":         {Class: cliClassFiniteEnvelope, JSON: true},
 	"ob secrets edit":   {Class: cliClassTrustedEditor, JSON: true},
 	"ob secrets list":   {Class: cliClassFiniteEnvelope, JSON: true},

cmd/ob/output_test.go

@@ -518,6 +518,7 @@ func TestLeafOutputMatrixIsClosedAndHasNoAliases(t *testing.T) {
 		"ob proxy apply":    {Class: "finite_stream", JSON: true, NDJSON: true},
 		"ob resume":         {Class: "finite_stream", JSON: true, NDJSON: true},
 		"ob rollback":       {Class: "finite_stream", JSON: true, NDJSON: true},
+		"ob schedule apply": {Class: "finite_stream", JSON: true, NDJSON: true},
 		"ob schema":         {Class: "finite_envelope", JSON: true},
 		"ob secrets edit":   {Class: "trusted_editor", JSON: true},
 		"ob secrets list":   {Class: "finite_envelope", JSON: true},

e2e/server_test.go

@@ -189,6 +189,42 @@ func TestServerLifecycle(t *testing.T) {
 			"/etc/systemd/system/ob-observer-timeout--chore.service "+
 			"/etc/systemd/system/ob-observer-timeout--chore.timer")
 
+		// Model a host last touched by v2026.8.5: the timer exists, but its
+		// service invokes Compose directly and has no bounded runner or notifier.
+		// Upgrading the local package is intentionally side-effect free; the
+		// scoped apply command must bridge that installed generation without an
+		// unrelated release deploy.
+		legacyService := `[Unit]
+Description=Onebox scheduled job chore for observer
+After=docker.service
+Requires=docker.service
+
+[Service]
+Type=oneshot
+ExecStart=/usr/bin/docker compose -p observer -f /var/lib/ob/observer/current/compose.yaml run --rm --no-deps chore
+`
+		encodedLegacy := base64.StdEncoding.EncodeToString([]byte(legacyService))
+		s.run(t, strings.Join([]string{
+			"printf '%s' '" + encodedLegacy + "' | base64 -d > /etc/systemd/system/ob-observer-chore.service",
+			"rm -f /etc/systemd/system/ob-observer-chore.run /etc/systemd/system/ob-observer-chore.notify",
+			"systemctl daemon-reload",
+		}, "\n"))
+		before := s.run(t, "systemctl cat ob-observer-chore.service")
+		if strings.Contains(before, "TimeoutStartSec=") || strings.Contains(before, "ExecStopPost=") {
+			t.Fatalf("legacy fixture already has the current unit contract:\n%s", before)
+		}
+		s.mustOb(t, dir, "schedule", "apply")
+		after := s.run(t, strings.Join([]string{
+			"test -s /etc/systemd/system/ob-observer-chore.run",
+			"test -s /etc/systemd/system/ob-observer-chore.notify",
+			"systemctl cat ob-observer-chore.service",
+		}, "\n"))
+		for _, want := range []string{"ExecStart=/bin/sh", "ExecStopPost=/bin/sh", "TimeoutStartSec="} {
+			if !strings.Contains(after, want) {
+				t.Fatalf("schedule apply did not restore %q:\n%s", want, after)
+			}
+		}
+
 		// A normal host-fired run proves the generated runner, current-release
 		// lookup, Docker invocation and app-wide schedule lock compose on systemd.
 		s.run(t, "systemctl start ob-observer-chore.service")

internal/engine/schedule_apply.go

@@ -0,0 +1,75 @@
+package engine
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/labstack/onebox/internal/journal"
+)
+
+// ScheduleApply explicitly converges the host units generated for scheduled
+// jobs without deploying a release. Package installation stays local and
+// side-effect free; an operator chooses when a newer runner may rewrite remote
+// units.
+//
+// The application lock also serializes against deploys. AcquireLock takes the
+// schedule flock while publishing that lock, so a timer cannot begin between
+// the ownership check and the fence write.
+func (e *Engine) ScheduleApply(ctx context.Context, operationID string) (err error) {
+	if err := e.RequireHostOwner(ctx); err != nil {
+		return err
+	}
+	jobs, err := e.Spec.ScheduledJobs()
+	if err != nil {
+		return err
+	}
+	names := make([]string, 0, len(jobs))
+	for _, job := range jobs {
+		names = append(names, job.Name)
+	}
+	detail := strings.Join(names, ",")
+	if detail == "" {
+		detail = "none"
+	}
+
+	epoch, err := e.AcquireLock(ctx, operationID, e.Opts.ForceLock)
+	if err != nil {
+		return err
+	}
+	defer e.ReleaseLock(ctx)
+	if len(jobs) > 0 {
+		current, readErr := e.T.Run(ctx, "test -f "+q(e.names().CurrentLink()+"/compose.yaml")+" && echo ok || true")
+		if readErr != nil {
+			return readErr
+		}
+		if strings.TrimSpace(current.Stdout) != "ok" {
+			return errors.New("cannot apply schedules before the first release; deploy the application first")
+		}
+	}
+	if err := e.WriteFence(ctx, operationID, epoch); err != nil {
+		return err
+	}
+
+	jw := &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,
+	}
+	if err := jw.Append(ctx, journal.Record{Phase: "schedule-apply", Event: "start", Detail: detail}); err != nil {
+		return fmt.Errorf("journal schedule apply start: %w", err)
+	}
+	defer func() {
+		finish := journal.Record{Phase: "schedule-apply", Event: "finish", Status: "ok"}
+		if err != nil {
+			finish.Status = "fail"
+			finish.Detail = err.Error()
+		}
+		if journalErr := jw.Append(ctx, finish); journalErr != nil {
+			err = errors.Join(err, fmt.Errorf("journal schedule apply finish: %w", journalErr))
+		}
+	}()
+
+	return e.SyncSchedules(ctx)
+}

internal/engine/schedule_test.go

@@ -47,6 +47,125 @@ func TestSyncSchedulesRetainsManualScheduledJob(t *testing.T) {
 	}
 }
 
+// A package upgrade cannot mutate an agentless target. ScheduleApply is the
+// explicit bridge from a unit written by an older runner to the current unit
+// contract, without requiring an unrelated release deploy.
+func TestScheduleApplyUpgradesLegacyUnitsUnderRegime(t *testing.T) {
+	cfg := testConfig()
+	cfg.Workloads["nightly"] = app.Workload{
+		Role: app.RoleJob, When: "manual", DataEffect: "none",
+		Schedule: &app.JobSchedule{Cron: "0 2 * * *", Timezone: "UTC", Timeout: "45m", CatchUp: false},
+	}
+	f := happyFake()
+	base := f.Dynamic
+	f.Dynamic = func(cmd string) (transport.Result, bool) {
+		switch {
+		case strings.Contains(cmd, "current/compose.yaml"):
+			return transport.Result{Stdout: "ok\n"}, true
+		case strings.Contains(cmd, "list-unit-files"):
+			// v2026.8.5 installed this timer, but its service had no bounded
+			// runner or failure notifier. Presence must not make apply skip it.
+			return transport.Result{Stdout: "ob-sample-nightly.timer\n"}, true
+		case strings.Contains(cmd, "systemd-analyze calendar"):
+			return transport.Result{Stdout: "ok\n"}, true
+		case strings.Contains(cmd, "command -v flock"):
+			return transport.Result{Stdout: "ok\n"}, true
+		}
+		return base(cmd)
+	}
+	e := New(cfg, testProject(t), f, Options{
+		Out: &bytes.Buffer{}, Sleep: noSleep, Environment: "production",
+	})
+	if err := e.ScheduleApply(context.Background(), "R9-schedule-apply"); err != nil {
+		t.Fatalf("schedule apply: %v\n%s", err, strings.Join(f.Commands, "\n"))
+	}
+	seq := strings.Join(f.Commands, "\n")
+	for _, want := range []string{
+		`"phase":"schedule-apply","event":"start"`,
+		"systemctl enable --now ob-sample-nightly.timer",
+		`"phase":"schedule-apply","event":"finish","status":"ok"`,
+		"rm -f '/var/lib/ob/sample/lock'",
+	} {
+		if !strings.Contains(seq, want) {
+			t.Errorf("schedule apply is missing %q:\n%s", want, seq)
+		}
+	}
+	artifacts := strings.Join(f.Inputs, "\n")
+	for _, want := range []string{
+		"ExecStart=/bin/sh /etc/systemd/system/ob-sample-nightly.run",
+		"ExecStopPost=/bin/sh /etc/systemd/system/ob-sample-nightly.notify",
+		"TimeoutStartSec=45m",
+		"flock --exclusive --nonblock",
+		"Persistent=false",
+	} {
+		if !strings.Contains(artifacts, want) {
+			t.Errorf("upgraded artifacts are missing %q:\n%s", want, artifacts)
+		}
+	}
+	for _, command := range f.Commands {
+		if strings.Contains(command, "/etc/systemd/system/ob-sample-nightly") &&
+			strings.Contains(command, ".ob-tmp") && !strings.Contains(command, "ob-fenced") {
+			t.Errorf("schedule artifact write escaped the fence: %s", command)
+		}
+	}
+}
+
+func TestScheduleApplyRefusesBeforeFirstRelease(t *testing.T) {
+	cfg := testConfig()
+	cfg.Workloads["nightly"] = app.Workload{
+		Role: app.RoleJob, When: "manual", DataEffect: "none",
+		Schedule: &app.JobSchedule{Cron: "0 2 * * *", Timezone: "UTC", Timeout: "1h", CatchUp: true},
+	}
+	f := happyFake() // its current release has no Compose runtime
+	base := f.Dynamic
+	f.Dynamic = func(cmd string) (transport.Result, bool) {
+		if strings.Contains(cmd, "command -v flock") {
+			return transport.Result{Stdout: "ok\n"}, true
+		}
+		return base(cmd)
+	}
+	e := New(cfg, testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep, Environment: "production"})
+	err := e.ScheduleApply(context.Background(), "R9-schedule-apply")
+	if err == nil || !strings.Contains(err.Error(), "deploy the application first") {
+		t.Fatalf("error = %v, want first-release refusal", err)
+	}
+	seq := strings.Join(f.Commands, "\n")
+	if len(f.Inputs) != 0 || !strings.Contains(seq, "rm -f '/var/lib/ob/sample/lock'") {
+		t.Fatalf("schedule apply wrote units or leaked its lock before refusing:\n%s", seq)
+	}
+}
+
+func TestScheduleApplyStopsBeforeUnitWritesWhenJournalStartFails(t *testing.T) {
+	cfg := testConfig()
+	cfg.Workloads["nightly"] = app.Workload{
+		Role: app.RoleJob, When: "manual", DataEffect: "none",
+		Schedule: &app.JobSchedule{Cron: "0 2 * * *", Timezone: "UTC", Timeout: "1h", CatchUp: true},
+	}
+	f := happyFake()
+	base := f.Dynamic
+	f.Dynamic = func(cmd string) (transport.Result, bool) {
+		switch {
+		case strings.Contains(cmd, "current/compose.yaml"):
+			return transport.Result{Stdout: "ok\n"}, true
+		case strings.Contains(cmd, "command -v flock"):
+			return transport.Result{Stdout: "ok\n"}, true
+		case strings.Contains(cmd, `"phase":"schedule-apply","event":"start"`):
+			return transport.Result{ExitCode: 74, Stderr: "journal is read-only"}, true
+		}
+		return base(cmd)
+	}
+	e := New(cfg, testProject(t), f, Options{Out: &bytes.Buffer{}, Sleep: noSleep, Environment: "production"})
+	err := e.ScheduleApply(context.Background(), "R9-schedule-apply")
+	if err == nil || !strings.Contains(err.Error(), "journal schedule apply start") {
+		t.Fatalf("error = %v, want journal refusal", err)
+	}
+	for _, command := range f.Commands {
+		if strings.Contains(command, "/etc/systemd/system/ob-sample-nightly") {
+			t.Fatalf("unit mutation followed failed journal start: %s", command)
+		}
+	}
+}
+
 func TestScheduledJobUnitContract(t *testing.T) {
 	job := app.ScheduledJob{
 		Name: "nightly", Cron: "0 2 * * *", Timezone: "UTC",

internal/onebox/binding.go

@@ -31,7 +31,7 @@ func (s *Service) ResolveExecutionBinding(ctx context.Context, kind OperationKin
 func operationUsesInspectionRuntime(kind OperationKind) bool {
 	switch kind {
 	case KindResume, KindAbort, KindRollback, KindBootstrap, KindServiceApply,
-		KindProxyApply, KindSecretsPush, KindDestroy,
+		KindProxyApply, KindScheduleApply, KindSecretsPush, KindDestroy,
 		// Backup operates on a service's data, never on the application's
 		// release images, so a placeholder image must not stop a backup.
 		KindBackupEnable, KindBackupDisable, KindBackupCreate, KindBackupPrune, KindAssuranceCheck,

internal/onebox/execute.go

@@ -204,6 +204,9 @@ func (s *Service) Execute(ctx context.Context, request ExecuteRequest) (Operatio
 	case KindProxyApply:
 		result.EvidenceID = operationID
 		err = e.ProxyApply(ctx, operationID)
+	case KindScheduleApply:
+		result.EvidenceID = operationID
+		err = e.ScheduleApply(ctx, operationID)
 	case KindSecretsPush:
 		entries := encryptedEntries(lp.resolved)
 		externalProjections := externalConnectionProjections(lp.resolved)

internal/onebox/operation_types.go

@@ -30,16 +30,17 @@ type OperationKind string
 type OperationStatus string
 
 const (
-	KindDeploy       OperationKind = "deploy"
-	KindResume       OperationKind = "resume"
-	KindAbort        OperationKind = "abort"
-	KindRollback     OperationKind = "rollback"
-	KindBootstrap    OperationKind = "bootstrap"
-	KindServiceApply OperationKind = "service_apply"
-	KindProxyApply   OperationKind = "proxy_apply"
-	KindSecretsPush  OperationKind = "secrets_push"
-	KindDestroy      OperationKind = "destroy"
-	KindJobRun       OperationKind = "job_run"
+	KindDeploy        OperationKind = "deploy"
+	KindResume        OperationKind = "resume"
+	KindAbort         OperationKind = "abort"
+	KindRollback      OperationKind = "rollback"
+	KindBootstrap     OperationKind = "bootstrap"
+	KindServiceApply  OperationKind = "service_apply"
+	KindProxyApply    OperationKind = "proxy_apply"
+	KindScheduleApply OperationKind = "schedule_apply"
+	KindSecretsPush   OperationKind = "secrets_push"
+	KindDestroy       OperationKind = "destroy"
+	KindJobRun        OperationKind = "job_run"
 
 	KindServiceImagePatch OperationKind = "service_image_patch"
 	KindBackupEnable      OperationKind = "backup_enable"
@@ -448,7 +449,7 @@ func requireJSONEOF(decoder *json.Decoder) error {
 func validOperationKind(kind OperationKind) bool {
 	switch kind {
 	case KindDeploy, KindResume, KindAbort, KindRollback, KindBootstrap, KindJobRun,
-		KindServiceApply, KindProxyApply, KindSecretsPush, KindDestroy,
+		KindServiceApply, KindProxyApply, KindScheduleApply, KindSecretsPush, KindDestroy,
 		KindServiceImagePatch, KindBackupEnable, KindBackupDisable,
 		KindBackupCreate, KindBackupPrune, KindReplayArchive,
 		KindRestoreTest, KindRestorePrepare, KindRestoreCutover, KindRestoreAbort,

site/src/content/docs/guides/schedule-a-job.mdx

@@ -56,6 +56,23 @@ The target must provide `flock` (part of `util-linux` on supported Linux hosts).
 Onebox refuses to install or run schedules when that serialization primitive is
 missing.
 
+## Reconcile after upgrading Onebox
+
+The `ob` binary is agentless: upgrading it on the operator workstation does not
+silently connect to or mutate a target. Existing timers therefore keep the unit
+content written by the previous runner until the next deploy or an explicit
+schedule apply:
+
+```sh
+ob schedule apply
+```
+
+This rewrites every declared timer, service, runner, and failure notifier from
+the current contract, and removes units for jobs no longer declared. It runs
+under the same application lock, schedule mutex, fence, and journal boundary as
+a deploy, but it does not stage or activate a release. Run it after upgrading
+Onebox when an application may not be deployed again soon.
+
 ## Failures remain visible
 
 systemd retains the last oneshot result. `ob status` reads it alongside the