From f2725eea7e855e9d16772d3bd5e8c27b0995f203 Mon Sep 17 00:00:00 2001 From: acmore Date: Sun, 5 Jul 2026 16:50:18 +0800 Subject: [PATCH 1/2] feat(sync): per-path sync direction (spec.sync.paths[].direction: bi|up|down) A single bidirectional folder carries both inputs (code, local->pod) and outputs (results, pod->local), so a bad local write can clobber pod-generated data. Finer per-subpath direction inside one folder is impossible in Syncthing (.stignore lives in the directory; per-path folder types were rejected upstream), so direction is a per-mapping (per-folder) setting. Config: spec.sync.paths entries now accept a structured form alongside the compact "local:remote" string (which stays fully backward compatible and marshals back unchanged): paths: - local: . remote: /workspace direction: down # bi (default) | up | down up = local sendonly / pod receiveonly (code push-only; nothing on the pod can alter local files). down = pod sendonly / local receiveonly (the pod is the authority; a local write can never clobber pod results). The schema is per-path from day one so lifting the current single-mapping limit later will not change config shape. Plumbing: syncengine.Pair carries the resolved direction; `okdev sync --mode` defaults to the configured path's direction with an explicit flag winning; `okdev up` starts sync in that direction and skips the two-phase bootstrap for directional folders (one side is already sendonly, so the final folder types are safe from the first start). Direction participates in the sync config hash (explicit bi hashes like the legacy default, so upgrades do not restart existing sessions). `up --reset-workspace` and `sync reset-remote` are rejected under "down": they reseed the remote from local, which would sync back as deletions over the pod's data. `status --details` shows the direction. Co-Authored-By: Claude Fable 5 --- docs/command-reference.md | 1 + docs/config-manifest.md | 25 ++++++- internal/cli/status_details.go | 9 +++ internal/cli/status_details_test.go | 4 +- internal/cli/sync.go | 33 ++++++++- internal/cli/sync_test.go | 41 +++++++++++ internal/cli/syncthing.go | 16 ++++- internal/cli/up.go | 29 ++++++-- internal/cli/up_ui_test.go | 22 ++++-- internal/config/config.go | 104 ++++++++++++++++++++++++---- internal/config/config_test.go | 73 ++++++++++++++++++- internal/config/snapshot_test.go | 4 +- internal/sync/sync.go | 17 +++-- internal/sync/sync_test.go | 20 +++++- skills/okdev/references/multipod.md | 2 + 15 files changed, 357 insertions(+), 43 deletions(-) diff --git a/docs/command-reference.md b/docs/command-reference.md index ac6b4daa..c1170f46 100644 --- a/docs/command-reference.md +++ b/docs/command-reference.md @@ -304,6 +304,7 @@ agents can react without launching a diagnostic chain on every blip: ### `okdev sync [--mode up|down|bi] [--foreground] [--reset] [--dry-run]` - Advanced command. Starts detached background sync by default; use `--foreground` for sync debugging, or explicit one-way sync (`up`/`down`). +- `--mode` defaults to the configured path's `direction` (`spec.sync.paths[].direction`, falling back to `bi`); an explicit flag overrides it for that invocation. See the config manifest for choosing a persistent direction — `down` makes the pod the authority so local writes can never clobber pod-generated results. - For default `--mode bi`, no-op when background sync is already active for the session. - `--background`: explicitly request detached background mode. - `--reset`: check local-to-hub sync and mesh receiver health, then reset only what is broken. Skips the local sync teardown when the primary sync is already healthy. For sessions with mesh receivers, probes each receiver and re-runs mesh setup only when broken or disconnected receivers are found. diff --git a/docs/config-manifest.md b/docs/config-manifest.md index b3aafc04..7f79eff1 100644 --- a/docs/config-manifest.md +++ b/docs/config-manifest.md @@ -233,7 +233,8 @@ spec: | Field | Type | Default | Description | |-------|------|---------|-------------| | `engine` | `string` | — | Sync engine (currently only `syncthing`) | -| `paths` | `[]string` | — | Mappings in `local:remote` format (max 1 entry) | +| `paths` | `[]entry` | — | Sync mappings (max 1 entry). Each entry is either the compact string `local:remote`, or a mapping `{local, remote, direction}` with an optional per-path `direction` | +| `paths[].direction` | `string` | `bi` | Sync direction for this mapping's folder: `bi` (bidirectional), `up` (local is the authority: local sendonly, pod receiveonly), `down` (pod is the authority: pod sendonly, local receiveonly) | | `syncthing.version` | `string` | `v1.29.7` | Local Syncthing binary version | | `syncthing.autoInstall` | `bool` | `true` | Auto-install local Syncthing | | `syncthing.image` | `string` | `ghcr.io/acmore/okdev:` | Sidecar image (fallback: `edge`) | @@ -242,7 +243,27 @@ spec: | `syncthing.compression` | `bool` | `false` | Use Syncthing `always` compression for peer connections instead of the default `metadata` mode | | `syncthing.versioningDays` | `int` | `30` | Keep files overwritten or deleted by sync in `.stversions` for this many days (staggered versioning, both sides); `0` disables | -**Validation:** `engine` must be `syncthing`; each `paths[]` entry must be `local:remote`; `versioningDays` must be `>= 0`. +**Validation:** `engine` must be `syncthing`; each `paths[]` entry must have non-empty local and remote; `direction` must be `bi`, `up`, or `down`; `versioningDays` must be `>= 0`. + +**Sync direction.** `direction` applies to a mapping's whole synced folder — finer per-subpath direction inside one folder is impossible with Syncthing (ignore rules live in the directory itself; per-path folder types were rejected upstream). Pick the direction by what the folder carries: + +- `bi` (default): source code you edit on both sides. Either side can overwrite the other, so **keep generated outputs (results, checkpoints, logs) out of the synced path** — write them to a non-synced location and fetch with `okdev cp` or `okdev jobs logs`. Versioning (`.stversions/`) is the safety net, not the design. +- `up`: the folder is a code drop. Nothing the pod writes can ever touch local files; pod-side writes into the folder are flagged by Syncthing but not propagated. +- `down`: the folder carries pod-generated results. A stray local write (for example an empty file from a failed redirect) can never clobber the pod's data. `okdev up --reset-workspace` and `okdev sync reset-remote` are rejected in this direction because they reseed the remote from local. + +An explicit `okdev sync --mode` overrides the configured direction for that invocation. Changing `direction` takes effect on the next `okdev up` (it restarts sync with the new folder types). Mesh receiver pods are always receiveonly regardless of direction. + +```yaml +# compact form (direction defaults to bi): +paths: + - .:/workspace + +# structured form with an explicit direction: +paths: + - local: . + remote: /workspace + direction: up +``` Local ignore rules come from the synced workspace's `.stignore`. `okdev init` writes a starter `.stignore` for built-in templates, and `okdev up` creates one with default patterns if the local sync root does not already have one. Editing `.stignore` takes effect automatically as Syncthing notices the change, but it does not remove files that were already synced to the remote workspace. For faster initial syncs, consider ignoring large generated build outputs or local test artifacts such as `debug/`, `release/`, caches, and dataset directories when they do not need to exist remotely. diff --git a/internal/cli/status_details.go b/internal/cli/status_details.go index 33ad8ff7..3d2e09bb 100644 --- a/internal/cli/status_details.go +++ b/internal/cli/status_details.go @@ -114,6 +114,7 @@ type detailedStatusSSH struct { type detailedStatusSync struct { Engine string `json:"engine"` + Direction string `json:"direction,omitempty"` Health string `json:"health,omitempty"` HealthDetail string `json:"healthDetail,omitempty"` ConfiguredPaths []string `json:"configuredPaths,omitempty"` @@ -420,8 +421,13 @@ func buildDetailedSync(sessionName string, cfg *config.DevEnvironment, cfgPath s if engine == "" { engine = "syncthing" } + direction := "" + if pairs, err := syncengine.ParsePairs(cfg.Spec.Sync.Paths, cfg.EffectiveWorkspaceMountPath(cfgPath)); err == nil && len(pairs) > 0 { + direction = pairs[0].Direction + } detail := detailedStatusSync{ Engine: engine, + Direction: direction, ConfiguredPaths: summarizeConfiguredSyncPaths(cfg, cfgPath), } if engine != "syncthing" { @@ -593,6 +599,9 @@ func printDetailedStatus(w io.Writer, detail detailedStatus) { fmt.Fprintln(w, "\nSync:") fmt.Fprintf(w, "- engine: %s\n", detail.Sync.Engine) + if detail.Sync.Direction != "" { + fmt.Fprintf(w, "- direction: %s (%s)\n", detail.Sync.Direction, modeSymbol(detail.Sync.Direction)) + } if detail.Sync.BackgroundStatus != "" { status := detail.Sync.BackgroundStatus if detail.Sync.BackgroundPID > 0 { diff --git a/internal/cli/status_details_test.go b/internal/cli/status_details_test.go index fec4db61..10043044 100644 --- a/internal/cli/status_details_test.go +++ b/internal/cli/status_details_test.go @@ -65,7 +65,7 @@ func TestGatherDetailedStatusIncludesDiagnostics(t *testing.T) { if err := os.WriteFile(conflictPath, []byte("conflict"), 0o644); err != nil { t.Fatalf("write conflict file: %v", err) } - cfg.Spec.Sync.Paths = []string{syncRoot + ":/workspace"} + cfg.Spec.Sync.Paths = []config.SyncPathSpec{{Local: syncRoot, Remote: "/workspace"}} cfg.Spec.Ports = []config.PortMapping{{Name: "http", Local: 8080, Remote: 80}} cfg.Spec.Agents = []config.AgentSpec{{Name: "codex", Auth: &config.AgentAuth{LocalPath: "~/.codex/auth.json"}}} @@ -347,6 +347,7 @@ func TestPrintDetailedStatusIncludesSections(t *testing.T) { }, Sync: detailedStatusSync{ Engine: "syncthing", + Direction: "down", BackgroundStatus: "running", BackgroundPID: 123, ConfiguredPaths: []string{"/tmp/repo -> /workspace"}, @@ -391,6 +392,7 @@ func TestPrintDetailedStatusIncludesSections(t *testing.T) { "SSH:", "managed forward: running", "Sync:", + "direction: down (<-)", "background: running (pid 123)", "conflicts: 2", "./a.sync-conflict-1", diff --git a/internal/cli/sync.go b/internal/cli/sync.go index 33a4ba35..ba67efcf 100644 --- a/internal/cli/sync.go +++ b/internal/cli/sync.go @@ -16,6 +16,7 @@ import ( "syscall" "time" + "github.com/acmore/okdev/internal/config" "github.com/acmore/okdev/internal/kube" "github.com/acmore/okdev/internal/logx" "github.com/acmore/okdev/internal/session" @@ -88,6 +89,10 @@ func newSyncCmd(opts *Options) *cobra.Command { if err != nil { return err } + mode = resolveSyncMode(cmd.Flags().Changed("mode"), mode, pairs[0].Direction) + if err := validateSyncMode(mode); err != nil { + return err + } if dryRun { fmt.Fprintf(cmd.OutOrStdout(), "DRY RUN: sync session=%s namespace=%s engine=%s mode=%s\n", cc.sessionName, cc.namespace, engine, mode) fmt.Fprintf(cmd.OutOrStdout(), "- paths: %v\n", pairs) @@ -166,7 +171,7 @@ func newSyncCmd(opts *Options) *cobra.Command { }, } - cmd.Flags().StringVar(&mode, "mode", "bi", "Sync mode: up|down|bi") + cmd.Flags().StringVar(&mode, "mode", "bi", "Sync mode: up|down|bi (default: spec.sync.direction, or bi)") cmd.Flags().BoolVar(&background, "background", true, "Run syncthing sync as a detached background process") cmd.Flags().BoolVar(&foreground, "foreground", false, "Run syncthing sync in the foreground for troubleshooting") cmd.Flags().BoolVar(&reset, "reset", false, "Check sync health and reset broken components") @@ -179,6 +184,26 @@ func newSyncCmd(opts *Options) *cobra.Command { return cmd } +// resolveSyncMode picks the effective sync mode: an explicit --mode always +// wins; otherwise the configured spec.sync.direction applies. +func resolveSyncMode(flagSet bool, flagMode, configDirection string) string { + if flagSet { + return flagMode + } + return configDirection +} + +// validateSyncMode accepts the documented modes plus the internal +// "two-phase" bootstrap mode that `okdev up` spawns for fresh sessions. +func validateSyncMode(mode string) error { + switch mode { + case "up", "down", "bi", "two-phase": + return nil + default: + return fmt.Errorf("invalid sync mode %q: must be one of up, down, bi", mode) + } +} + func newSyncResetRemoteCmd(opts *Options) *cobra.Command { return &cobra.Command{ Use: "reset-remote", @@ -204,6 +229,12 @@ synced workspace subtree is cleared.`, if len(pairs) != 1 { return fmt.Errorf("reset-remote requires exactly one sync path mapping, got %d", len(pairs)) } + if pairs[0].Direction == config.SyncDirectionDown { + // reset-remote clears the remote and reseeds it from local; + // with direction "down" the pod is the authority and local + // is receiveonly — the wipe would sync back as deletions. + return fmt.Errorf("sync reset-remote conflicts with sync direction \"down\" (the pod is the sync authority)") + } target, err := resolveTargetRef(cmd.Context(), cc.opts, cc.cfg, cc.namespace, cc.sessionName, cc.kube) if err != nil { diff --git a/internal/cli/sync_test.go b/internal/cli/sync_test.go index e9f07c67..7a56aa0c 100644 --- a/internal/cli/sync_test.go +++ b/internal/cli/sync_test.go @@ -417,6 +417,47 @@ func TestSyncthingSessionConfigHashTracksSyncSettings(t *testing.T) { } } +func TestSyncthingSessionConfigHashTracksDirection(t *testing.T) { + cfg := &config.DevEnvironment{} + cfg.SetDefaults() + base := syncthingSessionConfigHash(cfg, "/tmp/local", "/workspace") + + // Explicit "bi" hashes identically to the legacy empty value so + // upgrading okdev (or spelling out the default) does not force a sync + // restart on existing sessions. + cfg.Spec.Sync.Paths = []config.SyncPathSpec{{Local: ".", Remote: "/workspace", Direction: "bi"}} + if got := syncthingSessionConfigHash(cfg, "/tmp/local", "/workspace"); got != base { + t.Fatal("explicit bi direction must hash like the legacy default") + } + + cfg.Spec.Sync.Paths = []config.SyncPathSpec{{Local: ".", Remote: "/workspace", Direction: "down"}} + if got := syncthingSessionConfigHash(cfg, "/tmp/local", "/workspace"); got == base { + t.Fatal("expected direction change to affect sync config hash") + } +} + +func TestResolveSyncMode(t *testing.T) { + // Explicit --mode wins over the configured direction. + if got := resolveSyncMode(true, "up", "down"); got != "up" { + t.Fatalf("explicit flag must win, got %q", got) + } + // Without the flag, spec.sync.direction applies. + if got := resolveSyncMode(false, "bi", "down"); got != "down" { + t.Fatalf("config direction must apply, got %q", got) + } +} + +func TestValidateSyncMode(t *testing.T) { + for _, ok := range []string{"up", "down", "bi", "two-phase"} { + if err := validateSyncMode(ok); err != nil { + t.Fatalf("mode %q should validate, got %v", ok, err) + } + } + if err := validateSyncMode("push"); err == nil { + t.Fatal("expected invalid mode to be rejected") + } +} + func TestProcStatIsZombie(t *testing.T) { if !procStatIsZombie([]byte("123 (okdev) Z 1 2 3")) { t.Fatal("expected zombie proc stat to be detected") diff --git a/internal/cli/syncthing.go b/internal/cli/syncthing.go index a93d5e54..9895ef9f 100644 --- a/internal/cli/syncthing.go +++ b/internal/cli/syncthing.go @@ -2465,9 +2465,14 @@ func resetRemoteWorkspace(ctx context.Context, k interface { } type syncthingSessionConfigState struct { - Engine string `json:"engine"` - LocalPath string `json:"localPath"` - RemotePath string `json:"remotePath"` + Engine string `json:"engine"` + LocalPath string `json:"localPath"` + RemotePath string `json:"remotePath"` + // Direction is part of the hash so editing spec.sync.direction makes + // the next `okdev up` restart sync with the new folder types. The + // default "bi" is hashed as the empty legacy value to avoid a + // restart on upgrade for existing sessions. + Direction string `json:"direction,omitempty"` WatcherDelaySeconds int `json:"watcherDelaySeconds,omitempty"` RescanIntervalSeconds int `json:"rescanIntervalSeconds,omitempty"` RelaysEnabled bool `json:"relaysEnabled,omitempty"` @@ -2479,10 +2484,15 @@ func syncthingSessionConfigHash(cfg *config.DevEnvironment, localPath, remotePat if cfg == nil { return "" } + direction := "" + if pairs, err := syncengine.ParsePairs(cfg.Spec.Sync.Paths, remotePath); err == nil && len(pairs) > 0 && pairs[0].Direction != config.SyncDirectionBi { + direction = pairs[0].Direction + } state := syncthingSessionConfigState{ Engine: strings.TrimSpace(cfg.Spec.Sync.Engine), LocalPath: localPath, RemotePath: remotePath, + Direction: direction, WatcherDelaySeconds: cfg.Spec.Sync.Syncthing.WatcherDelaySeconds, RescanIntervalSeconds: cfg.Spec.Sync.Syncthing.RescanIntervalSeconds, RelaysEnabled: cfg.Spec.Sync.Syncthing.RelaysEnabled, diff --git a/internal/cli/up.go b/internal/cli/up.go index b349353f..fe24abbe 100644 --- a/internal/cli/up.go +++ b/internal/cli/up.go @@ -175,6 +175,13 @@ func upValidate(cmd *cobra.Command, opts *Options, flags upOptions) (*upState, e if err != nil { return nil, err } + if flags.resetWorkspace && len(syncPairs) > 0 && syncPairs[0].Direction == config.SyncDirectionDown { + // reset-workspace clears the remote and reseeds it from local; with + // direction "down" the pod is the authority and local is + // receiveonly, so the combination would wipe the remote and then + // pull the deletions back over the local files. + return nil, fmt.Errorf("--reset-workspace conflicts with sync direction \"down\" (the pod is the sync authority); remove the flag or switch direction") + } runtime, err := sessionRuntime(cc.cfg, cc.cfgPath, cc.sessionName, workloadName, labels, annotations, cc.cfg.Spec.PodTemplate.Spec, volumes, enableTmux, resolvePreStopCommand(cc.cfg, cc.cfgPath)) if err != nil { return nil, err @@ -1092,12 +1099,18 @@ func upSetupSync(state *upState, target workload.TargetRef) (string, string, str } configChanged = changed } + direction := config.SyncDirectionBi + if len(state.syncPairs) > 0 { + direction = state.syncPairs[0].Direction + } active := syncthingSessionActive(state.command.sessionName) restartRequired := state.flags.resetWorkspace || configChanged || !active - startMode = syncStartMode(state.flags.resetWorkspace, targetReset, hasConfigState, configChanged, active) + startMode = syncStartMode(direction, state.flags.resetWorkspace, targetReset, hasConfigState, configChanged, active) // Signal waitForInitialSync to probe the remote folder type for an - // incomplete bootstrap, reusing the port-forward it already opens. - bootstrapResume = len(state.syncPairs) == 1 && hasConfigState && !state.flags.resetWorkspace && !targetReset + // incomplete bootstrap, reusing the port-forward it already opens. Only + // meaningful for bi: directional folders never run the two-phase + // bootstrap, so there is nothing to resume. + bootstrapResume = len(state.syncPairs) == 1 && hasConfigState && !state.flags.resetWorkspace && !targetReset && direction == config.SyncDirectionBi if restartRequired { if err := refreshSyncthingSessionProcesses(state.command.sessionName); err != nil { return "", "", "", false, fmt.Errorf("refresh local syncthing session state: %w", err) @@ -1832,7 +1845,15 @@ func modeSymbol(mode string) string { } } -func syncStartMode(resetWorkspace, targetReset, hasConfigState, configChanged, active bool) string { +func syncStartMode(direction string, resetWorkspace, targetReset, hasConfigState, configChanged, active bool) string { + // A directional folder never needs the two-phase bootstrap: that + // protocol exists to protect the remote from the local side's initial + // emptiness before flipping both sides to sendreceive. With up/down one + // side is already sendonly, so the final folder types are safe from the + // first start. + if direction == config.SyncDirectionUp || direction == config.SyncDirectionDown { + return direction + } switch { case resetWorkspace: return "two-phase" diff --git a/internal/cli/up_ui_test.go b/internal/cli/up_ui_test.go index e8826189..6bceb91e 100644 --- a/internal/cli/up_ui_test.go +++ b/internal/cli/up_ui_test.go @@ -119,6 +119,7 @@ func TestSyncHelpers(t *testing.T) { func TestSyncStartMode(t *testing.T) { cases := []struct { name string + direction string resetWorkspace bool targetReset bool hasConfigState bool @@ -126,16 +127,23 @@ func TestSyncStartMode(t *testing.T) { active bool want string }{ - {name: "first bootstrap", hasConfigState: false, active: false, want: "two-phase"}, - {name: "explicit reset", resetWorkspace: true, hasConfigState: true, configChanged: true, active: true, want: "two-phase"}, - {name: "target recreated", targetReset: true, hasConfigState: true, active: false, want: "two-phase"}, - {name: "config changed on active session", hasConfigState: true, configChanged: true, active: true, want: "bi"}, - {name: "daemon recovery", hasConfigState: true, active: false, want: "bi"}, - {name: "steady state", hasConfigState: true, active: true, want: "bi"}, + {name: "first bootstrap", direction: "bi", hasConfigState: false, active: false, want: "two-phase"}, + {name: "explicit reset", direction: "bi", resetWorkspace: true, hasConfigState: true, configChanged: true, active: true, want: "two-phase"}, + {name: "target recreated", direction: "bi", targetReset: true, hasConfigState: true, active: false, want: "two-phase"}, + {name: "config changed on active session", direction: "bi", hasConfigState: true, configChanged: true, active: true, want: "bi"}, + {name: "daemon recovery", direction: "bi", hasConfigState: true, active: false, want: "bi"}, + {name: "steady state", direction: "bi", hasConfigState: true, active: true, want: "bi"}, + // Directional folders skip the two-phase bootstrap entirely: one + // side is already sendonly, so the final types are safe from the + // first start. + {name: "up direction fresh", direction: "up", hasConfigState: false, active: false, want: "up"}, + {name: "up direction target reset", direction: "up", targetReset: true, hasConfigState: true, active: false, want: "up"}, + {name: "down direction fresh", direction: "down", hasConfigState: false, active: false, want: "down"}, + {name: "down direction steady", direction: "down", hasConfigState: true, active: true, want: "down"}, } for _, tc := range cases { - if got := syncStartMode(tc.resetWorkspace, tc.targetReset, tc.hasConfigState, tc.configChanged, tc.active); got != tc.want { + if got := syncStartMode(tc.direction, tc.resetWorkspace, tc.targetReset, tc.hasConfigState, tc.configChanged, tc.active); got != tc.want { t.Fatalf("%s: got %q want %q", tc.name, got, tc.want) } } diff --git a/internal/config/config.go b/internal/config/config.go index 5340a4e1..fab5dbb5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,6 +1,7 @@ package config import ( + "encoding/json" "errors" "fmt" "os" @@ -140,11 +141,91 @@ type MetadataMap struct { Labels map[string]string `yaml:"labels"` } +// Sync directions; see SyncPathSpec.Direction. +const ( + SyncDirectionBi = "bi" + SyncDirectionUp = "up" + SyncDirectionDown = "down" +) + type SyncSpec struct { - Paths []string `yaml:"paths"` - PreservePaths []string `yaml:"preservePaths"` - Engine string `yaml:"engine"` - Syncthing SyncthingSpec `yaml:"syncthing"` + Paths []SyncPathSpec `yaml:"paths"` + PreservePaths []string `yaml:"preservePaths"` + Engine string `yaml:"engine"` + Syncthing SyncthingSpec `yaml:"syncthing"` +} + +// SyncPathSpec is one local<->remote sync mapping. In YAML it accepts either +// the compact string form "local:remote" or a mapping with an optional +// per-path direction: +// +// paths: +// - .:/workspace +// - local: ./results +// remote: /data/results +// direction: down +// +// Direction sets the syncthing folder types for this mapping's folder: +// +// bi — both sides sendreceive (default; either side can overwrite the +// other — keep generated outputs out of the synced path) +// up — local sendonly, pod receiveonly: code push-only; nothing on the +// pod can alter local files +// down — local receiveonly, pod sendonly: the pod is the authority; a +// local write can never clobber pod-side results +// +// Direction is per-folder by necessity: syncthing stores .stignore in the +// directory itself, so per-path direction inside one folder is impossible +// (per-path folder types were rejected upstream, syncthing/syncthing#8061). +// Mesh receivers stay receiveonly regardless. +type SyncPathSpec struct { + Local string `json:"local"` + Remote string `json:"remote"` + Direction string `json:"direction,omitempty"` +} + +// UnmarshalJSON accepts the compact "local:remote" string form alongside the +// structured mapping form. (Config decoding goes through sigs.k8s.io/yaml, +// which converts YAML to JSON first.) +func (p *SyncPathSpec) UnmarshalJSON(data []byte) error { + var compact string + if err := json.Unmarshal(data, &compact); err == nil { + parts := strings.Split(compact, ":") + if len(parts) != 2 { + return fmt.Errorf("sync path entry %q must be local:remote", compact) + } + p.Local = strings.TrimSpace(parts[0]) + p.Remote = strings.TrimSpace(parts[1]) + p.Direction = "" + return nil + } + type syncPathSpecAlias SyncPathSpec + var alias syncPathSpecAlias + if err := json.Unmarshal(data, &alias); err != nil { + return err + } + *p = SyncPathSpec(alias) + return nil +} + +// MarshalJSON keeps the compact string form for plain bidirectional +// mappings so round-tripped configs stay stable. +func (p SyncPathSpec) MarshalJSON() ([]byte, error) { + if strings.TrimSpace(p.Direction) == "" { + return json.Marshal(p.Local + ":" + p.Remote) + } + type syncPathSpecAlias SyncPathSpec + return json.Marshal(syncPathSpecAlias(p)) +} + +// EffectiveDirection resolves the mapping's direction, defaulting to +// bidirectional. +func (p SyncPathSpec) EffectiveDirection() string { + d := strings.TrimSpace(p.Direction) + if d == "" { + return SyncDirectionBi + } + return d } type SyncthingSpec struct { @@ -593,16 +674,15 @@ func workspaceMountPathForContainer(containers []corev1.Container, name string) return "" } -func validateSyncPaths(paths []string) error { +func validateSyncPaths(paths []SyncPathSpec) error { for _, p := range paths { - parts := strings.Split(p, ":") - if len(parts) != 2 { - return fmt.Errorf("spec.sync.paths entry %q must be local:remote", p) + if strings.TrimSpace(p.Local) == "" || strings.TrimSpace(p.Remote) == "" { + return fmt.Errorf("spec.sync.paths entry %q must have non-empty local and remote", p.Local+":"+p.Remote) } - local := strings.TrimSpace(parts[0]) - remote := strings.TrimSpace(parts[1]) - if local == "" || remote == "" { - return fmt.Errorf("spec.sync.paths entry %q must have non-empty local and remote", p) + switch strings.TrimSpace(p.Direction) { + case "", SyncDirectionBi, SyncDirectionUp, SyncDirectionDown: + default: + return fmt.Errorf("spec.sync.paths entry %q direction must be one of %s, %s, %s", p.Local+":"+p.Remote, SyncDirectionBi, SyncDirectionUp, SyncDirectionDown) } } return nil diff --git a/internal/config/config_test.go b/internal/config/config_test.go index c080e12d..87adc648 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -9,6 +9,7 @@ import ( "github.com/acmore/okdev/internal/version" "gopkg.in/yaml.v3" corev1 "k8s.io/api/core/v1" + sigsyaml "sigs.k8s.io/yaml" ) func validConfig() *DevEnvironment { @@ -240,6 +241,74 @@ func TestValidateRejectsInvalidEngine(t *testing.T) { } } +func TestValidateSyncPathDirection(t *testing.T) { + for _, valid := range []string{"", "bi", "up", "down"} { + cfg := validConfig() + cfg.Spec.Sync.Paths = []SyncPathSpec{{Local: ".", Remote: "/workspace", Direction: valid}} + cfg.SetDefaults() + if err := cfg.Validate(); err != nil { + t.Fatalf("direction %q should be valid, got %v", valid, err) + } + } + for _, invalid := range []string{"push", "pull", "sendonly", "both"} { + cfg := validConfig() + cfg.Spec.Sync.Paths = []SyncPathSpec{{Local: ".", Remote: "/workspace", Direction: invalid}} + cfg.SetDefaults() + if err := cfg.Validate(); err == nil { + t.Fatalf("direction %q should be rejected", invalid) + } + } +} + +func TestSyncPathSpecEffectiveDirection(t *testing.T) { + var p SyncPathSpec + if got := p.EffectiveDirection(); got != SyncDirectionBi { + t.Fatalf("expected default bi, got %q", got) + } + p.Direction = " down " + if got := p.EffectiveDirection(); got != SyncDirectionDown { + t.Fatalf("expected trimmed down, got %q", got) + } +} + +func TestSyncPathSpecYAMLForms(t *testing.T) { + // Compact string form and structured form must both decode; the string + // form keeps direction empty (= bi). + var spec SyncSpec + raw := []byte("paths:\n - .:/workspace\n - local: ./results\n remote: /data/results\n direction: down\n") + if err := sigsyaml.Unmarshal(raw, &spec); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(spec.Paths) != 2 { + t.Fatalf("expected 2 paths, got %+v", spec.Paths) + } + if spec.Paths[0].Local != "." || spec.Paths[0].Remote != "/workspace" || spec.Paths[0].Direction != "" { + t.Fatalf("unexpected compact entry: %+v", spec.Paths[0]) + } + if spec.Paths[1].Local != "./results" || spec.Paths[1].Remote != "/data/results" || spec.Paths[1].Direction != "down" { + t.Fatalf("unexpected structured entry: %+v", spec.Paths[1]) + } + + // Round-trip: plain mappings marshal back to the compact string form, + // directional ones keep the structured form. + out, err := sigsyaml.Marshal(spec.Paths) + if err != nil { + t.Fatalf("marshal: %v", err) + } + text := string(out) + if !strings.Contains(text, ".:/workspace") { + t.Fatalf("expected compact form in output, got %q", text) + } + if !strings.Contains(text, "direction: down") { + t.Fatalf("expected structured form for directional entry, got %q", text) + } + + // Malformed compact entries fail decoding. + if err := sigsyaml.Unmarshal([]byte("paths:\n - nocolon\n"), &spec); err == nil { + t.Fatal("expected error for entry without colon") + } +} + func TestValidateAllowsInterPodSSHToOverrideDisabledSidecars(t *testing.T) { cfg := validConfig() cfg.Spec.Workload.Type = "pytorchjob" @@ -347,7 +416,7 @@ func TestValidateAcceptsAgentLocalPathWithTilde(t *testing.T) { func TestValidateRejectsInvalidSyncPath(t *testing.T) { cfg := validConfig() - cfg.Spec.Sync.Paths = []string{"./local-only"} + cfg.Spec.Sync.Paths = []SyncPathSpec{{Local: "./local-only", Remote: ""}} if err := cfg.Validate(); err == nil { t.Fatal("expected validation error") } @@ -355,7 +424,7 @@ func TestValidateRejectsInvalidSyncPath(t *testing.T) { func TestValidateRejectsSyncthingMultiplePaths(t *testing.T) { cfg := validConfig() - cfg.Spec.Sync.Paths = []string{"./a:/workspace/a", "./b:/workspace/b"} + cfg.Spec.Sync.Paths = []SyncPathSpec{{Local: "./a", Remote: "/workspace/a"}, {Local: "./b", Remote: "/workspace/b"}} if err := cfg.Validate(); err == nil { t.Fatal("expected validation error") } diff --git a/internal/config/snapshot_test.go b/internal/config/snapshot_test.go index e84b438a..a15d75e5 100644 --- a/internal/config/snapshot_test.go +++ b/internal/config/snapshot_test.go @@ -91,7 +91,7 @@ func TestBuildWorkloadSnapshotExcludesNonWorkloadFields(t *testing.T) { Sidecar: SidecarSpec{Image: "img:1"}, Ports: []PortMapping{{Name: "http", Local: 8080, Remote: 80}}, SSH: SSHSpec{User: "alice"}, - Sync: SyncSpec{Paths: []string{"."}}, + Sync: SyncSpec{Paths: []SyncPathSpec{{Local: ".", Remote: "/workspace"}}}, }, } cfg2 := &DevEnvironment{ @@ -100,7 +100,7 @@ func TestBuildWorkloadSnapshotExcludesNonWorkloadFields(t *testing.T) { Sidecar: SidecarSpec{Image: "img:1"}, Ports: []PortMapping{{Name: "grpc", Local: 9090, Remote: 90}}, SSH: SSHSpec{User: "bob"}, - Sync: SyncSpec{Paths: []string{"src/"}}, + Sync: SyncSpec{Paths: []SyncPathSpec{{Local: "src/", Remote: "/workspace"}}}, }, } snap1 := BuildWorkloadSnapshot(cfg1, "/workspace", "dev", false, "", "", "", "") diff --git a/internal/sync/sync.go b/internal/sync/sync.go index f6502c51..0db01eae 100644 --- a/internal/sync/sync.go +++ b/internal/sync/sync.go @@ -4,25 +4,30 @@ import ( "fmt" "strings" + "github.com/acmore/okdev/internal/config" "github.com/acmore/okdev/internal/shellutil" ) type Pair struct { Local string Remote string + // Direction is the resolved sync direction for this mapping's folder + // (bi/up/down, never empty; see config.SyncPathSpec). + Direction string } -func ParsePairs(configured []string, defaultRemote string) ([]Pair, error) { +func ParsePairs(configured []config.SyncPathSpec, defaultRemote string) ([]Pair, error) { if len(configured) == 0 { - return []Pair{{Local: ".", Remote: defaultRemote}}, nil + return []Pair{{Local: ".", Remote: defaultRemote, Direction: config.SyncDirectionBi}}, nil } out := make([]Pair, 0, len(configured)) for _, item := range configured { - parts := strings.Split(item, ":") - if len(parts) != 2 { - return nil, fmt.Errorf("invalid sync path mapping %q, expected local:remote", item) + local := strings.TrimSpace(item.Local) + remote := strings.TrimSpace(item.Remote) + if local == "" || remote == "" { + return nil, fmt.Errorf("invalid sync path mapping %q, expected local:remote", item.Local+":"+item.Remote) } - out = append(out, Pair{Local: strings.TrimSpace(parts[0]), Remote: strings.TrimSpace(parts[1])}) + out = append(out, Pair{Local: local, Remote: remote, Direction: item.EffectiveDirection()}) } return out, nil } diff --git a/internal/sync/sync_test.go b/internal/sync/sync_test.go index 3a04ea87..a3a2a21a 100644 --- a/internal/sync/sync_test.go +++ b/internal/sync/sync_test.go @@ -1,6 +1,10 @@ package sync -import "testing" +import ( + "testing" + + "github.com/acmore/okdev/internal/config" +) func TestParsePairsDefault(t *testing.T) { pairs, err := ParsePairs(nil, "/workspace") @@ -10,20 +14,30 @@ func TestParsePairsDefault(t *testing.T) { if len(pairs) != 1 || pairs[0].Local != "." || pairs[0].Remote != "/workspace" { t.Fatalf("unexpected pairs: %+v", pairs) } + if pairs[0].Direction != config.SyncDirectionBi { + t.Fatalf("expected default bi direction, got %q", pairs[0].Direction) + } } func TestParsePairsConfigured(t *testing.T) { - pairs, err := ParsePairs([]string{".:/workspace", "./cfg:/etc/app"}, "/workspace") + pairs, err := ParsePairs([]config.SyncPathSpec{ + {Local: ".", Remote: "/workspace"}, + {Local: "./cfg", Remote: "/etc/app", Direction: "down"}, + }, "/workspace") if err != nil { t.Fatal(err) } if len(pairs) != 2 { t.Fatalf("unexpected pair count: %d", len(pairs)) } + // Direction is normalized to the effective value, never empty. + if pairs[0].Direction != config.SyncDirectionBi || pairs[1].Direction != config.SyncDirectionDown { + t.Fatalf("unexpected directions: %+v", pairs) + } } func TestParsePairsInvalid(t *testing.T) { - if _, err := ParsePairs([]string{"invalid"}, "/workspace"); err == nil { + if _, err := ParsePairs([]config.SyncPathSpec{{Local: "", Remote: "/workspace"}}, "/workspace"); err == nil { t.Fatal("expected error") } } diff --git a/skills/okdev/references/multipod.md b/skills/okdev/references/multipod.md index 36e5f99d..7db0ba6c 100644 --- a/skills/okdev/references/multipod.md +++ b/skills/okdev/references/multipod.md @@ -42,6 +42,8 @@ Multi-pod sync behavior depends on workload layout: Do not assume all pods independently sync from the local machine in the same way. Check the workload pattern first. +Direction: `spec.sync.paths[].direction: bi|up|down` sets the folder direction on the local↔hub edge for that mapping (`down` = pod is the authority, local writes can never clobber pod results). It applies to the mapping's whole folder — finer per-subpath direction is impossible in Syncthing. Mesh receiver pods are always receiveonly regardless. Generated outputs belong outside the synced path either way; fetch them with `okdev cp` or `okdev jobs logs`. + ## PyTorchJob For PyTorchJob questions, keep these points in mind: From c32acc652414f3145fa6dd3063be10c336459950 Mon Sep 17 00:00:00 2001 From: acmore Date: Sun, 5 Jul 2026 18:19:31 +0800 Subject: [PATCH 2/2] test(e2e): cover per-path sync direction semantics in kind smoke Four scenarios on the live smoke session: rewriting the sync path entry to the structured form with direction "up" takes effect on the next `okdev up` (config-hash driven restart, verified via status --details); up-direction still pushes local edits while pod-side writes never flow back; down-direction returns pod-generated results while a stray local truncation cannot clobber the pod's copy (the reported data-loss incident, reproduced and asserted harmless); and `up --reset-workspace` is rejected under "down" with the remote file left intact. Co-Authored-By: Claude Fable 5 --- scripts/e2e_kind_smoke.sh | 106 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/scripts/e2e_kind_smoke.sh b/scripts/e2e_kind_smoke.sh index bb428fd2..bf6dfde6 100644 --- a/scripts/e2e_kind_smoke.sh +++ b/scripts/e2e_kind_smoke.sh @@ -496,6 +496,112 @@ if [[ "$RECONCILED_IMAGE" != "ubuntu:24.04" ]]; then fi echo "Pod reconcile verified" +# --- Sync direction: per-path direction semantics on a live session -------- +# Covers the direction chain end-to-end: editing spec.sync.paths[].direction +# restarts sync with new folder types on the next `okdev up` (config-hash +# driven), `up` enforces one-way local->pod, `down` makes the pod the +# authority (a stray local write cannot clobber pod results — the NCCL +# data-loss reproduction), and reset-workspace is rejected under `down`. + +echo "Switching sync direction to up" +DIRECTION_UP_ENTRY="- local: \"$SYNC_DIR\" + remote: /workspace + direction: up" +replace_first_in_file "$CFG_PATH" "- \"$SYNC_DIR:/workspace\"" "$DIRECTION_UP_ENTRY" +if ! grep -q "direction: up" "$CFG_PATH"; then + echo "ERROR: failed to rewrite sync path entry to structured form; config is:" >&2 + cat "$CFG_PATH" >&2 + exit 1 +fi +"$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" up --wait-timeout 5m +DETAILS_UP=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" status --details) +if [[ "$DETAILS_UP" != *"direction: up (->)"* ]]; then + echo "ERROR: expected status --details to show direction up, got:" >&2 + echo "$DETAILS_UP" >&2 + exit 1 +fi + +echo "up-direction: local edits still reach the pod" +echo "up-probe-content" > "$SYNC_DIR/up-probe.txt" +UP_SYNC_OK=false +for i in $(seq 1 30); do + REMOTE_UP=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'cat /workspace/up-probe.txt 2>/dev/null' || true) + if [[ "$REMOTE_UP" == "up-probe-content" ]]; then + UP_SYNC_OK=true + break + fi + sleep 2 +done +if [[ "$UP_SYNC_OK" != true ]]; then + echo "ERROR: local file did not reach pod in up direction" >&2 + exit 1 +fi + +echo "up-direction: pod-side writes must not flow back to local" +"$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'echo intruder > /workspace/pod-write.txt' +sleep 15 +if [[ -f "$SYNC_DIR/pod-write.txt" ]]; then + echo "ERROR: pod-side write leaked back to local in up direction" >&2 + exit 1 +fi +echo "sync direction up verified" + +echo "Switching sync direction to down" +replace_first_in_file "$CFG_PATH" "direction: up" "direction: down" +"$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" up --wait-timeout 5m +DETAILS_DOWN=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" status --details) +if [[ "$DETAILS_DOWN" != *"direction: down (<-)"* ]]; then + echo "ERROR: expected status --details to show direction down, got:" >&2 + echo "$DETAILS_DOWN" >&2 + exit 1 +fi + +echo "down-direction: pod-generated results flow back to local" +"$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'echo result-data > /workspace/result.txt' +DOWN_SYNC_OK=false +for i in $(seq 1 30); do + if [[ "$(cat "$SYNC_DIR/result.txt" 2>/dev/null)" == "result-data" ]]; then + DOWN_SYNC_OK=true + break + fi + sleep 2 +done +if [[ "$DOWN_SYNC_OK" != true ]]; then + echo "ERROR: pod result did not flow back to local in down direction" >&2 + exit 1 +fi + +echo "down-direction: a stray local write cannot clobber the pod's result" +# Reproduces the reported incident: a failed local redirect truncates the +# synced copy. With the pod as authority the empty file must never +# propagate; the pod-side result stays intact. +: > "$SYNC_DIR/result.txt" +sleep 15 +REMOTE_RESULT=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'cat /workspace/result.txt') +if [[ "$REMOTE_RESULT" != "result-data" ]]; then + echo "ERROR: pod-side result was clobbered by a local write in down direction, got '$REMOTE_RESULT'" >&2 + exit 1 +fi + +echo "down-direction: --reset-workspace is rejected" +set +e +RESET_OUTPUT=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" up --reset-workspace --wait-timeout 5m 2>&1) +RESET_STATUS=$? +set -e +if [[ "$RESET_STATUS" -eq 0 ]] || [[ "$RESET_OUTPUT" != *"conflicts with sync direction"* ]]; then + echo "ERROR: expected --reset-workspace to be rejected under direction down (status=$RESET_STATUS)" >&2 + echo "$RESET_OUTPUT" >&2 + exit 1 +fi +REMOTE_RESULT_AFTER=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'cat /workspace/result.txt') +if [[ "$REMOTE_RESULT_AFTER" != "result-data" ]]; then + echo "ERROR: remote result damaged by rejected reset-workspace, got '$REMOTE_RESULT_AFTER'" >&2 + exit 1 +fi +echo "sync direction down verified" + +echo "Sync direction tests completed" + # --- Detached jobs: runtime-volume logs, process-group stop, sidecar ------- # fallback. Covers the detach chain end-to-end on a real pod: logs land on # the shared /var/okdev volume, `jobs stop` kills the whole process tree