fix(db,reconcile,docker): plugin-name guard, plugin-dir drift, docker stream funnel, StartMarks (D-0602, D-0644, D-0711, D-0589)

.github/coverage-floors.txt

@@ -131,6 +131,7 @@ internal/plugin/scaffold 64  # measured 66.3 2026-10-05
 internal/plugin/verify 85  # measured 87.2 2026-10-05
 internal/plugin/verify/signature 95  # measured 99.1 2026-10-13 (P7-PLUG-23)
 internal/plugin/verify/signature/signaturetest 0  # measured 0.0 2026-10-13; test-only key-table swap helper, exercised from other packages' tests (P7-PLUG-23)
+internal/pluginname 85  # measured 90.0 2026-10-26 (D-0602)
 internal/portable 85  # measured 87.5 2026-10-05
 internal/ports 9  # measured 11.5 2026-10-05
 internal/postgres 49  # measured 51.4 2026-10-05

cmd/commands/backup_credentials_test.go

@@ -53,6 +53,9 @@ func TestBackupStreamProdMinioCredentials(t *testing.T) {
 
 func prodBackupMinioProject(t *testing.T) {
 	t.Helper()
+	// The loader exports this .env into the process; keep it from leaking into
+	// later v1.5 remote-deploy tests, which refuse process-only config (D-0275).
+	restoreProcessEnv(t)
 	for _, key := range []string{"ENV", "MINIO_ENABLED", "STORAGE_ENABLED", "MINIO_ROOT_USER", "MINIO_ROOT_PASSWORD", "NSELF_V15", "IMAGE_PINNING"} {
 		t.Setenv(key, "")
 	}

cmd/commands/db_hasura.go

@@ -90,11 +90,14 @@ func runDBHasuraMetadataReload(cmd *cobra.Command, _ []string) error {
 }
 
 func runDBVerifyChecksums(cmd *cobra.Command, _ []string) error {
+	plugin, _ := cmd.Flags().GetString("plugin")
+	if err := validatePluginFlag(plugin); err != nil {
+		return err
+	}
 	cfg, err := loadProjectConfig()
 	if err != nil {
 		return err
 	}
-	plugin, _ := cmd.Flags().GetString("plugin")
 	mismatches, err := database.VerifyChecksums(cmd.Context(), cfg, plugin)
 	if err != nil {
 		return fmt.Errorf("verify checksums: %w", err)

cmd/commands/db_verify_checksums_plugin_test.go

@@ -0,0 +1,37 @@
+package commands
+
+// Purpose: `db verify-checksums --plugin` runs the same plugin-name grammar as
+// `db migrate` (D-0602): a backup directory or a reserved name is refused
+// before any project config or database is touched.
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/nself-org/cli/internal/errs"
+	"github.com/spf13/cobra"
+)
+
+func TestDBVerifyChecksumsRefusesNonPluginNames(t *testing.T) {
+	// The config loads fine from an empty dir and database.VerifyChecksums
+	// re-checks the name (wrapped as "verify checksums: ..."), so E401 alone
+	// does not prove the guard. Only the guard returns the error unwrapped,
+	// before the config load and the database layer are reached.
+	t.Chdir(t.TempDir())
+	for _, bad := range []string{"notify.prev", "bin", "common", "status", "../x", "a/b"} {
+		cmd := &cobra.Command{Use: "verify-checksums"}
+		cmd.Flags().String("plugin", "", "")
+		if err := cmd.Flags().Set("plugin", bad); err != nil {
+			t.Fatal(err)
+		}
+		cmd.SetContext(context.Background())
+		err := runDBVerifyChecksums(cmd, nil)
+		if d := errs.Describe(err); d == nil || d.Code != "E401" {
+			t.Errorf("--plugin %q: err = %v, want E401", bad, err)
+		}
+		if err != nil && strings.Contains(err.Error(), "verify checksums") {
+			t.Errorf("--plugin %q: err = %v came from the database layer, want the guard's error", bad, err)
+		}
+	}
+}

cmd/commands/deploy_env_resolve_test.go

@@ -87,6 +87,7 @@ func TestReadEnvFileOverrides_ParsesAndStripsQuotes(t *testing.T) {
 // a stale dev-tier value that no longer matches what was baked into
 // docker-compose.yml.
 func TestWriteResolvedDeployEnv_LaterFilesOverrideEarlier(t *testing.T) {
+	isolateDeployConfigEnv(t)
 	dir := t.TempDir()
 	if err := os.WriteFile(filepath.Join(dir, ".env"), []byte("POSTGRES_DB=dev_db\nSHARED_KEY=from_dev\n"), 0o600); err != nil {
 		t.Fatalf("write .env: %v", err)
@@ -128,6 +129,7 @@ func TestWriteResolvedDeployEnv_LaterFilesOverrideEarlier(t *testing.T) {
 // function actually deletes the temp snapshot so repeated deploys don't
 // accumulate stray files in the project root.
 func TestWriteResolvedDeployEnv_CleanupRemovesFile(t *testing.T) {
+	isolateDeployConfigEnv(t)
 	dir := t.TempDir()
 	path, cleanup, err := writeResolvedDeployEnv(dir, "prod")
 	if err != nil {

cmd/commands/deploy_env_scope_test.go

@@ -167,6 +167,7 @@ func scopeFixture(t *testing.T, unreachable bool) (string, *scopeProber) {
 // leaks into a fixture and a cascade load never leaks out.
 func isolateDeployEnv(t *testing.T) {
 	t.Helper()
+	isolateDeployConfigEnv(t)
 	for _, kv := range os.Environ() {
 		if k, _, ok := strings.Cut(kv, "="); ok && strings.HasPrefix(k, "NSELF_DEPLOY_HOST_") {
 			t.Setenv(k, "")

cmd/commands/deploy_follow.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/nself-org/cli/internal/controlplane"
+	"github.com/nself-org/cli/internal/docker"
 	"github.com/nself-org/cli/internal/ui"
 	"github.com/nself-org/cli/sdk/go/v2/remote"
 	"github.com/spf13/cobra"
@@ -31,11 +32,8 @@ func followDeployLogs(cmd *cobra.Command, workdir, target, strategy string, star
 	host := legacyDeployHost(target)
 	if host == "" {
 		// Local host: stream logs directly.
-		lc := exec.CommandContext(cmd.Context(), "docker", "compose", "logs", "-f", "--tail=50")
-		lc.Dir = workdir
-		lc.Stdout = os.Stdout
-		lc.Stderr = os.Stderr
-		_ = lc.Run()
+		// Ctrl-C and a non-zero exit of the tail are not errors.
+		_ = docker.Stream(cmd.Context(), workdir, []string{"compose", "logs", "-f", "--tail=50"}, os.Stdout, os.Stderr)
 		return nil
 	}
 	// Remote follow: tail logs on the remote host via SSH.

cmd/commands/deploy_scratch_nginx_test.go

@@ -53,6 +53,7 @@ func stdoutOf(t *testing.T, fn func()) string {
 // nginxProject is an operator project with marked, unmarked and conf.d nginx files.
 func nginxProject(t *testing.T, env string) string {
 	t.Helper()
+	isolateDeployConfigEnv(t)
 	root := t.TempDir()
 	wf(t, filepath.Join(root, ".env"), "PROJECT_NAME=demo\n"+env)
 	wf(t, filepath.Join(root, "nginx", "sites", "stale.conf"), testBuildMarker+"# stale generated output\n")

cmd/commands/doctor_drift_plugindir_test.go

@@ -0,0 +1,43 @@
+package commands
+
+// Purpose: the standalone doctor drift check resolves "@plugins/" record keys
+// under the plugin directory the project's env cascade names (D-0644).
+// Constraints: temp dirs only; the shell leaves NSELF_PLUGIN_DIR unset.
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/nself-org/cli/internal/reconcile"
+)
+
+func TestDoctorDriftPluginDirOnlyInProjectEnv(t *testing.T) {
+	isolateDoctorHome(t)
+	t.Setenv("NSELF_PLUGIN_DIR", "")
+	dir, plugins := t.TempDir(), t.TempDir()
+	env := "PROJECT_NAME=fx\nENV=dev\nBASE_DOMAIN=example.test\nSSL_MODE=none\nNSELF_PLUGIN_DIR=" + plugins + "\n"
+	if err := os.WriteFile(filepath.Join(dir, ".env"), []byte(env), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	const key = reconcile.PluginsPrefix + "fxplug/docker-compose.plugin.yml"
+	fragment := filepath.Join(plugins, "fxplug", "docker-compose.plugin.yml")
+	if err := os.MkdirAll(filepath.Dir(fragment), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(fragment, []byte("services: {}\n# hand edit\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	sum := sha256.Sum256([]byte("services: {}\n")) // the bytes the record says were written
+	st := &reconcile.GeneratedState{Files: map[string]string{key: hex.EncodeToString(sum[:])}}
+	if err := st.Save(dir); err != nil {
+		t.Fatal(err)
+	}
+	r, _, err := checkGeneratedDrift(context.Background(), dir)
+	if err != nil || r.Status != "fail" || r.Message != "drift: "+key+" (hand-edited)" {
+		t.Fatalf("hand-edited fragment under the .env plugin dir: %+v, %v", r, err)
+	}
+}

cmd/commands/functions_logs.go

@@ -11,12 +11,10 @@ package commands
 // functionsCmd (parent) remain in functions.go.
 
 import (
-	"bufio"
 	"fmt"
-	"io"
-	"os/exec"
 	"strings"
 
+	"github.com/nself-org/cli/internal/docker"
 	"github.com/spf13/cobra"
 )
 
@@ -68,33 +66,12 @@ func runFunctionsLogs(cmd *cobra.Command, args []string) error {
 	dockerArgs = append(dockerArgs, fmt.Sprintf("--tail=%d", functionsLogsFlags.tail))
 	dockerArgs = append(dockerArgs, containerID)
 
-	logsCmd := exec.CommandContext(cmd.Context(), "docker", dockerArgs...)
-	stdout, err := logsCmd.StdoutPipe()
-	if err != nil {
-		return fmt.Errorf("pipe: %w", err)
-	}
-	stderr, err := logsCmd.StderrPipe()
-	if err != nil {
-		return fmt.Errorf("stderr pipe: %w", err)
-	}
-
-	if err := logsCmd.Start(); err != nil {
-		return fmt.Errorf("starting docker logs: %w", err)
-	}
-
-	// Filter lines containing the function name and print.
-	filterAndPrint := func(r io.Reader) {
-		scanner := bufio.NewScanner(r)
-		for scanner.Scan() {
-			line := scanner.Text()
-			if strings.Contains(line, name) || strings.Contains(line, "/v1/"+name) {
-				fmt.Println(line)
-			}
+	// Both docker streams carry function output; keep the lines that mention
+	// the function.
+	return docker.StreamLines(cmd.Context(), "", dockerArgs, nil, true, maxLogLineBytes, func(line string) error {
+		if strings.Contains(line, name) || strings.Contains(line, "/v1/"+name) {
+			fmt.Println(line)
 		}
-	}
-
-	go filterAndPrint(stderr)
-	filterAndPrint(stdout)
-
-	return logsCmd.Wait()
+		return nil
+	})
 }

cmd/commands/logs_stream.go

@@ -12,13 +12,14 @@ package commands
 // docker spawn of the logs stream (human, legacy JSON and v1.5 JSON paths).
 
 import (
-	"bufio"
 	"context"
 	"fmt"
 	"io"
 	"os"
 	"os/exec"
 	"strings"
+
+	"github.com/nself-org/cli/internal/docker"
 )
 
 // streamFilteredLogs runs docker with composeArgs and streams filtered/formatted output to w.
@@ -44,56 +45,13 @@ const maxLogLineBytes = 4 << 20
 // error from fn stops the stream, kills docker and is returned. A cancelled
 // parent context ends the stream without an error.
 func streamLogLines(parent context.Context, workdir string, composeArgs []string, opts LogsOptions, fn func(line string) error) error {
-	ctx, cancel := context.WithCancel(parent)
-	defer cancel()
-	dockerCmd := exec.CommandContext(ctx, "docker", composeArgs...)
-	dockerCmd.Dir = workdir
-	dockerCmd.Stderr = os.Stderr
-
-	if opts.Follow {
-		dockerCmd.Stdin = os.Stdin
-	}
-
-	stdout, err := dockerCmd.StdoutPipe()
-	if err != nil {
-		return fmt.Errorf("creating stdout pipe: %w", err)
-	}
-
-	if err := dockerCmd.Start(); err != nil {
-		return fmt.Errorf("starting docker: %w", err)
-	}
-
-	var fnErr error
-	scanner := bufio.NewScanner(stdout)
-	scanner.Buffer(make([]byte, 0, 64*1024), maxLogLineBytes)
-	for scanner.Scan() {
-		filtered, ok := filterLogLine(scanner.Text(), opts)
+	return docker.StreamLines(parent, workdir, composeArgs, os.Stderr, false, maxLogLineBytes, func(line string) error {
+		filtered, ok := filterLogLine(line, opts)
 		if !ok {
-			continue
-		}
-		if fnErr = fn(filtered); fnErr != nil {
-			cancel()
-			break
+			return nil
 		}
-	}
-
-	// A read error (a line over maxLogLineBytes) must not pass for the end of
-	// the logs: stop docker so Wait cannot block on a full pipe, and report it.
-	if fnErr == nil {
-		if serr := scanner.Err(); serr != nil {
-			cancel()
-			fnErr = fmt.Errorf("reading docker output: %w", serr)
-		}
-	}
-
-	waitErr := dockerCmd.Wait()
-	if fnErr != nil {
-		return fnErr
-	}
-	if parent.Err() != nil {
-		return nil
-	}
-	return waitErr
+		return fn(filtered)
+	})
 }
 
 // getRunningServices returns the list of service names from docker compose ps --services.

cmd/commands/plugin_logs.go

@@ -7,6 +7,7 @@ import (
 	"regexp"
 	"strings"
 
+	"github.com/nself-org/cli/internal/docker"
 	"github.com/spf13/cobra"
 )
 
@@ -66,13 +67,7 @@ func runPluginLogs(cmd *cobra.Command, args []string) error {
 
 	if grepPattern == "" {
 		// Simple passthrough.
-		dockerCmd := exec.Command("docker", dockerArgs...)
-		dockerCmd.Stdout = os.Stdout
-		dockerCmd.Stderr = os.Stderr
-		if err := dockerCmd.Run(); err != nil {
-			return fmt.Errorf("docker logs: %w", err)
-		}
-		return nil
+		return docker.Stream(cmd.Context(), "", dockerArgs, os.Stdout, os.Stderr)
 	}
 
 	// Validate the grep pattern before use so we get a helpful error on invalid regex.
@@ -81,34 +76,33 @@ func runPluginLogs(cmd *cobra.Command, args []string) error {
 	}
 
 	// With grep: pipe docker logs output through grep.
-	dockerCmd := exec.Command("docker", dockerArgs...)
 	grepCmd := exec.Command("grep", "-E", grepPattern)
-
-	dockerOut, err := dockerCmd.StdoutPipe()
+	grepIn, err := grepCmd.StdinPipe()
 	if err != nil {
-		return fmt.Errorf("creating docker stdout pipe: %w", err)
+		return fmt.Errorf("creating grep stdin pipe: %w", err)
 	}
-	dockerCmd.Stderr = os.Stderr
-	grepCmd.Stdin = dockerOut
 	grepCmd.Stdout = os.Stdout
 	grepCmd.Stderr = os.Stderr
-
-	if err := dockerCmd.Start(); err != nil {
-		return fmt.Errorf("starting docker logs: %w", err)
-	}
 	if err := grepCmd.Start(); err != nil {
 		return fmt.Errorf("starting grep: %w", err)
 	}
 
 	// Wait for docker to finish, then grep will EOF.
-	_ = dockerCmd.Wait()
+	streamErr := docker.Stream(cmd.Context(), "", dockerArgs, grepIn, os.Stderr)
+	_ = grepIn.Close()
 	// grep exits 1 on "no match" (normal) and 2+ on error. Only surface errors.
-	if err := grepCmd.Wait(); err != nil {
-		if exitErr, ok := err.(*exec.ExitError); ok && exitErr.ExitCode() == 1 {
+	waitErr := grepCmd.Wait()
+	// A docker that failed to start or exited non-zero is the root cause; it
+	// outranks grep's "no match" exit.
+	if streamErr != nil {
+		return streamErr
+	}
+	if waitErr != nil {
+		if exitErr, ok := waitErr.(*exec.ExitError); ok && exitErr.ExitCode() == 1 {
 			// No lines matched — not an error.
 			return nil
 		}
-		return fmt.Errorf("grep error: %w", err)
+		return fmt.Errorf("grep error: %w", waitErr)
 	}
 	return nil
 }