fix(cli): unify teardown drain into a single bounded flush mode

cmd/relayfile-cli/main.go

@@ -6410,6 +6410,7 @@ func runMount(args []string) error {
 	logFileFlag := fs.String("log-file", "", "log file path for background mode")
 	daemonized := fs.Bool("daemonized", false, "internal flag used by relayfile mount --background")
 	once := fs.Bool("once", false, "run one sync cycle and exit")
+	flushOutboxOnce := fs.Bool("flush-outbox-once", false, "ingest watcher-journaled local drafts, flush the durable outbox, and exit without scanning or reconciling the mirror")
 	resetAfterClobber := fs.Bool("reset-after-clobber", boolEnv("RELAYFILE_RESET_AFTER_CLOBBER", false), "acknowledge a mount-root clobber and authorize daemon to recreate the directory")
 	rehome := fs.Bool("rehome", false, "allow re-homing an already-registered workspace mirror to a different LOCAL_DIR")
 	if err := fs.Parse(normalizeFlagArgs(args, map[string]bool{
@@ -6439,6 +6440,7 @@ func runMount(args []string) error {
 		"log-file":            true,
 		"daemonized":          false,
 		"once":                false,
+		"flush-outbox-once":   false,
 		"reset-after-clobber": false,
 		"rehome":              false,
 		"local-dir":           true,
@@ -6835,7 +6837,7 @@ func runMount(args []string) error {
 		}
 		return spawnBackgroundMountProcessFn(args, resolvedRemotePaths, absLocalDir, pidFile, logFile, resolvedLocalLayout)
 	}
-	registerPID := shouldRegisterMountPID(*daemonized, *once)
+	registerPID := shouldRegisterMountPID(*daemonized, *once || *flushOutboxOnce)
 	if *daemonized {
 		if err := rotateLogFile(logFile); err != nil {
 			return err
@@ -6900,6 +6902,14 @@ func runMount(args []string) error {
 		if initialCredExpiresAt != "" {
 			syncer.SetCredentialExpiry(initialCredExpiresAt)
 		}
+		if *flushOutboxOnce {
+			drainCtx, cancel := context.WithTimeout(scopeCtx, *timeout)
+			defer cancel()
+			if err := syncer.FlushOutboxOnce(drainCtx); err != nil {
+				return fmt.Errorf("drain writebacks for %s: %w", scope.RemotePath, err)
+			}
+			return nil
+		}
 		return runMountLoopWithAuthLock(
 			scopeCtx,
 			syncer,
@@ -7095,6 +7105,9 @@ Common flags:
   --interval 30s       sync interval (default 30s)
   --background         detach and keep syncing in the background
   --once               run one sync cycle and exit (used by setup/CI)
+  --flush-outbox-once
+                       bounded teardown: ingest watcher-journaled local drafts,
+                       flush the durable outbox, and exit without a tree scan
   --timeout 5m         per-sync timeout
   --bootstrap-timeout 0s
                        hard cap for initial/full-tree bootstrap (0 = progress-based)

cmd/relayfile-cli/main_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"errors"
@@ -23,6 +24,7 @@ import (
 	"github.com/agentworkforce/relayfile/internal/delegatedauth"
 	"github.com/agentworkforce/relayfile/internal/mountscope"
 	"github.com/agentworkforce/relayfile/internal/mountsync"
+	"github.com/fsnotify/fsnotify"
 )
 
 const relayfileCLITestSubprocessEnv = "RELAYFILE_CLI_TEST_SUBPROCESS"
@@ -2404,6 +2406,121 @@ func TestMountUsesRecordedLocalDirWhenOmitted(t *testing.T) {
 	}
 }
 
+func TestMountFlushOutboxOnceIngestsJournaledDraftWithoutTreeScan(t *testing.T) {
+	t.Setenv("HOME", t.TempDir())
+	clearRelayfileEnv(t)
+
+	const (
+		workspaceID = "ws_bounded_teardown"
+		remoteRoot  = "/slack/channels/C123/messages"
+	)
+	localDir := t.TempDir()
+	stateDir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(localDir, "unrelated.txt"), []byte("must not upload"), 0o644); err != nil {
+		t.Fatalf("seed unrelated file: %v", err)
+	}
+
+	token := testJWTWithWorkspace(workspaceID)
+	var bulkCalls atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		wantPath := remoteRoot + "/command.json"
+		switch {
+		case r.Method == http.MethodPost && r.URL.Path == "/v1/workspaces/"+workspaceID+"/fs/bulk":
+			bulkCalls.Add(1)
+			var req bulkWriteRequest
+			if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+				t.Fatalf("decode bounded drain request: %v", err)
+			}
+			if len(req.Files) != 1 {
+				t.Fatalf("bounded drain files = %+v, want only journaled draft", req.Files)
+			}
+			file := req.Files[0]
+			if file.Path != wantPath || file.Content != `{"text":"final reply"}` {
+				t.Fatalf("bounded drain file = %+v, want path %s and final reply", file, wantPath)
+			}
+			_, _ = io.WriteString(w, `{"written":1,"errorCount":0,"correlationId":"corr_drain","results":[{"path":"`+wantPath+`","revision":"rev_1","opId":"op_drain","writeback":{"provider":"slack","state":"succeeded"}}]}`)
+		case r.Method == http.MethodGet && r.URL.Path == "/v1/workspaces/"+workspaceID+"/ops/op_drain":
+			_, _ = io.WriteString(w, `{"opId":"op_drain","path":"`+wantPath+`","status":"succeeded","revision":"rev_1"}`)
+		default:
+			t.Fatalf("bounded drain made unexpected request: %s %s", r.Method, r.URL.Path)
+		}
+	}))
+	defer server.Close()
+
+	producer, err := mountsync.NewSyncer(
+		mountsync.NewHTTPClient(server.URL, token, server.Client()),
+		mountsync.SyncerOptions{
+			WorkspaceID: workspaceID,
+			RemoteRoot:  remoteRoot,
+			LocalRoot:   localDir,
+			StateDir:    stateDir,
+			MountKind:   mountsync.MountKindDaemon,
+		},
+	)
+	if err != nil {
+		t.Fatalf("create producer syncer: %v", err)
+	}
+	watchCtx, cancelWatch := context.WithCancel(context.Background())
+	watcher, err := producer.NewFileWatcher(func(string, fsnotify.Op) {})
+	if err != nil {
+		t.Fatalf("create producer watcher: %v", err)
+	}
+	if err := watcher.Start(watchCtx); err != nil {
+		t.Fatalf("start producer watcher: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(localDir, "command.json"), []byte(`{"text":"final reply"}`), 0o644); err != nil {
+		t.Fatalf("write final draft: %v", err)
+	}
+	journalDir := filepath.Join(localDir, ".relay", "outbox", "local-pending")
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		entries, readErr := os.ReadDir(journalDir)
+		if readErr == nil && len(entries) == 1 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("watcher did not persist pending draft before teardown: entries=%d err=%v", len(entries), readErr)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	cancelWatch()
+	if err := watcher.Close(); err != nil {
+		t.Fatalf("close producer watcher: %v", err)
+	}
+
+	err = run([]string{
+		"mount", workspaceID, localDir,
+		"--server", server.URL,
+		"--token", token,
+		"--remote-path", remoteRoot,
+		"--state-dir", stateDir,
+		"--mount-kind", mountsync.MountKindDaemon,
+		"--timeout", "2s",
+		"--flush-outbox-once",
+	}, strings.NewReader(""), &bytes.Buffer{}, &bytes.Buffer{})
+	if err != nil {
+		t.Fatalf("run bounded teardown drain: %v", err)
+	}
+	if got := bulkCalls.Load(); got != 1 {
+		t.Fatalf("bulk calls = %d, want 1", got)
+	}
+	entries, err := os.ReadDir(journalDir)
+	if err != nil {
+		t.Fatalf("read drained journal: %v", err)
+	}
+	if len(entries) != 0 {
+		t.Fatalf("pending local journal still has %d entries", len(entries))
+	}
+}
+
+func TestMountRetiresPushLocalOnceSplitFlag(t *testing.T) {
+	err := runMount([]string{"--push-local-once"})
+	if err == nil || !strings.Contains(err.Error(), "flag provided but not defined") {
+		t.Fatalf("retired --push-local-once returned %v", err)
+	}
+}
+
 func TestMountMirrorsRepeatedRemotePathsUnderScopedLayout(t *testing.T) {
 	skipUntilScopedOperatorSurfacesReady(t)
 	t.Setenv("HOME", t.TempDir())

cmd/relayfile-mount/main.go

@@ -68,7 +68,6 @@ type mountConfig struct {
 	scopedChild           bool
 	once                  bool
 	flushOutboxOnce       bool
-	pushLocalOnce         bool
 	mode                  string
 	fuseContentTTL        time.Duration
 }
@@ -112,8 +111,7 @@ func main() {
 	fuse := flag.Bool("fuse", boolEnv("RELAYFILE_MOUNT_FUSE", false), "shortcut for --mode=fuse")
 	fuseContentTTL := flag.Duration("fuse-content-ttl", durationEnv("RELAYFILE_MOUNT_FUSE_CONTENT_TTL", 0), "FUSE in-memory file content cache TTL (default 30s; 0 = use default)")
 	once := flag.Bool("once", false, "run one sync cycle and exit")
-	flushOutboxOnce := flag.Bool("flush-outbox-once", false, "flush durable writeback outbox once and exit without reconciling the local mirror")
-	pushLocalOnce := flag.Bool("push-local-once", false, "ingest pending local writeback drafts (one pushLocal pass) then flush the outbox once and exit; no pullRemote/digest/reconcile — the teardown drain for last-moment drafts")
+	flushOutboxOnce := flag.Bool("flush-outbox-once", false, "ingest watcher-journaled local drafts, flush the durable outbox, and exit; bounded to pending paths with no mirror scan or reconcile")
 	flag.Parse()
 
 	resolvedToken := strings.TrimSpace(*token)
@@ -199,7 +197,6 @@ func main() {
 		scopes:                parseTokenScopes(resolvedToken),
 		once:                  *once,
 		flushOutboxOnce:       *flushOutboxOnce,
-		pushLocalOnce:         *pushLocalOnce,
 		mode:                  resolvedMode,
 		fuseContentTTL:        *fuseContentTTL,
 	}
@@ -433,22 +430,13 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error {
 	if _, err := mountsync.StartDiagnostics(rootCtx, cfg.pprofAddr, cfg.memlogInterval, log.Default()); err != nil {
 		return fmt.Errorf("start diagnostics: %w", err)
 	}
-	if cfg.pushLocalOnce {
-		ctx, cancel := context.WithTimeout(rootCtx, cfg.timeout)
-		defer cancel()
-		if err := syncer.PushLocalAndFlushOnce(ctx); err != nil {
-			return fmt.Errorf("push local and flush once: %w", err)
-		}
-		log.Printf("local push + outbox flush completed")
-		return nil
-	}
 	if cfg.flushOutboxOnce {
 		ctx, cancel := context.WithTimeout(rootCtx, cfg.timeout)
 		defer cancel()
 		if err := syncer.FlushOutboxOnce(ctx); err != nil {
 			return fmt.Errorf("flush outbox once: %w", err)
 		}
-		log.Printf("outbox flush completed")
+		log.Printf("bounded writeback drain completed")
 		return nil
 	}
 	log.Printf("%s", mountStartupLogLine(cfg))