diff --git a/cmd/relayfile-cli/main.go b/cmd/relayfile-cli/main.go index 6f378c80..e2302d5c 100644 --- a/cmd/relayfile-cli/main.go +++ b/cmd/relayfile-cli/main.go @@ -5,6 +5,7 @@ import ( "bytes" "context" "crypto/rand" + "crypto/sha256" "encoding/base64" "encoding/hex" "encoding/json" @@ -20,6 +21,7 @@ import ( "os" "os/exec" "os/signal" + "path" "path/filepath" "runtime" "sort" @@ -32,6 +34,7 @@ import ( "github.com/agentworkforce/relayfile/internal/delegatedauth" "github.com/agentworkforce/relayfile/internal/mountsync" + "github.com/agentworkforce/relayfile/internal/relayfile" "github.com/agentworkforce/relayfile/internal/writeback" "github.com/fsnotify/fsnotify" ) @@ -119,17 +122,31 @@ type bulkWriteRequest struct { } type bulkWriteFile struct { - Path string `json:"path"` - ContentType string `json:"contentType"` - Content string `json:"content"` - Encoding string `json:"encoding,omitempty"` + Path string `json:"path"` + ContentType string `json:"contentType"` + Content string `json:"content"` + Encoding string `json:"encoding,omitempty"` + ContentIdentity *contentIdentity `json:"contentIdentity,omitempty"` } type bulkWriteResponse struct { - Written int `json:"written"` - ErrorCount int `json:"errorCount"` - Errors []bulkWriteError `json:"errors"` - CorrelationID string `json:"correlationId"` + Written int `json:"written"` + ErrorCount int `json:"errorCount"` + Errors []bulkWriteError `json:"errors"` + Results []bulkWriteResult `json:"results,omitempty"` + CorrelationID string `json:"correlationId"` +} + +type bulkWriteResult struct { + Path string `json:"path"` + Revision string `json:"revision"` + ContentType string `json:"contentType,omitempty"` + OpID string `json:"opId,omitempty"` + ContentIdentity *contentIdentity `json:"contentIdentity,omitempty"` + Writeback *struct { + Provider string `json:"provider,omitempty"` + State string `json:"state,omitempty"` + } `json:"writeback,omitempty"` } type bulkWriteError struct { @@ -138,6 +155,12 @@ type bulkWriteError struct { Message string `json:"message"` } +type contentIdentity struct { + Kind string `json:"kind"` + Key string `json:"key"` + TTLSeconds int `json:"ttlSeconds,omitempty"` +} + type syncStatusResponse struct { WorkspaceID string `json:"workspaceId"` Providers []syncProviderStatus `json:"providers"` @@ -291,19 +314,23 @@ type cloudIntegrationListEntry struct { } type syncStateFile struct { - WorkspaceID string `json:"workspaceId"` - RemoteRoot string `json:"remoteRoot,omitempty"` - Mode string `json:"mode"` - LastReconcileAt string `json:"lastReconcileAt,omitempty"` - LastEventAt string `json:"lastEventAt,omitempty"` - IntervalMs int64 `json:"intervalMs"` - Providers []syncStateProvider `json:"providers,omitempty"` - PendingWriteback int `json:"pendingWriteback"` - PendingConflicts int `json:"pendingConflicts"` - DeniedPaths int `json:"deniedPaths"` - FailedWritebacks uint64 `json:"failedWritebacks"` - StallReason string `json:"stallReason,omitempty"` - Daemon *syncStateDaemon `json:"daemon,omitempty"` + WorkspaceID string `json:"workspaceId"` + RemoteRoot string `json:"remoteRoot,omitempty"` + Mode string `json:"mode"` + Status string `json:"status,omitempty"` + LastReconcileAt string `json:"lastReconcileAt,omitempty"` + LastSuccessfulReconcileAt string `json:"lastSuccessfulReconcileAt,omitempty"` + LastEventAt string `json:"lastEventAt,omitempty"` + IntervalMs int64 `json:"intervalMs"` + Providers []syncStateProvider `json:"providers,omitempty"` + PendingWriteback int `json:"pendingWriteback"` + PendingConflicts int `json:"pendingConflicts"` + DeniedPaths int `json:"deniedPaths"` + FailedWritebacks uint64 `json:"failedWritebacks"` + StallReason string `json:"stallReason,omitempty"` + LastError *statusError `json:"lastError,omitempty"` + IncrementalReadNotReadySince map[string]string `json:"incrementalReadNotReadySince,omitempty"` + Daemon *syncStateDaemon `json:"daemon,omitempty"` // Guards surfaces defensive-guard telemetry from the in-process // mountsync state. Existing consumers can ignore the field; it is // additive and uses omitempty. Counters come from .relay/state.json. @@ -340,6 +367,14 @@ type syncStateGuards struct { Circuit *syncStateGuardCirc `json:"circuit,omitempty"` } +type statusError struct { + Kind string `json:"kind"` + StatusCode int `json:"statusCode,omitempty"` + Code string `json:"code,omitempty"` + Message string `json:"message"` + At string `json:"at"` +} + // syncStateGuardCirc is the JSON shape of the cloud-error circuit breaker // state surfaced via status. Mirrors mountsync.CircuitState fields. type syncStateGuardCirc struct { @@ -586,6 +621,8 @@ func printWorkspaceUsage(w io.Writer, subcommand string) { fmt.Fprintln(w, "Usage: relayfile workspace list [--names-only]") case "current": fmt.Fprintln(w, "Usage: relayfile workspace current [--verbose]") + case "status": + fmt.Fprintln(w, "Usage: relayfile workspace status [--workspace NAME] [--json]") case "delete": fmt.Fprintln(w, "Usage: relayfile workspace delete NAME [--yes]") default: @@ -595,6 +632,7 @@ func printWorkspaceUsage(w io.Writer, subcommand string) { relayfile workspace use NAME relayfile workspace list [--names-only] relayfile workspace current [--verbose] + relayfile workspace status [--workspace NAME] [--json] relayfile workspace delete NAME [--yes]`) } } @@ -646,8 +684,12 @@ func printWritebackUsage(w io.Writer, subcommand string) { fmt.Fprintln(w, writebackListUsage) case "status": fmt.Fprintln(w, "Usage: relayfile writeback status [WORKSPACE] [--json]") + case "push": + fmt.Fprintln(w, "Usage: relayfile writeback push LOCAL_PATH [--workspace WS] [--json] [--timeout 90s]") case "retry": fmt.Fprintln(w, "Usage: relayfile writeback retry --opId OP [WORKSPACE]") + case "skip-stuck": + fmt.Fprintln(w, "Usage: relayfile writeback skip-stuck [WORKSPACE] [--workspace WS] [--max N] [--json]") case "sweep-drafts": fmt.Fprintln(w, writebackSweepUsage) default: @@ -655,6 +697,8 @@ func printWritebackUsage(w io.Writer, subcommand string) { relayfile writeback list --state pending|dead [--workspace WS] [--json] relayfile writeback status [WORKSPACE] [--json] relayfile writeback retry --opId OP [WORKSPACE] + relayfile writeback push LOCAL_PATH [--workspace WS] [--json] [--timeout 90s] + relayfile writeback skip-stuck [WORKSPACE] [--workspace WS] [--max N] [--json] relayfile writeback sweep-drafts [WORKSPACE] [--path-prefix PREFIX] [--pattern GLOB ...] [--apply] [--json]`) } } @@ -681,6 +725,7 @@ Usage: relayfile workspace use NAME relayfile workspace list [--names-only] relayfile workspace current [--verbose] + relayfile workspace status [--workspace NAME] [--json] relayfile workspace delete NAME [--yes] relayfile integration connect PROVIDER [--backend BACKEND] [--workspace NAME] (for jira/confluence: prompts for the Atlassian site to bind after OAuth completes) @@ -695,6 +740,8 @@ Usage: relayfile writeback list --state pending|dead [--workspace WS] [--json] relayfile writeback status [WORKSPACE] [--json] relayfile writeback retry --opId OP [WORKSPACE] + relayfile writeback push LOCAL_PATH [--workspace WS] [--json] [--timeout 90s] + relayfile writeback skip-stuck [WORKSPACE] [--workspace WS] [--max N] [--json] relayfile digest rebuild --window today|yesterday|YYYY-MM-DD|this-week|last-week [--workspace NAME] [--json] relayfile pull [--workspace NAME] [--provider PROVIDER] [--reason TEXT] relayfile mount [WORKSPACE] [LOCAL_DIR] @@ -722,6 +769,8 @@ Subcommands: Show local pending, failed, and dead-lettered writebacks writeback retry Re-enqueue a local dead-lettered writeback op + writeback skip-stuck + Walk the events cursor past stuck (404) events without waiting the treat-as-deleted timer digest Regenerate workspace digests digest rebuild Regenerate daily, weekly, or date-stamped digest artifacts @@ -2015,7 +2064,7 @@ func loginWithAPIKey(serverValue, tokenValue string, stdout io.Writer) error { func runWorkspace(args []string, stdin io.Reader, stdout io.Writer) error { if len(args) == 0 { - return errors.New("workspace subcommand is required: create, join, use, list, current, or delete") + return errors.New("workspace subcommand is required: create, join, use, list, current, status, or delete") } switch args[0] { case "create": @@ -2028,6 +2077,8 @@ func runWorkspace(args []string, stdin io.Reader, stdout io.Writer) error { return runWorkspaceList(args[1:], stdout) case "current": return runWorkspaceCurrent(args[1:], stdout) + case "status": + return runWorkspaceStatus(args[1:], stdout) case "delete": return runWorkspaceDelete(args[1:], stdin, stdout) default: @@ -2798,6 +2849,63 @@ type writebackStatusReport struct { LastErrorByProvider map[string]string `json:"lastErrorByProvider"` } +type writebackPushResolvedPath struct { + LocalPath string + MountRoot string + WorkspaceID string + RemoteRoot string + RemotePath string +} + +type writebackPushReceipt struct { + CommandID string `json:"commandId"` + WorkspaceID string `json:"workspaceId"` + RemotePath string `json:"remotePath"` + ContentType string `json:"contentType"` + Content string `json:"content"` + Encoding string `json:"encoding,omitempty"` + Hash string `json:"hash"` + Exists bool `json:"exists"` + Status string `json:"status"` + FirstSeenAt string `json:"firstSeenAt"` + LastAttemptAt string `json:"lastAttemptAt,omitempty"` + NextAttemptAt string `json:"nextAttemptAt,omitempty"` + AttemptCount int `json:"attemptCount"` + LastError string `json:"lastError,omitempty"` + NeedsAttention bool `json:"needsAttention,omitempty"` + OpID string `json:"opId,omitempty"` + DispatchStatus string `json:"dispatchStatus,omitempty"` + AckedAt string `json:"ackedAt,omitempty"` + Revision string `json:"revision,omitempty"` + CorrelationID string `json:"correlationId,omitempty"` + ContentIdentity *contentIdentity `json:"contentIdentity,omitempty"` +} + +// withoutBody returns a copy with the file content stripped. Persisted +// terminal receipts (acked/failed) and any printed receipt must not retain or +// log arbitrary user file contents beyond the in-flight writeback; only the +// transient pending receipt keeps the body so a retry can resend it. +func (r writebackPushReceipt) withoutBody() writebackPushReceipt { + r.Content = "" + r.Encoding = "" + return r +} + +type workspaceHealthReport struct { + WorkspaceID string `json:"workspaceId"` + Name string `json:"name,omitempty"` + LocalDir string `json:"localDir,omitempty"` + Status string `json:"status,omitempty"` + LastSuccessfulReconcileAt string `json:"lastSuccessfulReconcileAt,omitempty"` + LastReconcileAt string `json:"lastReconcileAt,omitempty"` + LastError string `json:"lastError,omitempty"` + StuckEventCount int `json:"stuckEventCount"` + OutboxPending int `json:"outboxPending"` + OutboxFailed int `json:"outboxFailed"` + OutboxAcked int `json:"outboxAcked"` + IncrementalBacklogDraining bool `json:"incrementalBacklogDraining,omitempty"` +} + var errWritebackFailuresPresent = errors.New("writeback failures present") func deadLetterDirFor(localDir string) string { @@ -2808,17 +2916,526 @@ func deadLetterErrorPathFor(localDir, opID string) string { return filepath.Join(deadLetterDirFor(localDir), opID+".error.json") } +func runWritebackPush(args []string, stdout io.Writer) error { + fs := flag.NewFlagSet("writeback push", flag.ContinueOnError) + fs.SetOutput(io.Discard) + workspaceName := fs.String("workspace", "", "workspace name or id") + server := fs.String("server", "", "relayfile server URL override") + tokenOverride := fs.String("token", "", "relayfile token override") + jsonOutput := fs.Bool("json", false, "emit JSON") + timeout := fs.Duration("timeout", 90*time.Second, "operation receipt wait timeout") + if err := fs.Parse(normalizeFlagArgs(args, map[string]bool{ + "workspace": true, + "server": true, + "token": true, + "json": false, + "timeout": true, + })); err != nil { + return err + } + if fs.NArg() != 1 { + return errors.New("usage: relayfile writeback push LOCAL_PATH [--workspace WS] [--json] [--timeout 90s]") + } + + resolved, err := resolveWritebackPushPath(fs.Arg(0), strings.TrimSpace(*workspaceName)) + if err != nil { + return err + } + raw, err := os.ReadFile(resolved.LocalPath) + if err != nil { + return err + } + content, encoding := encodeLocalWritebackContent(raw) + contentType := detectContentType(resolved.LocalPath, raw) + contentHash := hashBytes(raw) + identity := writebackPushContentIdentity(resolved.WorkspaceID, resolved.RemotePath, contentHash) + if identity == nil { + return fmt.Errorf("writeback push currently supports only draft writeback paths and factory-create-*.json files; %s has no content identity for double-dispatch dedupe", resolved.RemotePath) + } + joinScopes := writebackPushJoinScopes(resolved.RemotePath) + requiredRelayfileScopes := writebackPushRequiredRelayfileScopes(resolved.RemotePath) + if strings.TrimSpace(*tokenOverride) == "" { + if err := ensureWritebackDelegatedCredentials(resolved.WorkspaceID, joinScopes, requiredRelayfileScopes); err != nil { + return err + } + } + commandClient, err := prepareWorkspaceCommandClient(resolved.WorkspaceID, *server, *tokenOverride, joinScopes) + if err != nil { + return err + } + + now := time.Now().UTC() + firstSeenAt := now.Format(time.RFC3339Nano) + commandID := newWritebackPushCommandID(resolved.WorkspaceID, resolved.RemotePath, contentHash, firstSeenAt) + pendingReceipt := writebackPushReceipt{ + CommandID: commandID, + WorkspaceID: resolved.WorkspaceID, + RemotePath: resolved.RemotePath, + ContentType: contentType, + Content: content, + Encoding: encoding, + Hash: contentHash, + Exists: true, + Status: "pending", + FirstSeenAt: firstSeenAt, + CorrelationID: commandID, + ContentIdentity: identity, + } + if err := writeWritebackPushReceipt(resolved.MountRoot, pendingReceipt); err != nil { + return err + } + + file := bulkWriteFile{ + Path: resolved.RemotePath, + ContentType: contentType, + Content: content, + Encoding: encoding, + ContentIdentity: identity, + } + var response bulkWriteResponse + err = commandClient.postWorkspaceJSON(context.Background(), func(workspaceID string) string { + return fmt.Sprintf("/v1/workspaces/%s/fs/bulk", url.PathEscape(workspaceID)) + }, bulkWriteRequest{Files: []bulkWriteFile{file}}, &response) + if err != nil { + failed := pendingReceipt + failed.Status = "failed" + failed.LastAttemptAt = time.Now().UTC().Format(time.RFC3339Nano) + failed.AttemptCount = 1 + failed.LastError = sanitizeCLIReceiptError(err) + failed.DispatchStatus = "failed" + if receiptErr := writeWritebackPushReceipt(resolved.MountRoot, failed); receiptErr != nil { + return fmt.Errorf("push failed: %w; additionally failed to write failure receipt: %v", err, receiptErr) + } + return err + } + result := firstBulkWriteResultForPath(response.Results, resolved.RemotePath) + if response.ErrorCount > 0 { + reason := firstBulkWriteErrorForPath(response.Errors, resolved.RemotePath) + if reason == "" { + reason = fmt.Sprintf("bulk write returned %d error(s)", response.ErrorCount) + } + failed := pendingReceipt + failed.Status = "failed" + failed.LastAttemptAt = time.Now().UTC().Format(time.RFC3339Nano) + failed.AttemptCount = 1 + failed.LastError = sanitizeCLIReceiptError(errors.New(reason)) + failed.DispatchStatus = "failed" + failed.CorrelationID = firstNonBlank(response.CorrelationID, failed.CorrelationID) + if receiptErr := writeWritebackPushReceipt(resolved.MountRoot, failed); receiptErr != nil { + return fmt.Errorf("%s; additionally failed to write failure receipt: %v", reason, receiptErr) + } + return errors.New(reason) + } + revision := firstNonBlank(result.Revision, "") + opID := strings.TrimSpace(result.OpID) + dispatchStatus := "succeeded" + if result.Writeback != nil && strings.TrimSpace(result.Writeback.State) != "" { + dispatchStatus = strings.TrimSpace(result.Writeback.State) + } + if opID != "" { + waitTimeout := *timeout + if waitTimeout <= 0 { + waitTimeout = 90 * time.Second + } + waitCtx, cancel := context.WithTimeout(context.Background(), waitTimeout) + op, err := waitForWritebackOperation(waitCtx, commandClient, opID) + cancel() + if err != nil { + // The write already landed on the cloud (we have an opID). A + // terminal op status (failed/dead_lettered/canceled) is a real + // failure; a poll timeout or transient GET error is "unknown" — + // the op may still succeed cloud-side, so keep the receipt pending + // (flagged needsAttention) instead of diverging local state with a + // premature failed receipt. + receipt := pendingReceipt + receipt.OpID = opID + receipt.LastAttemptAt = time.Now().UTC().Format(time.RFC3339Nano) + receipt.AttemptCount = 1 + receipt.LastError = sanitizeCLIReceiptError(err) + receipt.CorrelationID = firstNonBlank(response.CorrelationID, receipt.CorrelationID) + if isTerminalWritebackOpStatus(op.Status) { + receipt.Status = "failed" + receipt.DispatchStatus = firstNonBlank(strings.TrimSpace(op.Status), "failed") + } else { + receipt.Status = "pending" + receipt.DispatchStatus = firstNonBlank(strings.TrimSpace(op.Status), "dispatched") + receipt.NeedsAttention = true + } + if receiptErr := writeWritebackPushReceipt(resolved.MountRoot, receipt); receiptErr != nil { + return fmt.Errorf("%w; additionally failed to write receipt: %v", err, receiptErr) + } + return err + } + revision = firstNonBlank(revision, op.Revision) + dispatchStatus = firstNonBlank(op.Status, dispatchStatus) + } + acked := pendingReceipt + acked.Status = "acked" + acked.LastAttemptAt = time.Now().UTC().Format(time.RFC3339Nano) + acked.AttemptCount = 1 + acked.OpID = opID + acked.DispatchStatus = dispatchStatus + acked.AckedAt = time.Now().UTC().Format(time.RFC3339Nano) + acked.Revision = revision + acked.CorrelationID = firstNonBlank(response.CorrelationID, acked.CorrelationID) + if err := writeWritebackPushReceipt(resolved.MountRoot, acked); err != nil { + return err + } + if *jsonOutput { + return writeJSON(stdout, acked.withoutBody()) + } + fmt.Fprintf(stdout, "Pushed %s -> %s", resolved.LocalPath, resolved.RemotePath) + if opID != "" { + fmt.Fprintf(stdout, " (%s)", opID) + } + fmt.Fprintln(stdout) + return nil +} + +type writebackOperationStatus struct { + OpID string `json:"opId,omitempty"` + Path string `json:"path,omitempty"` + Status string `json:"status,omitempty"` + Revision string `json:"revision,omitempty"` +} + +// isTerminalWritebackOpStatus reports whether a cloud op status is a confirmed +// terminal failure. Anything else (empty, in-flight, or unknown) is treated as +// not-yet-final so the push keeps the receipt pending rather than failing it. +func isTerminalWritebackOpStatus(status string) bool { + switch strings.TrimSpace(status) { + case "failed", "dead_lettered", "canceled": + return true + default: + return false + } +} + +func waitForWritebackOperation(ctx context.Context, commandClient *workspaceCommandClient, opID string) (writebackOperationStatus, error) { + if strings.TrimSpace(opID) == "" { + return writebackOperationStatus{Status: "succeeded"}, nil + } + for { + var op writebackOperationStatus + err := commandClient.getWorkspaceJSON(ctx, func(workspaceID string) string { + return fmt.Sprintf("/v1/workspaces/%s/ops/%s", url.PathEscape(workspaceID), url.PathEscape(opID)) + }, &op) + if err == nil { + switch strings.TrimSpace(op.Status) { + case "succeeded": + return op, nil + case "failed", "dead_lettered", "canceled": + return op, fmt.Errorf("writeback operation %s %s", opID, strings.TrimSpace(op.Status)) + } + } + // A transient poll error (network glitch, 5xx) should not abort the + // whole push — keep retrying until the context deadline fires. The + // select also makes the poll cancellable (e.g. Ctrl+C) instead of + // sleeping blindly. + select { + case <-ctx.Done(): + if err != nil { + return op, fmt.Errorf("timed out waiting for writeback operation %s: %w", opID, err) + } + return op, fmt.Errorf("timed out waiting for writeback operation %s: %w", opID, ctx.Err()) + case <-time.After(500 * time.Millisecond): + } + } +} + +func ensureWritebackDelegatedCredentials(workspaceID string, joinScopes, requiredRelayfileScopes []string) error { + if bundle, _, err := loadDelegatedCredentials(""); err == nil && + workspaceRequestMatchesDelegatedCredentials(workspaceID, bundle.Workspace()) && + delegatedBundleHasScopes(bundle, requiredRelayfileScopes) { + return nil + } + if _, err := bootstrapDelegatedCredentialsFromAgentRelay(workspaceID, joinScopes); err != nil { + return err + } + // Re-validate after bootstrap: if Cloud returns a delegated bundle that + // omits the compiled relayfile:fs:write://** scope, fail loudly + // here rather than proceeding to write receipts and call /fs/bulk with an + // under-scoped token. + bundle, _, err := loadDelegatedCredentials("") + if err != nil { + return err + } + if !workspaceRequestMatchesDelegatedCredentials(workspaceID, bundle.Workspace()) || + !delegatedBundleHasScopes(bundle, requiredRelayfileScopes) { + return fmt.Errorf("delegated relayfile credentials for workspace %s do not include required scope(s): %s", workspaceID, strings.Join(requiredRelayfileScopes, ", ")) + } + return nil +} + +func delegatedBundleHasScopes(bundle delegatedauth.Bundle, required []string) bool { + available := append(append([]string(nil), bundle.Scopes...), bundle.RelayfileScopes...) + for _, want := range required { + want = strings.TrimSpace(want) + if want == "" { + continue + } + found := false + for _, got := range available { + if strings.TrimSpace(got) == want { + found = true + break + } + } + if !found { + return false + } + } + return true +} + +func resolveWritebackPushPath(localPath, workspaceValue string) (writebackPushResolvedPath, error) { + abs, err := filepath.Abs(localPath) + if err != nil { + return writebackPushResolvedPath{}, err + } + if evaluated, err := filepath.EvalSymlinks(abs); err == nil { + abs = evaluated + } + info, err := os.Stat(abs) + if err != nil { + return writebackPushResolvedPath{}, err + } + if info.IsDir() { + return writebackPushResolvedPath{}, fmt.Errorf("%s is a directory", abs) + } + // Reject FIFOs, device nodes, sockets, etc. os.ReadFile on a special file + // can hang or read an unintended stream; direct push only accepts regular + // files. + if !info.Mode().IsRegular() { + return writebackPushResolvedPath{}, fmt.Errorf("%s is not a regular file", abs) + } + mountRoot, err := findRelayMountRoot(filepath.Dir(abs)) + if err != nil { + return writebackPushResolvedPath{}, err + } + state, err := readWritebackState(mountRoot) + if err != nil { + return writebackPushResolvedPath{}, err + } + workspaceID := strings.TrimSpace(state.WorkspaceID) + if workspaceID == "" { + return writebackPushResolvedPath{}, fmt.Errorf("%s missing workspaceId", filepath.Join(mountRoot, ".relay", "state.json")) + } + if strings.TrimSpace(workspaceValue) != "" && !workspaceRequestMatchesDelegatedCredentials(workspaceValue, workspaceID) { + return writebackPushResolvedPath{}, fmt.Errorf("local path belongs to workspace %s, not %s", workspaceID, workspaceValue) + } + root := mountRoot + rel, err := filepath.Rel(root, abs) + if err != nil { + return writebackPushResolvedPath{}, err + } + if rel == "." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) || rel == ".." { + return writebackPushResolvedPath{}, fmt.Errorf("%s is outside mount root %s", abs, root) + } + remotePath := joinRemotePath(readMountRemoteRoot(mountRoot), filepath.ToSlash(rel)) + return writebackPushResolvedPath{ + LocalPath: abs, + MountRoot: mountRoot, + WorkspaceID: workspaceID, + RemoteRoot: readMountRemoteRoot(mountRoot), + RemotePath: remotePath, + }, nil +} + +func findRelayMountRoot(start string) (string, error) { + dir := filepath.Clean(start) + for { + if _, err := os.Stat(filepath.Join(dir, ".relay", "state.json")); err == nil { + return dir, nil + } + parent := filepath.Dir(dir) + if parent == dir { + break + } + dir = parent + } + return "", fmt.Errorf("could not find .relay/state.json above %s", start) +} + +func joinRemotePath(root, rel string) string { + // Trim a trailing slash from the mount's remoteRoot (e.g. "/linear/") so + // the join does not produce "/linear//file.json"; the server may treat the + // double-slash path literally and break path matching/dedupe. + root = strings.TrimRight(normalizeWritebackFailurePath(root), "/") + if root == "" { + root = "/" + } + rel = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(rel)), "/") + if rel == "." || rel == "" { + return root + } + if root == "/" { + return "/" + rel + } + return root + "/" + rel +} + +func encodeLocalWritebackContent(raw []byte) (string, string) { + if utf8.Valid(raw) { + return string(raw), "" + } + return base64.StdEncoding.EncodeToString(raw), "base64" +} + +func hashBytes(raw []byte) string { + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + +func writebackPushJoinScopes(remotePath string) []string { + // ops:read is requested alongside fs:write because a successful /fs/bulk + // returns an opID that this command then polls at + // GET /v1/workspaces/{id}/ops/{opId} (server.go requires ops:read). Without + // it the write succeeds but the status poll is unauthorized; the push then + // leaves the op pending+needsAttention rather than failing it. + parts := strings.Split(strings.Trim(normalizeWritebackFailurePath(remotePath), "/"), "/") + if len(parts) == 0 || strings.TrimSpace(parts[0]) == "" { + return []string{"fs:write:/**", "ops:read"} + } + provider := strings.TrimSpace(parts[0]) + return []string{fmt.Sprintf("fs:write:/%s/**", provider), "ops:read"} +} + +func writebackPushRequiredRelayfileScopes(remotePath string) []string { + parts := strings.Split(strings.Trim(normalizeWritebackFailurePath(remotePath), "/"), "/") + if len(parts) == 0 || strings.TrimSpace(parts[0]) == "" { + return []string{"relayfile:fs:write:/**"} + } + provider := strings.TrimSpace(parts[0]) + return []string{fmt.Sprintf("relayfile:fs:write:/%s/**", provider)} +} + +func writebackPushContentIdentity(workspaceID, remotePath, contentHash string) *contentIdentity { + if !isWritebackDraftPath(remotePath) { + return nil + } + return &contentIdentity{ + Kind: "mount-writeback-create-draft", + Key: fmt.Sprintf("%s:%s:%s", strings.TrimSpace(workspaceID), normalizeWritebackFailurePath(remotePath), strings.TrimSpace(contentHash)), + TTLSeconds: 2592000, + } +} + +func isWritebackDraftPath(remotePath string) bool { + // Remote paths are slash-separated regardless of host OS, so use + // path.Base (not filepath.Base, which splits on "\\" on Windows). + base := path.Base(normalizeWritebackFailurePath(remotePath)) + return strings.HasPrefix(base, "factory-create-") && strings.HasSuffix(base, ".json") || + relayfile.IsDraftFilePath(remotePath) +} + +func newWritebackPushCommandID(workspaceID, remotePath, hash, firstSeenAt string) string { + sum := sha256.Sum256([]byte(strings.Join([]string{ + strings.TrimSpace(workspaceID), + normalizeWritebackFailurePath(remotePath), + strings.TrimSpace(hash), + strings.TrimSpace(firstSeenAt), + }, "\x00"))) + return "mountcmd_" + hex.EncodeToString(sum[:])[:32] +} + +func writeWritebackPushReceipt(mountRoot string, receipt writebackPushReceipt) error { + status := strings.TrimSpace(receipt.Status) + if status == "" { + status = "pending" + } + receipt.Status = status + if receipt.CorrelationID == "" { + receipt.CorrelationID = receipt.CommandID + } + outboxRoot := filepath.Join(mountRoot, ".relay", "outbox") + for _, dir := range []string{"pending", "acked", "failed"} { + if err := os.MkdirAll(filepath.Join(outboxRoot, dir), 0o755); err != nil { + return err + } + } + targetDir := "pending" + // Only the transient pending receipt retains the file body (so a retry can + // resend it), and it is written 0600 to limit exposure. Terminal receipts + // drop the body entirely. + perm := os.FileMode(0o600) + switch status { + case "acked": + targetDir = "acked" + receipt = receipt.withoutBody() + perm = 0o644 + case "failed": + targetDir = "failed" + receipt = receipt.withoutBody() + perm = 0o644 + } + data, err := json.Marshal(receipt) + if err != nil { + return err + } + if err := writeFileAtomically(filepath.Join(outboxRoot, targetDir, receipt.CommandID+".json"), data, perm); err != nil { + return err + } + if status != "pending" { + if err := os.Remove(filepath.Join(outboxRoot, "pending", receipt.CommandID+".json")); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + } + return nil +} + +func firstBulkWriteResultForPath(results []bulkWriteResult, remotePath string) bulkWriteResult { + remotePath = normalizeWritebackFailurePath(remotePath) + for _, result := range results { + if normalizeWritebackFailurePath(result.Path) == remotePath { + return result + } + } + if len(results) > 0 { + return results[0] + } + return bulkWriteResult{} +} + +func firstBulkWriteErrorForPath(errors []bulkWriteError, remotePath string) string { + remotePath = normalizeWritebackFailurePath(remotePath) + for _, item := range errors { + if normalizeWritebackFailurePath(item.Path) == remotePath { + return firstNonBlank(item.Message, item.Code) + } + } + if len(errors) > 0 { + return firstNonBlank(errors[0].Message, errors[0].Code) + } + return "" +} + +func sanitizeCLIReceiptError(err error) string { + if err == nil { + return "" + } + message := strings.TrimSpace(err.Error()) + if len(message) > 2000 { + return message[:2000] + } + return message +} + func runWriteback(args []string, stdout io.Writer) error { if len(args) == 0 { - return errors.New("writeback subcommand is required: list, status, retry, or sweep-drafts") + return errors.New("writeback subcommand is required: list, push, status, retry, or sweep-drafts") } switch args[0] { case "list": return runWritebackList(args[1:], stdout) + case "push": + return runWritebackPush(args[1:], stdout) case "status": return runWritebackStatus(args[1:], stdout) case "retry": return runWritebackRetry(args[1:], stdout) + case "skip-stuck": + return runWritebackSkipStuck(args[1:], stdout) case "sweep-drafts": return runWritebackSweepDrafts(args[1:], stdout) default: @@ -2826,6 +3443,102 @@ func runWriteback(args []string, stdout io.Writer) error { } } +// runWritebackSkipStuck is the operator escape hatch for a wedged events +// cursor. It walks the events cursor forward, dropping consecutive read-404 +// ("not readable yet") events immediately without waiting the 5-minute +// treat-as-deleted timer, and reports how many stuck events it skipped. +func runWritebackSkipStuck(args []string, stdout io.Writer) error { + fs := flag.NewFlagSet("writeback skip-stuck", flag.ContinueOnError) + fs.SetOutput(io.Discard) + workspaceName := fs.String("workspace", "", "workspace name or id") + maxSkips := fs.Int("max", 0, "maximum number of stuck events to skip (0 = unbounded)") + jsonOutput := fs.Bool("json", false, "emit JSON") + if err := fs.Parse(normalizeFlagArgs(args, map[string]bool{ + "workspace": true, + "max": true, + "json": false, + })); err != nil { + return err + } + if fs.NArg() > 1 { + return errors.New("usage: relayfile writeback skip-stuck [WORKSPACE] [--workspace WS] [--max N] [--json]") + } + if *maxSkips < 0 { + return errors.New("--max must be >= 0") + } + + workspaceValue := firstNonBlank(strings.TrimSpace(*workspaceName), firstArg(fs)) + workspaceID, record, err := resolveWorkspaceLikeStatus(workspaceValue) + if err != nil { + return err + } + if strings.TrimSpace(record.LocalDir) == "" { + return errors.New("workspace has no local mirror") + } + + creds, err := loadCredentials() + if err != nil { + return err + } + tokenValue := resolveToken("", creds) + if tokenValue == "" { + return errors.New("token is required; set RELAYFILE_TOKEN or pass explicit relayfile credentials") + } + server := strings.TrimSpace(record.Server) + if server == "" { + server = resolveServer("", creds) + } + timeout := durationEnv("RELAYFILE_MOUNT_TIMEOUT", defaultMountTimeout) + if timeout <= 0 { + timeout = defaultMountTimeout + } + client := mountsync.NewHTTPClient(server, tokenValue, &http.Client{ + Transport: newWritebackFailureTransport(record.LocalDir, log.Default(), mountsync.NewSyncTransport()), + }) + remoteRoot := readMountRemoteRoot(record.LocalDir) + websocketDisabled := true + syncer, err := mountsync.NewSyncer(client, mountsync.SyncerOptions{ + WorkspaceID: workspaceID, + RemoteRoot: remoteRoot, + LocalRoot: record.LocalDir, + WebSocket: &websocketDisabled, + RootCtx: context.Background(), + Logger: log.Default(), + }) + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + skipped, syncErr := syncer.SkipStuck(ctx, *maxSkips) + backlog := syncer.BacklogDraining() + + if *jsonOutput { + result := struct { + Workspace string `json:"workspace"` + Skipped int `json:"skipped"` + Backlog bool `json:"backlogRemaining"` + Error string `json:"error,omitempty"` + }{Workspace: workspaceID, Skipped: skipped, Backlog: backlog} + if syncErr != nil { + result.Error = syncErr.Error() + } + if err := writeJSON(stdout, result); err != nil { + return err + } + return syncErr + } + + fmt.Fprintf(stdout, "Skipped %d stuck event(s)\n", skipped) + if backlog { + fmt.Fprintln(stdout, "Backlog remains — re-run 'relayfile writeback skip-stuck' to continue clearing") + } else { + fmt.Fprintln(stdout, "Events cursor caught up to live head") + } + return syncErr +} + func runWritebackStatus(args []string, stdout io.Writer) error { fs := flag.NewFlagSet("writeback status", flag.ContinueOnError) fs.SetOutput(io.Discard) @@ -3046,16 +3759,29 @@ func resolveWorkspaceLikeStatus(value string) (string, workspaceRecord, error) { // supplies a name/id; only fall back to the credentials path when // nothing local matches (or when no value was given and we need // the JWT's `wks` claim to identify the default). + // + // When no value is given, resolve the modern selection chain (env + // RELAYFILE_WORKSPACE, active Agent Relay workspace, catalog default) + // against the local registry before the credentials path: in the + // delegated Agent Relay flow credentials.json is removed, so falling + // straight through to loadCredentials would fail even when a default + // local mirror exists. trimmed := strings.TrimSpace(value) - if trimmed != "" { - if local, ok := workspaceRecordByName(trimmed); ok { + candidate := trimmed + if candidate == "" { + if name, _ := activeWorkspaceName(""); strings.TrimSpace(name) != "" { + candidate = strings.TrimSpace(name) + } + } + if candidate != "" { + if local, ok := workspaceRecordByName(candidate); ok { workspaceID := strings.TrimSpace(local.ID) if workspaceID == "" { workspaceID = local.Name } return workspaceID, local, nil } - if local, ok := workspaceRecordByID(trimmed); ok { + if local, ok := workspaceRecordByID(candidate); ok { return strings.TrimSpace(local.ID), local, nil } } @@ -3844,6 +4570,135 @@ func runWorkspaceCurrent(args []string, stdout io.Writer) error { return nil } +func runWorkspaceStatus(args []string, stdout io.Writer) error { + fs := flag.NewFlagSet("workspace status", flag.ContinueOnError) + fs.SetOutput(io.Discard) + workspaceName := fs.String("workspace", "", "workspace name or id") + jsonOutput := fs.Bool("json", false, "emit JSON") + if err := fs.Parse(normalizeFlagArgs(args, map[string]bool{ + "workspace": true, + "json": false, + })); err != nil { + return err + } + if fs.NArg() > 1 { + return errors.New("usage: relayfile workspace status [--workspace NAME] [--json]") + } + value := strings.TrimSpace(*workspaceName) + if value == "" && fs.NArg() == 1 { + value = strings.TrimSpace(fs.Arg(0)) + } + workspaceID, record, err := resolveWorkspaceLikeStatus(value) + if err != nil { + return err + } + report := buildWorkspaceHealthReport(workspaceID, record) + if *jsonOutput { + return writeJSON(stdout, report) + } + printWorkspaceHealthReport(stdout, report) + return nil +} + +func buildWorkspaceHealthReport(workspaceID string, record workspaceRecord) workspaceHealthReport { + report := workspaceHealthReport{ + WorkspaceID: strings.TrimSpace(workspaceID), + Name: strings.TrimSpace(record.Name), + LocalDir: strings.TrimSpace(record.LocalDir), + } + if report.LocalDir == "" { + return report + } + state := readWritebackStateBestEffort(report.LocalDir) + report.Status = strings.TrimSpace(state.Status) + report.LastSuccessfulReconcileAt = strings.TrimSpace(state.LastSuccessfulReconcileAt) + report.LastReconcileAt = strings.TrimSpace(state.LastReconcileAt) + if state.LastError != nil { + report.LastError = strings.TrimSpace(firstNonBlank(state.LastError.Message, state.LastError.Code)) + } + // Use the public sync state's not-ready set as the stuck-event baseline so + // the count is non-zero even when the private cursor files are absent or + // the first readable one lacks the field; then take the max with the + // private cursor health (which also carries the backlog-draining flag). + report.StuckEventCount = len(state.IncrementalReadNotReadySince) + cursorStuckCount, backlogDraining := readLocalMountCursorHealth(report.LocalDir) + if cursorStuckCount > report.StuckEventCount { + report.StuckEventCount = cursorStuckCount + } + report.IncrementalBacklogDraining = backlogDraining + report.OutboxPending = countJSONFiles(filepath.Join(report.LocalDir, ".relay", "outbox", "pending")) + report.OutboxFailed = countJSONFiles(filepath.Join(report.LocalDir, ".relay", "outbox", "failed")) + report.OutboxAcked = countJSONFiles(filepath.Join(report.LocalDir, ".relay", "outbox", "acked")) + return report +} + +func readWritebackStateBestEffort(localDir string) syncStateFile { + state, err := readWritebackState(localDir) + if err == nil { + return state + } + return syncStateFile{} +} + +func readLocalMountCursorHealth(localDir string) (stuckCount int, backlogDraining bool) { + for _, path := range []string{ + filepath.Join(localDir, ".relayfile-mount-state.json"), + filepath.Join(localDir, mountsync.DefaultMountStateDirName, "state.json"), + } { + payload, err := os.ReadFile(path) + if err != nil { + continue + } + var state struct { + IncrementalReadNotReadySince map[string]string `json:"incrementalReadNotReadySince"` + IncrementalBacklogDraining bool `json:"incrementalBacklogDraining"` + } + if json.Unmarshal(payload, &state) != nil { + continue + } + return len(state.IncrementalReadNotReadySince), state.IncrementalBacklogDraining + } + return 0, false +} + +func countJSONFiles(dir string) int { + entries, err := os.ReadDir(dir) + if err != nil { + return 0 + } + count := 0 + for _, entry := range entries { + if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".json") { + count++ + } + } + return count +} + +func printWorkspaceHealthReport(stdout io.Writer, report workspaceHealthReport) { + label := report.WorkspaceID + if report.Name != "" && report.Name != report.WorkspaceID { + label = fmt.Sprintf("%s (%s)", report.WorkspaceID, report.Name) + } + fmt.Fprintf(stdout, "workspace: %s\n", label) + if report.LocalDir == "" { + fmt.Fprintln(stdout, "local mirror: not configured") + return + } + fmt.Fprintf(stdout, "local mirror: %s\n", report.LocalDir) + fmt.Fprintf(stdout, "status: %s\n", defaultIfBlank(report.Status, "-")) + fmt.Fprintf(stdout, "last successful reconcile: %s\n", defaultIfBlank(report.LastSuccessfulReconcileAt, "-")) + fmt.Fprintf(stdout, "last reconcile: %s\n", defaultIfBlank(report.LastReconcileAt, "-")) + fmt.Fprintf(stdout, "stuck events: %d\n", report.StuckEventCount) + fmt.Fprintf(stdout, "outbox: pending=%d failed=%d acked=%d\n", report.OutboxPending, report.OutboxFailed, report.OutboxAcked) + if report.LastError != "" { + fmt.Fprintf(stdout, "last error: %s\n", report.LastError) + } + if report.IncrementalBacklogDraining { + fmt.Fprintln(stdout, "incremental backlog: draining") + } +} + func runWorkspaceDelete(args []string, stdin io.Reader, stdout io.Writer) error { fs := flag.NewFlagSet("workspace delete", flag.ContinueOnError) fs.SetOutput(io.Discard) @@ -7936,6 +8791,24 @@ func spawnBackgroundMountProcess(originalArgs []string, localDir, pidFile, logFi return nil } +// logStuckEventSummary surfaces the stuck-event drain outcome of a reconcile +// cycle. In the request-driven, one-shot pull model a cycle can be canceled +// mid-drain; without this the CLI exits silently and the operator cannot tell +// whether the cursor caught up. When events were skipped or a backlog remains, +// it points the operator at `relayfile writeback skip-stuck`. +func logStuckEventSummary(syncer *mountsync.Syncer, cycleErr error) { + skipped := syncer.StaleAliasSkips() + backlog := syncer.BacklogDraining() + if skipped == 0 && !backlog { + return + } + if backlog || errors.Is(cycleErr, context.Canceled) || errors.Is(cycleErr, context.DeadlineExceeded) { + log.Printf("stuck-event drain: %d stale index event(s) skipped this cycle; backlog remains — re-run or use 'relayfile writeback skip-stuck' to clear", skipped) + return + } + log.Printf("stuck-event drain: %d stale index event(s) skipped this cycle; events cursor caught up to live head", skipped) +} + func runMountLoop(rootCtx context.Context, syncer *mountsync.Syncer, localDir, workspaceID, serverURL, delegatedCredsFile string, timeout, interval time.Duration, intervalJitter float64, websocketEnabled, once, daemonized bool, pidFile, logFile string) error { interval = enforcePollIntervalFloor(interval) httpClient, _ := syncerClient(syncer) @@ -8135,6 +9008,7 @@ func runMountLoop(rootCtx context.Context, syncer *mountsync.Syncer, localDir, w log.Print(mountStartBanner(localDir, interval, intervalJitter)) initialErr := runCycle(true) + logStuckEventSummary(syncer, initialErr) if once { return initialErr } diff --git a/cmd/relayfile-cli/main_test.go b/cmd/relayfile-cli/main_test.go index 52fdbddd..4b693344 100644 --- a/cmd/relayfile-cli/main_test.go +++ b/cmd/relayfile-cli/main_test.go @@ -241,7 +241,7 @@ func TestWorkspaceRequiresSubcommandMentionsJoin(t *testing.T) { if err == nil { t.Fatal("expected workspace without subcommand to fail") } - if got := err.Error(); !strings.Contains(got, "create, join, use, list, current, or delete") { + if got := err.Error(); !strings.Contains(got, "create, join, use, list, current, status, or delete") { t.Fatalf("workspace subcommand error = %q", got) } } @@ -2353,6 +2353,279 @@ func writeDelegatedCredentialsForTest(t *testing.T, bundle delegatedauth.Bundle) return path } +func TestWritebackPushPostsBulkAndWritesAckedReceipt(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + clearRelayfileEnv(t) + + const workspaceID = "ws_demo" + const remotePath = "/linear/issues/factory-create-ar-272-test.json" + const opID = "op_ar_272" + localDir := t.TempDir() + if err := ensureMirrorLayout(localDir); err != nil { + t.Fatalf("ensureMirrorLayout failed: %v", err) + } + if err := writeMirrorStateFile(localDir, syncStateFile{ + WorkspaceID: workspaceID, + RemoteRoot: "/linear/issues", + Mode: defaultMountMode, + }); err != nil { + t.Fatalf("writeMirrorStateFile failed: %v", err) + } + if _, err := upsertWorkspaceDetails(workspaceRecord{ + Name: "demo", + ID: workspaceID, + LocalDir: localDir, + CreatedAt: time.Now().UTC().Format(time.RFC3339), + LastUsedAt: time.Now().UTC().Format(time.RFC3339), + }); err != nil { + t.Fatalf("upsertWorkspaceDetails failed: %v", err) + } + + localPath := filepath.Join(localDir, "factory-create-ar-272-test.json") + content := []byte(`{"title":"AR-272 synthetic test"}` + "\n") + if err := os.WriteFile(localPath, content, 0o644); err != nil { + t.Fatalf("write local payload failed: %v", err) + } + + if err := saveCloudCredentials(cloudCredentials{ + APIURL: "https://legacy-cloud-credentials.invalid", + AccessToken: "legacy_token_must_not_be_used", + }); err != nil { + t.Fatalf("save legacy cloud credentials failed: %v", err) + } + + var sawDelegated atomic.Bool + var sawBulk atomic.Bool + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/workspaces/"+workspaceID+"/relayfile/delegated-token": + sawDelegated.Store(true) + if got := r.Header.Get("Authorization"); got != "Bearer cld_access" { + t.Fatalf("unexpected delegated-token Authorization: %q", got) + } + var body cloudRelayfileDelegatedTokenRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatalf("decode delegated-token body failed: %v", err) + } + if body.AgentName != "relayfile-cli" { + t.Fatalf("agentName = %q, want relayfile-cli", body.AgentName) + } + if got, want := strings.Join(body.Scopes, ","), "fs:write:/linear/**,ops:read"; got != want { + t.Fatalf("delegated-token scopes = %q, want %q", got, want) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "relayfileUrl": server.URL, + "relayauthUrl": server.URL, + "relayfileWorkspaceId": workspaceID, + "relayfileToken": "rf_write", + "relayfileTokenExpiresAt": time.Now().Add(time.Hour).UTC().Format(time.RFC3339), + "relayfileRefreshToken": "refresh_write", + "relayfileRefreshTokenExpiresAt": time.Now().Add(24 * time.Hour).UTC().Format(time.RFC3339), + "relayfileScopes": []string{"relayfile:fs:write:/linear/**"}, + "delegationNotAfter": time.Now().Add(24 * time.Hour).UTC().Format(time.RFC3339), + "relayfileMountPaths": []string{"/linear/**"}, + }) + case r.Method == http.MethodPost && r.URL.Path == "/v1/workspaces/"+workspaceID+"/fs/bulk": + sawBulk.Store(true) + if got := r.Header.Get("Authorization"); got != "Bearer rf_write" { + t.Fatalf("unexpected bulk Authorization: %q", got) + } + var req bulkWriteRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode bulk request failed: %v", err) + } + if len(req.Files) != 1 { + t.Fatalf("bulk files = %d, want 1", len(req.Files)) + } + file := req.Files[0] + if file.Path != remotePath { + t.Fatalf("bulk path = %q, want %q", file.Path, remotePath) + } + if file.Content != string(content) { + t.Fatalf("bulk content = %q, want %q", file.Content, string(content)) + } + if file.ContentIdentity == nil { + t.Fatal("expected content identity for factory-create writeback") + } + if file.ContentIdentity.Kind != "mount-writeback-create-draft" { + t.Fatalf("content identity kind = %q", file.ContentIdentity.Kind) + } + if !strings.HasPrefix(file.ContentIdentity.Key, workspaceID+":"+remotePath+":") { + t.Fatalf("content identity key = %q", file.ContentIdentity.Key) + } + _, _ = io.WriteString(w, `{"written":1,"errorCount":0,"correlationId":"corr_ar_272","results":[{"path":"`+remotePath+`","revision":"rev_1","opId":"`+opID+`","writeback":{"provider":"linear","state":"succeeded"}}]}`) + case r.Method == http.MethodGet && r.URL.Path == "/v1/workspaces/"+workspaceID+"/ops/"+opID: + if got := r.Header.Get("Authorization"); got != "Bearer rf_write" { + t.Fatalf("unexpected op Authorization: %q", got) + } + _, _ = io.WriteString(w, `{"opId":"`+opID+`","path":"`+remotePath+`","status":"succeeded","revision":"rev_1"}`) + default: + t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path) + } + })) + defer server.Close() + + installFakeAgentRelaySession(t, server.URL, "cld_access", "demo", workspaceID, workspaceID) + + var stdout bytes.Buffer + if err := run([]string{"writeback", "push", localPath, "--workspace", "demo", "--timeout", "1s"}, strings.NewReader(""), &stdout, &stdout); err != nil { + t.Fatalf("writeback push failed: %v\n%s", err, stdout.String()) + } + if !sawDelegated.Load() { + t.Fatal("expected delegated-token request via agent-relay cloud session") + } + if !sawBulk.Load() { + t.Fatal("expected bulk write request") + } + if got := stdout.String(); !strings.Contains(got, "Pushed ") || !strings.Contains(got, opID) { + t.Fatalf("unexpected stdout: %s", got) + } + if got := countJSONFiles(filepath.Join(localDir, ".relay", "outbox", "pending")); got != 0 { + t.Fatalf("pending receipt count = %d, want 0", got) + } + ackedDir := filepath.Join(localDir, ".relay", "outbox", "acked") + entries, err := os.ReadDir(ackedDir) + if err != nil { + t.Fatalf("read acked dir failed: %v", err) + } + if len(entries) != 1 { + t.Fatalf("acked receipt count = %d, want 1", len(entries)) + } + payload, err := os.ReadFile(filepath.Join(ackedDir, entries[0].Name())) + if err != nil { + t.Fatalf("read acked receipt failed: %v", err) + } + var receipt writebackPushReceipt + if err := json.Unmarshal(payload, &receipt); err != nil { + t.Fatalf("decode acked receipt failed: %v", err) + } + if receipt.Status != "acked" || receipt.OpID != opID || receipt.RemotePath != remotePath || receipt.Revision != "rev_1" { + t.Fatalf("unexpected receipt: %+v", receipt) + } + // Terminal receipts must not retain the user's file body on disk. + if receipt.Content != "" || receipt.Encoding != "" { + t.Fatalf("acked receipt must redact content/encoding, got content=%q encoding=%q", receipt.Content, receipt.Encoding) + } + // The acked JSON printed to stdout must likewise omit the body. + if strings.Contains(stdout.String(), "AR-272 synthetic test") { + t.Fatalf("stdout leaked file content: %s", stdout.String()) + } +} + +func TestWritebackPushRejectsPathWithoutContentIdentity(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + clearRelayfileEnv(t) + + const workspaceID = "ws_demo" + localDir := t.TempDir() + if err := ensureMirrorLayout(localDir); err != nil { + t.Fatalf("ensureMirrorLayout failed: %v", err) + } + if err := writeMirrorStateFile(localDir, syncStateFile{ + WorkspaceID: workspaceID, + RemoteRoot: "/linear/issues", + Mode: defaultMountMode, + }); err != nil { + t.Fatalf("writeMirrorStateFile failed: %v", err) + } + + localPath := filepath.Join(localDir, "comment-ar-272-test.json") + if err := os.WriteFile(localPath, []byte(`{"body":"synthetic comment"}`+"\n"), 0o644); err != nil { + t.Fatalf("write local payload failed: %v", err) + } + + var stdout bytes.Buffer + err := run([]string{"writeback", "push", localPath}, strings.NewReader(""), &stdout, &stdout) + if err == nil { + t.Fatal("expected unsupported path error") + } + if got := err.Error(); !strings.Contains(got, "supports only draft writeback paths and factory-create-*.json") { + t.Fatalf("unexpected error: %v", err) + } + for _, state := range []string{"pending", "acked", "failed"} { + if got := countJSONFiles(filepath.Join(localDir, ".relay", "outbox", state)); got != 0 { + t.Fatalf("%s receipt count = %d, want 0", state, got) + } + } +} + +func TestWorkspaceStatusReportsLocalMountHealth(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + clearRelayfileEnv(t) + + localDir := t.TempDir() + if err := ensureMirrorLayout(localDir); err != nil { + t.Fatalf("ensureMirrorLayout failed: %v", err) + } + if err := writeMirrorStateFile(localDir, syncStateFile{ + WorkspaceID: "ws_demo", + RemoteRoot: "/linear/issues", + Mode: defaultMountMode, + Status: "syncing", + LastSuccessfulReconcileAt: "2026-06-11T15:14:48Z", + LastReconcileAt: "2026-06-15T10:00:00Z", + LastError: &statusError{Message: "changed event not readable yet"}, + }); err != nil { + t.Fatalf("writeMirrorStateFile failed: %v", err) + } + mountState := map[string]any{ + "incrementalReadNotReadySince": map[string]string{ + "/linear/issues/by-state/ready-for-agent/AR-1.json": "2026-06-15T09:00:00Z", + "/linear/issues/by-state/ready-for-agent/AR-2.json": "2026-06-15T09:01:00Z", + }, + "incrementalBacklogDraining": true, + } + mountStatePayload, err := json.Marshal(mountState) + if err != nil { + t.Fatalf("marshal mount state failed: %v", err) + } + if err := os.WriteFile(filepath.Join(localDir, ".relayfile-mount-state.json"), mountStatePayload, 0o644); err != nil { + t.Fatalf("write mount state failed: %v", err) + } + for _, path := range []string{ + filepath.Join(localDir, ".relay", "outbox", "pending", "mountcmd_pending.json"), + filepath.Join(localDir, ".relay", "outbox", "failed", "mountcmd_failed.json"), + filepath.Join(localDir, ".relay", "outbox", "acked", "mountcmd_acked.json"), + } { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatalf("mkdir %s failed: %v", filepath.Dir(path), err) + } + if err := os.WriteFile(path, []byte("{}\n"), 0o644); err != nil { + t.Fatalf("write %s failed: %v", path, err) + } + } + if _, err := upsertWorkspaceDetails(workspaceRecord{ + Name: "demo", + ID: "ws_demo", + LocalDir: localDir, + CreatedAt: time.Now().UTC().Format(time.RFC3339), + LastUsedAt: time.Now().UTC().Format(time.RFC3339), + }); err != nil { + t.Fatalf("upsertWorkspaceDetails failed: %v", err) + } + + var stdout bytes.Buffer + if err := run([]string{"workspace", "status", "--workspace", "demo"}, strings.NewReader(""), &stdout, &stdout); err != nil { + t.Fatalf("workspace status failed: %v\n%s", err, stdout.String()) + } + output := stdout.String() + for _, want := range []string{ + "workspace: ws_demo (demo)", + "status: syncing", + "last successful reconcile: 2026-06-11T15:14:48Z", + "stuck events: 2", + "outbox: pending=1 failed=1 acked=1", + "last error: changed event not readable yet", + "incremental backlog: draining", + } { + if !strings.Contains(output, want) { + t.Fatalf("workspace status output missing %q:\n%s", want, output) + } + } +} + func TestStatusSurfacesDegradedStallReason(t *testing.T) { t.Setenv("HOME", t.TempDir()) clearRelayfileEnv(t) @@ -4620,3 +4893,101 @@ func TestIntegrationConnectKeepsSelectedWorkspaceAndUsesJoinedRelayWorkspaceForS t.Fatalf("expected explicit runtime workspace output, got %q", stdout.String()) } } + +func TestIsWritebackDraftPathSlashSeparated(t *testing.T) { + cases := []struct { + path string + want bool + }{ + {"/linear/issues/factory-create-abc.json", true}, + {"linear/issues/factory-create-abc.json", true}, + {"/linear/issues/by-state/ready-for-agent/factory-create-9f.json", true}, + {"/linear/issues/AR-1.json", false}, + {"/linear/issues/factory-create-abc.txt", false}, + {"/notion/Docs/notes.md", false}, + } + for _, tc := range cases { + if got := isWritebackDraftPath(tc.path); got != tc.want { + t.Errorf("isWritebackDraftPath(%q) = %v, want %v", tc.path, got, tc.want) + } + } +} + +func TestWriteWritebackPushReceiptRedactsTerminalReceipts(t *testing.T) { + mountRoot := t.TempDir() + base := writebackPushReceipt{ + CommandID: "mountcmd_test", + WorkspaceID: "ws_demo", + RemotePath: "/linear/issues/factory-create-x.json", + ContentType: "application/json", + Content: "secret-body", + Encoding: "", + Hash: "deadbeef", + } + + // Pending receipt: retains the body (for retry) and is written 0600. + pending := base + pending.Status = "pending" + if err := writeWritebackPushReceipt(mountRoot, pending); err != nil { + t.Fatalf("write pending receipt: %v", err) + } + pendingPath := filepath.Join(mountRoot, ".relay", "outbox", "pending", "mountcmd_test.json") + info, err := os.Stat(pendingPath) + if err != nil { + t.Fatalf("stat pending receipt: %v", err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Fatalf("pending receipt mode = %o, want 600", perm) + } + var pendingReceipt writebackPushReceipt + readReceiptJSON(t, pendingPath, &pendingReceipt) + if pendingReceipt.Content != "secret-body" { + t.Fatalf("pending receipt must retain content, got %q", pendingReceipt.Content) + } + + // Terminal receipts (acked/failed): body stripped, pending copy removed. + for _, status := range []string{"acked", "failed"} { + r := base + r.Status = status + if err := writeWritebackPushReceipt(mountRoot, r); err != nil { + t.Fatalf("write %s receipt: %v", status, err) + } + p := filepath.Join(mountRoot, ".relay", "outbox", status, "mountcmd_test.json") + var got writebackPushReceipt + readReceiptJSON(t, p, &got) + if got.Content != "" || got.Encoding != "" { + t.Fatalf("%s receipt must redact content/encoding, got content=%q", status, got.Content) + } + if _, err := os.Stat(pendingPath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("writing %s receipt should remove the pending copy; stat err=%v", status, err) + } + // Re-create the pending copy for the next iteration's removal check. + if err := writeWritebackPushReceipt(mountRoot, pending); err != nil { + t.Fatalf("recreate pending receipt: %v", err) + } + } +} + +func TestIsTerminalWritebackOpStatus(t *testing.T) { + for _, s := range []string{"failed", "dead_lettered", "canceled", " failed "} { + if !isTerminalWritebackOpStatus(s) { + t.Errorf("isTerminalWritebackOpStatus(%q) = false, want true", s) + } + } + for _, s := range []string{"", "succeeded", "running", "pending", "dispatched"} { + if isTerminalWritebackOpStatus(s) { + t.Errorf("isTerminalWritebackOpStatus(%q) = true, want false", s) + } + } +} + +func readReceiptJSON(t *testing.T, path string, out *writebackPushReceipt) { + t.Helper() + payload, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read receipt %s: %v", path, err) + } + if err := json.Unmarshal(payload, out); err != nil { + t.Fatalf("decode receipt %s: %v", path, err) + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index eb9d7e22..cdae495c 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -19,6 +19,7 @@ import ( "net/http/pprof" "net/url" "os" + "path" "path/filepath" "runtime" "sort" @@ -227,7 +228,13 @@ type RemoteFile struct { } type WriteResult struct { + OpID string `json:"opId,omitempty"` + Status string `json:"status,omitempty"` TargetRevision string `json:"targetRevision"` + Writeback struct { + Provider string `json:"provider,omitempty"` + State string `json:"state,omitempty"` + } `json:"writeback,omitempty"` } type RemoteClient interface { @@ -979,19 +986,32 @@ type Syncer struct { readNotReadyTTL time.Duration forceFullReconcile bool incrementalCycles int - oversizedLogged map[string]struct{} - quarantinedPaths map[string]struct{} - lazyRepos bool - lowMemory bool - writeOnly bool - layoutRegistrar ProviderLayoutRegistrar - githubWorkingTree *githubWorkingTreeMount - closeScheduler *CloseScheduler - rollingCoalescer *RollingDigestCoalescer - circuit *CloudErrorCircuit - maxOutboxAttempts int - nowFn func() time.Time - mu sync.Mutex + // staleAliasSkips counts stale provider-layout-alias events drained in the + // current reconcile cycle. Reset at the start of each cycle and surfaced to + // the CLI (e.g. `relayfile pull` summary, `workspace status`) so operators + // can see the stuck-event drain making progress. + staleAliasSkips int + // skip-stuck escape hatch (`relayfile writeback skip-stuck`). When + // skipStuckMode is set, the incremental drain drops every read-404 event + // immediately — not just provider-layout-alias paths — without waiting the + // read-not-ready TTL. skipStuckMax bounds the number of events dropped + // (0 = unbounded); skipStuckCount tracks how many were dropped. + skipStuckMode bool + skipStuckMax int + skipStuckCount int + oversizedLogged map[string]struct{} + quarantinedPaths map[string]struct{} + lazyRepos bool + lowMemory bool + writeOnly bool + layoutRegistrar ProviderLayoutRegistrar + githubWorkingTree *githubWorkingTreeMount + closeScheduler *CloseScheduler + rollingCoalescer *RollingDigestCoalescer + circuit *CloudErrorCircuit + maxOutboxAttempts int + nowFn func() time.Time + mu sync.Mutex } // now returns the current time using the injected clock when set (tests), @@ -1506,6 +1526,50 @@ func (s *Syncer) Reconcile(ctx context.Context) error { return s.sync(ctx, true) } +// StaleAliasSkips returns the number of stale provider-layout-alias events the +// most recent reconcile cycle drained without waiting the read-not-ready TTL. +// Operators use this (via `relayfile pull` / `workspace status`) to confirm the +// stuck-event drain is making progress. +func (s *Syncer) StaleAliasSkips() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.staleAliasSkips +} + +// BacklogDraining reports whether the events feed still has unprocessed pages +// after the most recent cycle (e.g. the cycle was canceled mid-drain). The CLI +// uses this to tell the operator to re-run or run `writeback skip-stuck`. +func (s *Syncer) BacklogDraining() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.state.IncrementalBacklogDraining +} + +// SkipStuck is the operator escape hatch for a wedged events cursor. It runs a +// reconcile cycle that drops every read-404 ("not readable yet") event +// immediately — without waiting the read-not-ready TTL — so the cursor walks +// forward past consecutive stuck events in one invocation. max bounds the +// number of events dropped (0 = unbounded). It returns the number of stuck +// events skipped along with any reconcile error. +func (s *Syncer) SkipStuck(ctx context.Context, max int) (int, error) { + s.mu.Lock() + s.skipStuckMode = true + s.skipStuckMax = max + s.skipStuckCount = 0 + s.mu.Unlock() + defer func() { + s.mu.Lock() + s.skipStuckMode = false + s.skipStuckMax = 0 + s.mu.Unlock() + }() + err := s.sync(ctx, true) + s.mu.Lock() + count := s.skipStuckCount + s.mu.Unlock() + return count, err +} + // FlushOutboxOnce uploads only persisted durable outbox records and exits // without reconciling the local mirror. It is intentionally O(outbox): no // local tree scan, pushLocal, pullRemote, websocket, or digest work. @@ -2024,12 +2088,20 @@ func bulkWriteFilesForPending(workspaceID string, pending []pendingBulkWrite) [] } func mountWritebackCreateDraftContentIdentity(workspaceID, normalizedRemotePath, contentHash string) *ContentIdentity { - if !relayfile.IsDraftFilePath(normalizedRemotePath) { + if !isMountWritebackCreateDraftPath(normalizedRemotePath) { return nil } return newMountWritebackCreateDraftContentIdentity(workspaceID, normalizedRemotePath, contentHash) } +func isMountWritebackCreateDraftPath(remotePath string) bool { + if relayfile.IsDraftFilePath(remotePath) { + return true + } + base := path.Base(normalizeRemotePath(remotePath)) + return strings.HasPrefix(base, "factory-create-") && strings.HasSuffix(base, ".json") +} + func newMountWritebackCreateDraftContentIdentity(workspaceID, normalizedRemotePath, contentHash string) *ContentIdentity { return &ContentIdentity{ Kind: mountWritebackCreateDraftContentIdentityKind, @@ -2441,6 +2513,7 @@ func (s *Syncer) sync(ctx context.Context, forcePoll bool) error { s.mu.Lock() defer s.mu.Unlock() s.markReconcileStarted() + s.staleAliasSkips = 0 if err := s.runClosingDigestJobsLocked(ctx); err != nil { s.markSyncError(err) _ = s.saveState() @@ -4393,6 +4466,23 @@ func isProviderLayoutAliasSegment(segment string) bool { return false } +// isProviderLayoutAliasRemotePath reports whether remotePath lives under a +// provider layout alias index (by-state, by-id, by-title, ...). Those paths are +// derived index views: the emitter removes the alias entry when the underlying +// record changes state, but the events feed may still carry the now-dangling +// path. A read 404 on such a path means the event is stale, not slow — the +// alias mirror is reconstructable from the canonical record and the periodic +// full pull — so the cycle can skip it immediately instead of holding the +// events cursor for the full read-not-ready TTL. +func isProviderLayoutAliasRemotePath(remotePath string) bool { + for _, segment := range strings.Split(strings.Trim(normalizeRemotePath(remotePath), "/"), "/") { + if isProviderLayoutAliasSegment(segment) { + return true + } + } + return false +} + func (s *Syncer) pullRemoteIncremental(ctx context.Context, conflicted map[string]struct{}, cursor string) (string, error) { currentCursor := strings.TrimSpace(cursor) safeCursor := currentCursor @@ -4562,6 +4652,48 @@ func (s *Syncer) applyIncrementalChanges( if err != nil { var httpErr *HTTPError if errors.As(err, &httpErr) && httpErr.StatusCode == http.StatusNotFound { + // Operator escape hatch (`relayfile writeback skip-stuck`): drop + // every unreadable event immediately, regardless of path shape + // or TTL, until the optional max budget is exhausted. Once the + // budget is hit, preserve the cursor so a follow-up run resumes + // from here. + if s.skipStuckMode { + if s.skipStuckMax > 0 && s.skipStuckCount >= s.skipStuckMax { + s.logf("skip-stuck: reached max=%d skipped events; preserving events cursor for next run", s.skipStuckMax) + return &IncrementalReadNotReadyError{ + Path: remotePath, + StatusCode: httpErr.StatusCode, + Code: httpErr.Code, + Message: httpErr.Message, + } + } + s.skipStuckCount++ + s.logf("skip-stuck: dropping unreadable event for %s (404); advancing events cursor", remotePath) + if err := s.applyRemoteDelete(remotePath, conflicted); err != nil { + return err + } + s.markIncrementalCheckpoint(pageStartCursor, pageCursor, "changed", remotePath) + continue + } + // In-cycle drain of the stuck-event class. A read 404 on a + // provider-layout-alias path (by-state, by-id, ...) is a stale + // index event: the emitter dropped the alias entry when the + // record changed state, but the events feed still carries the + // dangling path. Skip it immediately and keep draining the rest + // of the page in this same cycle rather than holding the cursor + // for the full read-not-ready TTL and exiting after one event. + // Canonical records are reconstructable independently, so this + // cannot lose committed data; the periodic full pull self-heals + // any alias mirror that should still exist. + if isProviderLayoutAliasRemotePath(remotePath) { + s.staleAliasSkips++ + s.logf("changed event for %s is a stale index-alias path (read 404); skipping immediately and continuing drain", remotePath) + if err := s.applyRemoteDelete(remotePath, conflicted); err != nil { + return err + } + s.markIncrementalCheckpoint(pageStartCursor, pageCursor, "changed", remotePath) + continue + } if s.incrementalReadNotReadyExpired(remotePath, time.Now().UTC()) { s.logf("changed event for %s remained unreadable for at least %s; treating as deleted and advancing events cursor", remotePath, s.readNotReadyTTL) if err := s.applyRemoteDelete(remotePath, conflicted); err != nil { diff --git a/internal/mountsync/syncer_test.go b/internal/mountsync/syncer_test.go index 4e505ebd..a158217a 100644 --- a/internal/mountsync/syncer_test.go +++ b/internal/mountsync/syncer_test.go @@ -1905,6 +1905,14 @@ func TestBulkWrite_ContentIdentityOnlyForCreateDraftPaths(t *testing.T) { t.Fatal("expected content identity for space-uuid create draft path") } + factoryCreate := bulkWriteFilesForPending(workspaceID, []pendingBulkWrite{{ + remotePath: "/linear/issues/factory-create-ar-272-test.json", + snapshot: snapshot, + }}) + if factoryCreate[0].ContentIdentity == nil { + t.Fatal("expected content identity for factory-create writeback path") + } + stable := bulkWriteFilesForPending(workspaceID, []pendingBulkWrite{{ remotePath: "/notion/pages/page-1.md", snapshot: snapshot, @@ -7516,6 +7524,135 @@ func TestPullRemoteIncrementalChangedPath404RetriesWithoutAdvancingCursor(t *tes assertLocalFileContent(t, filepath.Join(localDir, "Docs", "003.md"), "# 003") } +// TestPullRemoteIncrementalDrainsStaleAliasEventsInOneCycle verifies the +// stuck-event drain (AR-272 Part 2a): a read 404 on a provider-layout-alias +// path (by-state index) is treated as a stale event and skipped immediately, +// so a single cycle chews through several consecutive stuck events instead of +// exiting after the first one. Real (non-alias) records on either side of the +// stuck run are still applied. +func TestPullRemoteIncrementalDrainsStaleAliasEventsInOneCycle(t *testing.T) { + files := map[string]RemoteFile{ + "/linear/issues/AR-1.json": { + Path: "/linear/issues/AR-1.json", + Revision: "rev_001", + ContentType: "application/json", + Content: `{"id":"AR-1"}`, + }, + "/linear/issues/AR-4.json": { + Path: "/linear/issues/AR-4.json", + Revision: "rev_004", + ContentType: "application/json", + Content: `{"id":"AR-4"}`, + }, + } + client := &fakeClient{ + files: files, + events: []FilesystemEvent{ + {EventID: "evt_001", Type: "file.updated", Path: "/linear/issues/AR-1.json", Revision: "rev_001"}, + // Two by-state alias paths the emitter dropped when the issues + // left ready-for-agent — the events feed still carries them, and + // reads 404. These are the stuck-event class. + {EventID: "evt_002", Type: "file.updated", Path: "/linear/issues/by-state/ready-for-agent/AR-2.json", Revision: "rev_002"}, + {EventID: "evt_003", Type: "file.updated", Path: "/linear/issues/by-state/ready-for-agent/AR-3.json", Revision: "rev_003"}, + {EventID: "evt_004", Type: "file.updated", Path: "/linear/issues/AR-4.json", Revision: "rev_004"}, + }, + revisionCounter: 4, + eventCounter: 4, + } + localDir := t.TempDir() + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_alias_drain", + RemoteRoot: "/linear", + LocalRoot: localDir, + FullPullEvery: -1, + WebSocket: boolPtr(false), + CursorTimeout: time.Second, + BootstrapTimeout: time.Second, + }) + if err != nil { + t.Fatalf("new syncer failed: %v", err) + } + syncer.loaded = true + syncer.state = mountState{ + Files: map[string]trackedFile{}, + EventsCursor: "evt_000", + BootstrapComplete: true, + } + + if err := syncer.Reconcile(context.Background()); err != nil { + t.Fatalf("expected stale alias events to drain in one cycle, got %v", err) + } + if got := syncer.state.EventsCursor; got != "evt_004" { + t.Fatalf("EventsCursor = %q, want fully-drained evt_004", got) + } + if got := syncer.StaleAliasSkips(); got != 2 { + t.Fatalf("StaleAliasSkips = %d, want 2", got) + } + if syncer.state.IncrementalReadNotReadySince != nil { + t.Fatalf("expected no read-not-ready timestamps for stale alias paths, got %#v", syncer.state.IncrementalReadNotReadySince) + } + assertLocalFileContent(t, filepath.Join(localDir, "issues", "AR-1.json"), `{"id":"AR-1"}`) + assertLocalFileContent(t, filepath.Join(localDir, "issues", "AR-4.json"), `{"id":"AR-4"}`) +} + +// TestSkipStuckDropsConsecutiveUnreadableEvents verifies the operator escape +// hatch (AR-272 Part 2b): SkipStuck advances the events cursor past every +// read-404 event — including non-alias canonical paths — without waiting the +// read-not-ready TTL, and reports the number skipped. +func TestSkipStuckDropsConsecutiveUnreadableEvents(t *testing.T) { + client := &fakeClient{ + files: map[string]RemoteFile{ + "/notion/Docs/004.md": { + Path: "/notion/Docs/004.md", + Revision: "rev_004", + ContentType: "text/markdown", + Content: "# 004", + }, + }, + events: []FilesystemEvent{ + // Non-alias canonical paths that 404 — normally these retry for + // the full TTL; skip-stuck drops them immediately. + {EventID: "evt_001", Type: "file.updated", Path: "/notion/Docs/001.md", Revision: "rev_001"}, + {EventID: "evt_002", Type: "file.updated", Path: "/notion/Docs/002.md", Revision: "rev_002"}, + {EventID: "evt_003", Type: "file.updated", Path: "/notion/Docs/003.md", Revision: "rev_003"}, + {EventID: "evt_004", Type: "file.updated", Path: "/notion/Docs/004.md", Revision: "rev_004"}, + }, + revisionCounter: 4, + eventCounter: 4, + } + localDir := t.TempDir() + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_skip_stuck", + RemoteRoot: "/notion", + LocalRoot: localDir, + FullPullEvery: -1, + WebSocket: boolPtr(false), + CursorTimeout: time.Second, + BootstrapTimeout: time.Second, + }) + if err != nil { + t.Fatalf("new syncer failed: %v", err) + } + syncer.loaded = true + syncer.state = mountState{ + Files: map[string]trackedFile{}, + EventsCursor: "evt_000", + BootstrapComplete: true, + } + + skipped, err := syncer.SkipStuck(context.Background(), 0) + if err != nil { + t.Fatalf("SkipStuck failed: %v", err) + } + if skipped != 3 { + t.Fatalf("skipped = %d, want 3 unreadable events dropped", skipped) + } + if got := syncer.state.EventsCursor; got != "evt_004" { + t.Fatalf("EventsCursor = %q, want evt_004 (caught up to head)", got) + } + assertLocalFileContent(t, filepath.Join(localDir, "Docs", "004.md"), "# 004") +} + func TestPullRemoteIncrementalCreatedThreadReply404RetriesWithoutAdvancingCursor(t *testing.T) { const replyPath = "/slack/channels/C123ABC__proj-cloud/threads/1780871788_370329/replies/1780914176_827829.json" client := &fakeClient{ diff --git a/package-lock.json b/package-lock.json index 82750239..4198dd1a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "relayfile", - "version": "0.8.28", + "version": "0.8.29", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "relayfile", - "version": "0.8.28", + "version": "0.8.29", "license": "Apache-2.0", "workspaces": [ "packages/core", @@ -1999,9 +1999,9 @@ "link": true }, "node_modules/@relayfile/mount-darwin-arm64": { - "version": "0.8.28", - "resolved": "https://registry.npmjs.org/@relayfile/mount-darwin-arm64/-/mount-darwin-arm64-0.8.28.tgz", - "integrity": "sha512-M6Yhqz6TAw/VgAh5URmFb0sMvjTbHDC6qjz6sjifkXpFfqn0v4gJsLeuYYN1rh2/e5KInTTk+QvwplIlMa17GA==", + "version": "0.8.29", + "resolved": "https://registry.npmjs.org/@relayfile/mount-darwin-arm64/-/mount-darwin-arm64-0.8.29.tgz", + "integrity": "sha512-BuxmPfMz/kTGPrb5dblwHf7Q3w08t2edbtNpfKjd7DvNOLiltRz8Dek5hw0ogeWU3GeQUQYmgGFE38XvpsAKCA==", "cpu": [ "arm64" ], @@ -2012,9 +2012,9 @@ ] }, "node_modules/@relayfile/mount-darwin-x64": { - "version": "0.8.28", - "resolved": "https://registry.npmjs.org/@relayfile/mount-darwin-x64/-/mount-darwin-x64-0.8.28.tgz", - "integrity": "sha512-017dDLPCyF9OJIRAadwGd3z/SricGMucnqIakA1kVS82UJQYSZR1U6kq6mA0l7Fyz/he5JvQ8U9A6q1EwSEolw==", + "version": "0.8.29", + "resolved": "https://registry.npmjs.org/@relayfile/mount-darwin-x64/-/mount-darwin-x64-0.8.29.tgz", + "integrity": "sha512-hC8Ym8Izwur1Et0of2J+AxiVxEHzoXvL01oMhLwCQC/EVZwCjy5AcyrU4d0jc7pOck7rRQ4ga5W/gXE9DM9sYQ==", "cpu": [ "x64" ], @@ -2025,9 +2025,9 @@ ] }, "node_modules/@relayfile/mount-linux-arm64": { - "version": "0.8.28", - "resolved": "https://registry.npmjs.org/@relayfile/mount-linux-arm64/-/mount-linux-arm64-0.8.28.tgz", - "integrity": "sha512-vdPX0w4JQhbF5xUS2ep+PwRaneKmiA9Ue7c/rVC/DWpxfLb1nKVW9Avb/T5I648CefNHJ+nlRyOz66NnWUpxMQ==", + "version": "0.8.29", + "resolved": "https://registry.npmjs.org/@relayfile/mount-linux-arm64/-/mount-linux-arm64-0.8.29.tgz", + "integrity": "sha512-Rm+r1LCcwNR7o8Hj12xXIjDjw2lLsFBk6XNpfSG/fu0kgihQibpEvMp6ZCkli4xn6XM6T9uZT/e7s6XKv2e4mQ==", "cpu": [ "arm64" ], @@ -2038,9 +2038,9 @@ ] }, "node_modules/@relayfile/mount-linux-x64": { - "version": "0.8.28", - "resolved": "https://registry.npmjs.org/@relayfile/mount-linux-x64/-/mount-linux-x64-0.8.28.tgz", - "integrity": "sha512-lv/YCFEjmQkNBZQ+AFa9ChW3wqDfPlelINU1tviRcLms1R5cXpp0/HERo152VtAq8WyJIlPcauJfI1kjxnTYZQ==", + "version": "0.8.29", + "resolved": "https://registry.npmjs.org/@relayfile/mount-linux-x64/-/mount-linux-x64-0.8.29.tgz", + "integrity": "sha512-tiWdfM2/jIRsL/mMmfsjJLKZl1SSLSANglQJUXMa8eANTwB/ItxJ5mwUz7ryuWoBSB+43beWK8Euj2zHDDlicQ==", "cpu": [ "x64" ], @@ -5287,7 +5287,7 @@ }, "packages/cli": { "name": "relayfile", - "version": "0.8.28", + "version": "0.8.29", "hasInstallScript": true, "license": "MIT", "bin": { @@ -5299,7 +5299,7 @@ }, "packages/core": { "name": "@relayfile/core", - "version": "0.8.28", + "version": "0.8.29", "license": "MIT", "devDependencies": { "@types/node": "^22.0.0", @@ -5312,7 +5312,7 @@ }, "packages/file-observer": { "name": "@relayfile/file-observer", - "version": "0.8.28", + "version": "0.8.29", "license": "MIT", "dependencies": { "class-variance-authority": "^0.7.0", @@ -6120,7 +6120,7 @@ }, "packages/local-mount": { "name": "@relayfile/local-mount", - "version": "0.8.28", + "version": "0.8.29", "license": "MIT", "dependencies": { "@parcel/watcher": "^2.5.6", @@ -6149,10 +6149,10 @@ }, "packages/sdk/typescript": { "name": "@relayfile/sdk", - "version": "0.8.28", + "version": "0.8.29", "license": "MIT", "dependencies": { - "@relayfile/core": "0.8.28", + "@relayfile/core": "0.8.29", "ignore": "^7.0.5", "tar": "^7.5.10" }, @@ -6164,10 +6164,10 @@ "node": ">=18" }, "optionalDependencies": { - "@relayfile/mount-darwin-arm64": "0.8.28", - "@relayfile/mount-darwin-x64": "0.8.28", - "@relayfile/mount-linux-arm64": "0.8.28", - "@relayfile/mount-linux-x64": "0.8.28" + "@relayfile/mount-darwin-arm64": "0.8.29", + "@relayfile/mount-darwin-x64": "0.8.29", + "@relayfile/mount-linux-arm64": "0.8.29", + "@relayfile/mount-linux-x64": "0.8.29" } } }