diff --git a/internal/agenteval/suite.go b/internal/agenteval/suite.go index 03d7eafbc..638627e1e 100644 --- a/internal/agenteval/suite.go +++ b/internal/agenteval/suite.go @@ -186,10 +186,6 @@ func validateFileList(taskPath string, field string, files []string, required bo return problems } -func validateExpectedChangedFiles(taskPath string, files []string) []string { - return validateFileList(taskPath, "expectedChangedFiles", files, true) -} - func validateStringList(taskPath string, field string, values []string) []string { problems := []string{} seen := map[string]int{} diff --git a/internal/cli/auth.go b/internal/cli/auth.go index d744e34bd..2dc20d327 100644 --- a/internal/cli/auth.go +++ b/internal/cli/auth.go @@ -129,19 +129,18 @@ func validateAuthFlags(sub string, a authArgs) error { // ZERO_OAUTH_STORAGE=encrypted-file selects the AES-256-GCM encrypted-at-rest // backend (a per-user secret is created beside the token file). func newAuthManager(deps appDeps, out io.Writer) (*oauth.Manager, error) { - // Validate ZERO_OAUTH_STORAGE up front: a mistyped non-empty value must fail - // fast rather than silently fall back to plaintext while the user believes - // encryption is on. Empty = default (plaintext 0600); "encrypted-file" = AES. - encrypted := false - if mode := strings.TrimSpace(os.Getenv("ZERO_OAUTH_STORAGE")); mode != "" { - if !strings.EqualFold(mode, "encrypted-file") { - return nil, fmt.Errorf("invalid ZERO_OAUTH_STORAGE %q (supported: encrypted-file)", mode) - } - encrypted = true + // Validate ZERO_OAUTH_STORAGE up front: a mistyped value must fail fast rather + // than silently change the backend. Empty = default (plaintext 0600 file); + // "encrypted-file" = AES-256-GCM; "keyring" = the OS keyring. + storage := strings.ToLower(strings.TrimSpace(os.Getenv("ZERO_OAUTH_STORAGE"))) + switch storage { + case "", "file", "encrypted-file", "keyring": + default: + return nil, fmt.Errorf("invalid ZERO_OAUTH_STORAGE %q (supported: file, encrypted-file, keyring)", storage) } store, err := oauth.NewStore(oauth.StoreOptions{ - Now: deps.now, - Encrypted: encrypted, + Now: deps.now, + Storage: storage, }) if err != nil { return nil, err @@ -350,8 +349,10 @@ built in). For a provider named , set: Endpoint URLs must be https (loopback exempt). Storage: tokens are written 0600 under $XDG_CONFIG_HOME/zero (override with -ZERO_OAUTH_TOKENS_PATH). Set ZERO_OAUTH_STORAGE=encrypted-file to encrypt them -at rest with AES-256-GCM (a per-user secret is created beside the token file). +ZERO_OAUTH_TOKENS_PATH). Set ZERO_OAUTH_STORAGE=encrypted-file to encrypt them at +rest with AES-256-GCM (a per-user secret beside the file), or +ZERO_OAUTH_STORAGE=keyring to use the OS keyring (macOS Keychain / Linux +secret-tool). MCP server tokens share the same store. Flags: --device Use the device-code flow (headless/SSH; no browser) diff --git a/internal/cli/daemon.go b/internal/cli/daemon.go index 79995ead5..5074312e4 100644 --- a/internal/cli/daemon.go +++ b/internal/cli/daemon.go @@ -38,6 +38,8 @@ func runDaemon(args []string, stdout io.Writer, stderr io.Writer, _ appDeps) int return runDaemonAttach(rest, stdout, stderr) case "serve-remote": return runDaemonServeRemote(rest, stdout, stderr) + case "link": + return runDaemonLink(rest, stdout, stderr) case "-h", "--help", "help": return writeDaemonUsage(stdout, exitSuccess) default: @@ -56,14 +58,20 @@ Commands: run --session [--cwd ] [--prompt ] [exec flags...] Create/route a session and stream its output. attach Attach to a running session's stream. - serve-remote --addr --tls-cert --tls-key + serve-remote --addr --tls-cert --tls-key [--bundle-dir ] Serve an opt-in, TLS-only network bridge to this daemon. Requires a bearer token in $ZERO_DAEMON_REMOTE_TOKEN - (or $ZERO_DAEMON_REMOTE_TOKEN_FILE). + (or $ZERO_DAEMON_REMOTE_TOKEN_FILE). --bundle-dir enables + git-bundle uploads, extracted into per-link work trees. + link --remote --repo --id [--out ] + Upload repo's git history to the remote as a bundle and + print the extracted remote path. --out saves a session + link file (0600). Accepts --token/--ca-cert/--server-name. + link --show Print a saved session link. run and attach accept --remote [--token ] [--ca-cert ] [--server-name ] to drive a remote daemon over the bridge instead of the -local socket. +local socket. Use the link's remote path as --cwd to run against a linked repo. `) return code } @@ -399,7 +407,7 @@ func daemonDialError(flags remoteDialFlags, err error) string { // bridge. TLS and a bearer token are mandatory (fail closed): it refuses to // start without a cert/key pair and a token from the environment. func runDaemonServeRemote(args []string, stdout io.Writer, stderr io.Writer) int { - addr, certFile, keyFile := "", "", "" + addr, certFile, keyFile, bundleDir := "", "", "", "" minVersion, maxConns := 0, 0 for i := 0; i < len(args); i++ { a := args[i] @@ -449,6 +457,14 @@ func runDaemonServeRemote(args []string, stdout io.Writer, stderr io.Writer) int return writeExecUsageError(stderr, "--max-conns requires a value") } maxConns = atoiOrZero(v) + case a == "--bundle-dir": + v, ok := value() + if !ok { + return writeExecUsageError(stderr, "--bundle-dir requires a value") + } + bundleDir = v + case strings.HasPrefix(a, "--bundle-dir="): + bundleDir = strings.TrimPrefix(a, "--bundle-dir=") default: return writeExecUsageError(stderr, fmt.Sprintf("unknown flag %q for daemon serve-remote", a)) } @@ -491,7 +507,7 @@ func runDaemonServeRemote(args []string, stdout io.Writer, stderr io.Writer) int return writeAppError(stderr, err.Error(), exitCrash) } bridge, err := remote.NewBridge(remote.BridgeOptions{ - Server: srv, Authenticator: auth, MinVersion: minVersion, MaxConnections: maxConns, Log: logf, + Server: srv, Authenticator: auth, MinVersion: minVersion, MaxConnections: maxConns, BundleDir: bundleDir, Log: logf, }) if err != nil { return writeAppError(stderr, err.Error(), exitCrash) @@ -524,6 +540,75 @@ func runDaemonServeRemote(args []string, stdout io.Writer, stderr io.Writer) int } } +// runDaemonLink uploads a repo's git history to a remote bridge as a bundle +// (link without --show), or prints a saved session link (--show ). +func runDaemonLink(args []string, stdout io.Writer, stderr io.Writer) int { + var addr, repo, id, token, caCert, serverName, out, show string + // flags maps each "--name" to the string it sets; both "--name v" and + // "--name=v" forms are accepted. + flags := map[string]*string{ + "--remote": &addr, "--repo": &repo, "--id": &id, "--token": &token, + "--ca-cert": &caCert, "--server-name": &serverName, "--out": &out, "--show": &show, + } + for i := 0; i < len(args); i++ { + a := args[i] + if a == "-h" || a == "--help" { + return writeDaemonUsage(stdout, exitSuccess) + } + name, inlineVal, hasInline := a, "", false + if eq := strings.IndexByte(a, '='); eq >= 0 { + name, inlineVal, hasInline = a[:eq], a[eq+1:], true + } + dst, ok := flags[name] + if !ok { + return writeExecUsageError(stderr, fmt.Sprintf("unknown flag %q for daemon link", a)) + } + if hasInline { + *dst = inlineVal + continue + } + if i+1 >= len(args) { + return writeExecUsageError(stderr, fmt.Sprintf("%s requires a value", name)) + } + i++ + *dst = args[i] + } + + if strings.TrimSpace(show) != "" { + link, err := remote.LoadSessionLink(show) + if err != nil { + return writeAppError(stderr, err.Error(), exitCrash) + } + fmt.Fprintf(stdout, "link %s -> %s on %s\n", link.LinkID, link.RemotePath, link.Address) + return exitSuccess + } + + if strings.TrimSpace(addr) == "" || strings.TrimSpace(repo) == "" || strings.TrimSpace(id) == "" { + return writeExecUsageError(stderr, "daemon link requires --remote, --repo, and --id (or --show )") + } + if strings.TrimSpace(token) == "" { + token, _ = remote.TokenFromEnv() // best effort; UploadRepoBundle rejects an empty token + } + link, err := remote.UploadRepoBundle(remote.RemoteConfig{ + Address: addr, + Token: token, + CACertFile: caCert, + ServerName: serverName, + }, repo, id) + if err != nil { + return writeAppError(stderr, err.Error(), exitCrash) + } + fmt.Fprintf(stdout, "uploaded %s; remote repo at %s\n", link.LinkID, link.RemotePath) + fmt.Fprintf(stdout, "run it with: zero daemon run --remote %s --cwd %s ...\n", link.Address, link.RemotePath) + if strings.TrimSpace(out) != "" { + if err := link.Save(out); err != nil { + return writeAppError(stderr, err.Error(), exitCrash) + } + fmt.Fprintf(stdout, "saved session link to %s\n", out) + } + return exitSuccess +} + // remoteDialFlags holds the optional flags that redirect run/attach to a remote // daemon. When Addr is empty the local control socket is used. type remoteDialFlags struct { diff --git a/internal/cli/exec.go b/internal/cli/exec.go index 3ef6c7e76..f87b4a652 100644 --- a/internal/cli/exec.go +++ b/internal/cli/exec.go @@ -367,6 +367,12 @@ func runExec(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) in Mode: notify.Mode(strings.TrimSpace(execNotifyMode(options, resolved))), FocusMode: notify.FocusAlways, }) + // Opt-in webhook fan-out (ZERO_NOTIFY_WEBHOOK_URL). Headless runs can safely + // log a failed delivery to stderr (never stdout). The sink redacts before + // logging, so a token in the URL or message is masked. + notify.MaybeAddWebhookSink(notifier, os.Getenv, func(format string, args ...any) { + fmt.Fprintf(stderr, "[notify] "+format+"\n", args...) + }) if options.useSpec { return runExecSpecDraft(execSpecDraftRun{ options: options, diff --git a/internal/daemon/remote/auth.go b/internal/daemon/remote/auth.go index 1fbed66f7..4f1574a9d 100644 --- a/internal/daemon/remote/auth.go +++ b/internal/daemon/remote/auth.go @@ -96,12 +96,23 @@ type noopAttestation struct{} func (noopAttestation) Verify(map[string]string) error { return nil } +// Connection modes negotiated in the auth handshake. The default (empty) is a +// daemon session, preserving the original behavior; "bundle" requests a one-shot +// git-bundle upload instead of a session. +const ( + ModeSession = "session" + ModeBundle = "bundle" +) + // authRequest is the first frame a remote client sends (before the daemon // hello). Token is never logged. type authRequest struct { Token string `json:"token"` Version int `json:"version"` Meta map[string]string `json:"meta,omitempty"` + // Mode selects the connection's purpose: "" / "session" => a daemon session + // (default), "bundle" => a git-bundle upload. Unknown modes are rejected. + Mode string `json:"mode,omitempty"` } // authResponse is the bridge's reply to the auth handshake. diff --git a/internal/daemon/remote/bridge.go b/internal/daemon/remote/bridge.go index 28bf51bb0..ef4057acc 100644 --- a/internal/daemon/remote/bridge.go +++ b/internal/daemon/remote/bridge.go @@ -16,6 +16,9 @@ const ( defaultMaxConnections = 32 defaultHandshakeTimeout = 10 * time.Second defaultAuthFailDelay = 250 * time.Millisecond + // defaultMaxBundleBytes caps a single uploaded git bundle (256 MiB) so a + // remote peer cannot exhaust the host's disk. + defaultMaxBundleBytes = 256 << 20 ) // Bridge serves authenticated remote connections and drives the local daemon's @@ -27,6 +30,8 @@ type Bridge struct { minVersion int handshakeTimeout time.Duration authFailDelay time.Duration + bundleDir string + maxBundleBytes int64 log func(string) sem chan struct{} @@ -51,7 +56,13 @@ type BridgeOptions struct { HandshakeTimeout time.Duration // AuthFailDelay slows brute-force attempts; <0 => none, 0 => default. AuthFailDelay time.Duration - Log func(string) + // BundleDir is the directory under which uploaded git bundles are extracted + // into per-link working trees. Empty disables bundle uploads entirely (a + // bundle-mode connection is then refused — opt-in, fail closed). + BundleDir string + // MaxBundleBytes caps a single uploaded bundle; 0 => default. + MaxBundleBytes int64 + Log func(string) } // NewBridge validates options and builds a Bridge. @@ -85,6 +96,10 @@ func NewBridge(opts BridgeOptions) (*Bridge, error) { if attest == nil { attest = noopAttestation{} } + maxBundleBytes := opts.MaxBundleBytes + if maxBundleBytes <= 0 { + maxBundleBytes = defaultMaxBundleBytes + } return &Bridge{ server: opts.Server, auth: opts.Authenticator, @@ -92,6 +107,8 @@ func NewBridge(opts BridgeOptions) (*Bridge, error) { minVersion: minVersion, handshakeTimeout: handshakeTimeout, authFailDelay: authFailDelay, + bundleDir: strings.TrimSpace(opts.BundleDir), + maxBundleBytes: maxBundleBytes, log: opts.Log, sem: make(chan struct{}, maxConns), }, nil @@ -179,13 +196,37 @@ func (b *Bridge) handle(conn net.Conn) { b.deny(conn, "attestation failed") return } + // Resolve the connection mode before accepting it. Validating here (before the + // success response) lets the bridge fail closed: an unknown mode, or a bundle + // upload when bundle transfer is disabled, is denied without side effects. + mode := req.Mode + if mode == "" { + mode = ModeSession + } + switch mode { + case ModeSession: + case ModeBundle: + if b.bundleDir == "" { + b.deny(conn, "bundle transfer not enabled") + return + } + default: + b.deny(conn, "unsupported mode") + return + } if err := writeAuthResponse(conn, authResponse{OK: true, Version: daemon.ProtoVersion}); err != nil { _ = conn.Close() return } - // Clear the handshake deadline: a session may stream for a long time. + // Clear the handshake deadline: a session may stream, and a bundle upload may + // transfer many megabytes, for a long time. _ = conn.SetDeadline(time.Time{}) - b.server.ServeConn(conn) // performs the daemon handshake + one command, then closes conn + switch mode { + case ModeBundle: + b.handleBundle(conn) // receives + extracts the bundle, then closes conn + default: + b.server.ServeConn(conn) // performs the daemon handshake + one command, then closes conn + } } // deny rejects an unauthenticated connection after a small backoff (to slow diff --git a/internal/daemon/remote/bundle.go b/internal/daemon/remote/bundle.go new file mode 100644 index 000000000..398c1bcb2 --- /dev/null +++ b/internal/daemon/remote/bundle.go @@ -0,0 +1,378 @@ +package remote + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/Gitlawb/zero/internal/daemon" +) + +// gitTimeout bounds a single git invocation (bundle create/verify, clone) so a +// hung git process cannot pin a connection or the upload path indefinitely. +const gitTimeout = 2 * time.Minute + +// bundleChunkSize is the per-frame payload when streaming a bundle file. It is +// kept comfortably under daemon.MaxFrameSize (1 MiB). +const bundleChunkSize = 512 << 10 + +// bundleHeader is the first control frame of a bundle upload (after the auth +// handshake): it declares the link id and the exact byte size that follows. +type bundleHeader struct { + LinkID string `json:"link_id"` + Size int64 `json:"size"` +} + +// bundleResult is the bridge's reply once the bundle is received and extracted. +type bundleResult struct { + OK bool `json:"ok"` + Path string `json:"path,omitempty"` + Message string `json:"message,omitempty"` +} + +func writeBundleHeader(w io.Writer, h bundleHeader) error { + payload, err := json.Marshal(h) + if err != nil { + return err + } + return daemon.WriteFrame(w, daemon.KindCtrl, payload) +} + +func readBundleHeader(r io.Reader) (bundleHeader, error) { + kind, payload, err := daemon.ReadFrame(r) + if err != nil { + return bundleHeader{}, err + } + if kind != daemon.KindCtrl { + return bundleHeader{}, errors.New("remote: expected bundle header frame") + } + var h bundleHeader + if err := json.Unmarshal(payload, &h); err != nil { + return bundleHeader{}, fmt.Errorf("remote: decode bundle header: %w", err) + } + return h, nil +} + +func writeBundleResult(w io.Writer, res bundleResult) error { + payload, err := json.Marshal(res) + if err != nil { + return err + } + return daemon.WriteFrame(w, daemon.KindCtrl, payload) +} + +func readBundleResult(r io.Reader) (bundleResult, error) { + kind, payload, err := daemon.ReadFrame(r) + if err != nil { + return bundleResult{}, err + } + if kind != daemon.KindCtrl { + return bundleResult{}, errors.New("remote: expected bundle result frame") + } + var res bundleResult + if err := json.Unmarshal(payload, &res); err != nil { + return bundleResult{}, fmt.Errorf("remote: decode bundle result: %w", err) + } + return res, nil +} + +// ---- server side ----------------------------------------------------------- + +// handleBundle receives an uploaded bundle, extracts it into a per-link working +// tree, and reports the outcome. It always closes conn. +func (b *Bridge) handleBundle(conn net.Conn) { + defer func() { _ = conn.Close() }() + res := b.receiveBundle(conn) + if !res.OK { + b.logf("remote: bundle upload rejected: %s", res.Message) + } + _ = writeBundleResult(conn, res) +} + +// receiveBundle reads the header + framed bundle bytes, verifies the bundle, and +// extracts it under bundleDir. Every failure returns a non-OK result rather than +// panicking, and the staged temp file is always removed. +func (b *Bridge) receiveBundle(conn net.Conn) bundleResult { + hdr, err := readBundleHeader(conn) + if err != nil { + return bundleResult{Message: "read bundle header: " + err.Error()} + } + id, err := sanitizeLinkID(hdr.LinkID) + if err != nil { + return bundleResult{Message: err.Error()} + } + if hdr.Size <= 0 || hdr.Size > b.maxBundleBytes { + return bundleResult{Message: fmt.Sprintf("invalid bundle size %d (max %d)", hdr.Size, b.maxBundleBytes)} + } + + tmp, err := os.CreateTemp("", "zero-remote-*.bundle") + if err != nil { + return bundleResult{Message: "stage bundle: " + err.Error()} + } + tmpName := tmp.Name() + defer func() { _ = os.Remove(tmpName) }() + if err := streamFramesToFile(conn, tmp, hdr.Size); err != nil { + _ = tmp.Close() + return bundleResult{Message: "receive bundle: " + err.Error()} + } + if err := tmp.Close(); err != nil { + return bundleResult{Message: "stage bundle: " + err.Error()} + } + + ctx, cancel := context.WithTimeout(context.Background(), gitTimeout) + defer cancel() + if err := gitBundleVerify(ctx, tmpName); err != nil { + return bundleResult{Message: "bundle verify: " + err.Error()} + } + dest := filepath.Join(b.bundleDir, id) + if !withinDir(b.bundleDir, dest) { + return bundleResult{Message: "invalid link id"} + } + if err := extractBundle(ctx, tmpName, dest); err != nil { + return bundleResult{Message: "extract bundle: " + err.Error()} + } + return bundleResult{OK: true, Path: dest} +} + +// streamFramesToFile copies exactly size bytes from KindData frames on r into w. +// A non-data frame, or any frame that would overrun the declared size, is an +// error (fail closed) so a peer cannot write past the cap. +func streamFramesToFile(r io.Reader, w io.Writer, size int64) error { + remaining := size + for remaining > 0 { + kind, payload, err := daemon.ReadFrame(r) + if err != nil { + return err + } + if kind != daemon.KindData { + return errors.New("expected bundle data frame") + } + if int64(len(payload)) > remaining { + return errors.New("bundle exceeds declared size") + } + if _, err := w.Write(payload); err != nil { + return err + } + remaining -= int64(len(payload)) + } + return nil +} + +// extractBundle clones bundleFile into a staging dir, then atomically renames it +// over dest (replacing any prior extraction for this link id). git clone needs a +// non-existent target, so the staging+rename keeps the live dest intact on error. +func extractBundle(ctx context.Context, bundleFile, dest string) error { + parent := filepath.Dir(dest) + if err := os.MkdirAll(parent, 0o700); err != nil { + return err + } + staging, err := os.MkdirTemp(parent, ".staging-*") + if err != nil { + return err + } + defer func() { _ = os.RemoveAll(staging) }() + cloneDest := filepath.Join(staging, "repo") + if err := gitClone(ctx, bundleFile, cloneDest); err != nil { + return err + } + if err := os.RemoveAll(dest); err != nil { + return err + } + return os.Rename(cloneDest, dest) +} + +// ---- client side ----------------------------------------------------------- + +// UploadRepoBundle creates a git bundle of repoDir's full history and uploads it +// to the remote bridge over an authenticated, bundle-mode TLS connection. The +// bridge extracts it into a per-link working tree and returns its path, captured +// in the returned SessionLink. repoDir must be a git work tree. +func UploadRepoBundle(cfg RemoteConfig, repoDir, linkID string) (*SessionLink, error) { + id, err := sanitizeLinkID(linkID) + if err != nil { + return nil, err + } + repoDir = strings.TrimSpace(repoDir) + if repoDir == "" { + return nil, errors.New("remote: repo dir is required") + } + if !isGitWorktree(repoDir) { + return nil, fmt.Errorf("remote: %s is not a git repository", repoDir) + } + + // Reserve a unique temp name, then let git create the bundle fresh at it. + tmp, err := os.CreateTemp("", "zero-bundle-*.bundle") + if err != nil { + return nil, fmt.Errorf("remote: stage bundle: %w", err) + } + tmpName := tmp.Name() + _ = tmp.Close() + _ = os.Remove(tmpName) + defer func() { _ = os.Remove(tmpName) }() + + ctx, cancel := context.WithTimeout(context.Background(), gitTimeout) + defer cancel() + if err := gitBundleCreate(ctx, repoDir, tmpName); err != nil { + return nil, fmt.Errorf("remote: create bundle: %w", err) + } + sum, size, err := hashFile(tmpName) + if err != nil { + return nil, fmt.Errorf("remote: hash bundle: %w", err) + } + + conn, err := dialAuthenticated(cfg, ModeBundle) + if err != nil { + return nil, err + } + defer func() { _ = conn.Close() }() + + if err := writeBundleHeader(conn, bundleHeader{LinkID: id, Size: size}); err != nil { + return nil, fmt.Errorf("remote: send bundle header: %w", err) + } + if err := streamFileFrames(conn, tmpName); err != nil { + return nil, fmt.Errorf("remote: send bundle: %w", err) + } + res, err := readBundleResult(conn) + if err != nil { + return nil, fmt.Errorf("remote: bundle result: %w", err) + } + if !res.OK { + return nil, fmt.Errorf("remote: bundle rejected: %s", res.Message) + } + + return &SessionLink{ + Address: strings.TrimSpace(cfg.Address), + ServerName: strings.TrimSpace(cfg.ServerName), + CACertFile: strings.TrimSpace(cfg.CACertFile), + LinkID: id, + RemotePath: res.Path, + BundleSHA256: sum, + }, nil +} + +// streamFileFrames writes the file at path to w as a sequence of KindData frames. +func streamFileFrames(w io.Writer, path string) error { + f, err := os.Open(path) + if err != nil { + return err + } + defer func() { _ = f.Close() }() + buf := make([]byte, bundleChunkSize) + for { + n, err := f.Read(buf) + if n > 0 { + if werr := daemon.WriteFrame(w, daemon.KindData, buf[:n]); werr != nil { + return werr + } + } + if err == io.EOF { + return nil + } + if err != nil { + return err + } + } +} + +// ---- git + path helpers ---------------------------------------------------- + +func gitBundleCreate(ctx context.Context, repoDir, outFile string) error { + return runGit(ctx, repoDir, "bundle", "create", outFile, "--all") +} + +func gitBundleVerify(ctx context.Context, bundleFile string) error { + return runGit(ctx, "", "bundle", "verify", bundleFile) +} + +func gitClone(ctx context.Context, bundleFile, destDir string) error { + return runGit(ctx, "", "clone", "--quiet", bundleFile, destDir) +} + +// isGitWorktree reports whether dir is inside a git work tree. +func isGitWorktree(dir string) bool { + cmd := exec.Command("git", "-C", dir, "rev-parse", "--is-inside-work-tree") + out, err := cmd.CombinedOutput() + return err == nil && strings.TrimSpace(string(out)) == "true" +} + +// runGit runs a git subcommand, returning a concise single-line error on failure. +func runGit(ctx context.Context, dir string, args ...string) error { + cmd := exec.CommandContext(ctx, "git", args...) + if dir != "" { + cmd.Dir = dir + } + out, err := cmd.CombinedOutput() + if err != nil { + msg := firstLine(strings.TrimSpace(string(out))) + if msg == "" { + msg = err.Error() + } + return fmt.Errorf("git %s: %s", args[0], msg) + } + return nil +} + +// sanitizeLinkID validates a link id used as a single path component under the +// bundle dir. It allows letters, digits, '-', '_', '.', forbids the traversal +// names, and caps the length — so it can never escape the bundle dir. +func sanitizeLinkID(id string) (string, error) { + id = strings.TrimSpace(id) + if id == "" { + return "", errors.New("remote: link id is required") + } + if len(id) > 128 { + return "", errors.New("remote: link id too long (max 128)") + } + if id == "." || id == ".." { + return "", errors.New("remote: invalid link id") + } + for _, r := range id { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '.': + default: + return "", errors.New("remote: link id may only contain letters, digits, '-', '_', '.'") + } + } + return id, nil +} + +// withinDir reports whether target resolves to a path inside root. +func withinDir(root, target string) bool { + rel, err := filepath.Rel(root, target) + if err != nil { + return false + } + return rel != ".." && !strings.HasPrefix(rel, ".."+string(os.PathSeparator)) +} + +// hashFile returns the hex SHA-256 and byte size of the file at path. +func hashFile(path string) (sum string, size int64, err error) { + f, err := os.Open(path) + if err != nil { + return "", 0, err + } + defer func() { _ = f.Close() }() + h := sha256.New() + n, err := io.Copy(h, f) + if err != nil { + return "", 0, err + } + return hex.EncodeToString(h.Sum(nil)), n, nil +} + +func firstLine(s string) string { + if i := strings.IndexByte(s, '\n'); i >= 0 { + return s[:i] + } + return s +} diff --git a/internal/daemon/remote/bundle_test.go b/internal/daemon/remote/bundle_test.go new file mode 100644 index 000000000..7f711a3bd --- /dev/null +++ b/internal/daemon/remote/bundle_test.go @@ -0,0 +1,193 @@ +package remote + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "runtime" + "testing" +) + +// initTestRepo creates a temp git work tree with one committed file and returns +// its path. It sets a deterministic identity so it does not depend on global git +// config. +func initTestRepo(t *testing.T, file, content string) string { + t.Helper() + dir := t.TempDir() + run := func(args ...string) { + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), + "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@example.test", + "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@example.test") + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("git %v: %v\n%s", args, err, out) + } + } + run("init", "-q") + run("config", "user.email", "t@example.test") + run("config", "user.name", "t") + if err := os.WriteFile(filepath.Join(dir, file), []byte(content), 0o644); err != nil { + t.Fatal(err) + } + run("add", ".") + run("commit", "-q", "-m", "init") + return dir +} + +func TestGitBundleRoundTrip(t *testing.T) { + repo := initTestRepo(t, "a.txt", "content") + ctx := context.Background() + bundle := filepath.Join(t.TempDir(), "x.bundle") + if err := gitBundleCreate(ctx, repo, bundle); err != nil { + t.Fatalf("create: %v", err) + } + if err := gitBundleVerify(ctx, bundle); err != nil { + t.Fatalf("verify: %v", err) + } + dest := filepath.Join(t.TempDir(), "clone") + if err := gitClone(ctx, bundle, dest); err != nil { + t.Fatalf("clone: %v", err) + } + if _, err := os.Stat(filepath.Join(dest, "a.txt")); err != nil { + t.Fatalf("cloned tree missing file: %v", err) + } +} + +func TestIsGitWorktree(t *testing.T) { + repo := initTestRepo(t, "f", "x") + if !isGitWorktree(repo) { + t.Fatal("a git repo should be detected") + } + if isGitWorktree(t.TempDir()) { + t.Fatal("a plain dir should not be detected as a git repo") + } +} + +func TestSanitizeLinkID(t *testing.T) { + for _, ok := range []string{"proj", "proj-1", "a_b.c", "ABC123"} { + if _, err := sanitizeLinkID(ok); err != nil { + t.Fatalf("sanitizeLinkID(%q) unexpected error: %v", ok, err) + } + } + for _, bad := range []string{"", " ", ".", "..", "a/b", "a\\b", "a b", "a$b", string(make([]byte, 200))} { + if _, err := sanitizeLinkID(bad); err == nil { + t.Fatalf("sanitizeLinkID(%q) should error", bad) + } + } +} + +func TestWithinDir(t *testing.T) { + root := t.TempDir() + if !withinDir(root, filepath.Join(root, "child")) { + t.Fatal("child should be within root") + } + if withinDir(root, filepath.Dir(root)) { + t.Fatal("parent should not be within root") + } +} + +func TestSessionLinkSaveLoad(t *testing.T) { + path := filepath.Join(t.TempDir(), "link.json") + link := SessionLink{Address: "host:9000", ServerName: "host", LinkID: "proj-1", RemotePath: "/bundles/proj-1", BundleSHA256: "deadbeef"} + if err := link.Save(path); err != nil { + t.Fatalf("Save: %v", err) + } + if runtime.GOOS != "windows" { + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Fatalf("link file perm = %v, want 0600", perm) + } + } + got, err := LoadSessionLink(path) + if err != nil { + t.Fatalf("Load: %v", err) + } + if *got != link { + t.Fatalf("roundtrip mismatch: %+v != %+v", *got, link) + } +} + +func TestSessionLinkValidate(t *testing.T) { + for _, bad := range []SessionLink{ + {LinkID: "p", RemotePath: "/r"}, // no address + {Address: "h:1", RemotePath: "/r"}, // no link id + {Address: "h:1", LinkID: "p"}, // no remote path + } { + if err := bad.Validate(); err == nil { + t.Fatalf("Validate(%+v) should error", bad) + } + if err := bad.Save(filepath.Join(t.TempDir(), "x.json")); err == nil { + t.Fatalf("Save of invalid link %+v should error", bad) + } + } +} + +func TestUploadRepoBundleRejectsNonRepo(t *testing.T) { + // A non-git dir is rejected before any dial, so the bogus address is never used. + _, err := UploadRepoBundle(RemoteConfig{Address: "127.0.0.1:1", Token: "t"}, t.TempDir(), "p") + if err == nil { + t.Fatal("a non-git dir should be rejected") + } +} + +func TestBridgeBundleUploadRoundTrip(t *testing.T) { + srv := newBridgeServer(t, staticLauncher()) + auth, _ := NewTokenAuthenticator("tok") + bundleRoot := t.TempDir() + addr, ca := startBridge(t, srv, BridgeOptions{Authenticator: auth, BundleDir: bundleRoot}) + + repo := initTestRepo(t, "hello.txt", "hi there") + link, err := UploadRepoBundle(RemoteConfig{Address: addr, Token: "tok", CACertFile: ca}, repo, "proj-1") + if err != nil { + t.Fatalf("UploadRepoBundle: %v", err) + } + wantPath := filepath.Join(bundleRoot, "proj-1") + if link.RemotePath != wantPath { + t.Fatalf("remote path = %q, want %q", link.RemotePath, wantPath) + } + if link.BundleSHA256 == "" { + t.Fatal("link should carry a bundle sha256") + } + // The extracted work tree should contain the committed file. + data, err := os.ReadFile(filepath.Join(link.RemotePath, "hello.txt")) + if err != nil { + t.Fatalf("extracted tree missing file: %v", err) + } + if string(data) != "hi there" { + t.Fatalf("extracted file content = %q", data) + } + + // A second upload to the same link id replaces the prior extraction. + if _, err := UploadRepoBundle(RemoteConfig{Address: addr, Token: "tok", CACertFile: ca}, repo, "proj-1"); err != nil { + t.Fatalf("re-upload: %v", err) + } +} + +func TestBridgeBundleDisabledByDefault(t *testing.T) { + srv := newBridgeServer(t, staticLauncher()) + auth, _ := NewTokenAuthenticator("tok") + addr, ca := startBridge(t, srv, BridgeOptions{Authenticator: auth, AuthFailDelay: -1}) // no BundleDir + + repo := initTestRepo(t, "f", "x") + _, err := UploadRepoBundle(RemoteConfig{Address: addr, Token: "tok", CACertFile: ca}, repo, "p") + if err == nil { + t.Fatal("bundle upload must be refused when --bundle-dir is unset") + } +} + +func TestBridgeBundleRejectsBadToken(t *testing.T) { + srv := newBridgeServer(t, staticLauncher()) + auth, _ := NewTokenAuthenticator("correct") + addr, ca := startBridge(t, srv, BridgeOptions{Authenticator: auth, BundleDir: t.TempDir(), AuthFailDelay: -1}) + + repo := initTestRepo(t, "f", "x") + _, err := UploadRepoBundle(RemoteConfig{Address: addr, Token: "wrong", CACertFile: ca}, repo, "p") + if err == nil { + t.Fatal("bundle upload with a bad token must be refused") + } +} diff --git a/internal/daemon/remote/client.go b/internal/daemon/remote/client.go index aea69c839..66f026b2c 100644 --- a/internal/daemon/remote/client.go +++ b/internal/daemon/remote/client.go @@ -37,6 +37,23 @@ type RemoteConfig struct { // share one protocol. The server certificate is always verified (never // InsecureSkipVerify). func DialRemote(cfg RemoteConfig) (*daemon.Client, error) { + conn, err := dialAuthenticated(cfg, ModeSession) + if err != nil { + return nil, err + } + client, err := daemon.NewClientConn(conn) + if err != nil { + _ = conn.Close() + return nil, err + } + return client, nil +} + +// dialAuthenticated establishes a verified TLS connection, performs the +// bearer-token auth handshake for the given mode, and returns the live conn with +// its deadline cleared (ready for the subsequent daemon handshake or bundle +// stream). The server certificate is always verified (never InsecureSkipVerify). +func dialAuthenticated(cfg RemoteConfig, mode string) (net.Conn, error) { address := strings.TrimSpace(cfg.Address) if address == "" { return nil, errors.New("remote: address is required") @@ -65,7 +82,7 @@ func DialRemote(cfg RemoteConfig) (*daemon.Client, error) { } // Bound the auth handshake; the daemon handshake + stream run without a deadline. _ = conn.SetDeadline(time.Now().Add(timeout)) - if err := writeAuthRequest(conn, authRequest{Token: token, Version: daemon.ProtoVersion}); err != nil { + if err := writeAuthRequest(conn, authRequest{Token: token, Version: daemon.ProtoVersion, Mode: mode}); err != nil { _ = conn.Close() return nil, fmt.Errorf("remote: send auth: %w", err) } @@ -79,7 +96,7 @@ func DialRemote(cfg RemoteConfig) (*daemon.Client, error) { return nil, fmt.Errorf("%w: %s", ErrUnauthorized, resp.Message) } _ = conn.SetDeadline(time.Time{}) - return daemon.NewClientConn(conn) + return conn, nil } // clientTLSConfig builds a verifying client TLS config. It never disables diff --git a/internal/daemon/remote/sessionlink.go b/internal/daemon/remote/sessionlink.go new file mode 100644 index 000000000..7f66fafb0 --- /dev/null +++ b/internal/daemon/remote/sessionlink.go @@ -0,0 +1,107 @@ +package remote + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" +) + +// SessionLink records the association between a local repo and the remote +// working tree a bundle upload produced. It carries everything needed to reach +// the linked repo again — the bridge address, TLS verification details, the link +// id, and the extracted remote path — except the bearer token, which is always +// supplied separately (never persisted to the link file). +type SessionLink struct { + // Address is host:port of the remote bridge. + Address string `json:"address"` + // ServerName overrides TLS/SNI verification; empty => host of Address. + ServerName string `json:"server_name,omitempty"` + // CACertFile is the CA trusted for the bridge cert (for a self-signed bridge). + CACertFile string `json:"ca_cert_file,omitempty"` + // LinkID is the per-link identifier (a single path component on the remote). + LinkID string `json:"link_id"` + // RemotePath is the extracted working tree on the remote; use it as --cwd for + // a remote run/attach against the linked repo. + RemotePath string `json:"remote_path"` + // BundleSHA256 is the hex SHA-256 of the uploaded bundle, for verification. + BundleSHA256 string `json:"bundle_sha256,omitempty"` +} + +// Validate checks the fields required to use a link. +func (l SessionLink) Validate() error { + if strings.TrimSpace(l.Address) == "" { + return errors.New("remote: session link address is required") + } + if strings.TrimSpace(l.LinkID) == "" { + return errors.New("remote: session link id is required") + } + if strings.TrimSpace(l.RemotePath) == "" { + return errors.New("remote: session link remote path is required") + } + return nil +} + +// Save writes the link to path as pretty JSON with 0600 permissions, atomically +// (write-temp-then-rename) so a reader never sees a partial file. +func (l SessionLink) Save(path string) error { + if err := l.Validate(); err != nil { + return err + } + data, err := json.MarshalIndent(l, "", " ") + if err != nil { + return err + } + return atomicWriteFile(path, append(data, '\n'), 0o600) +} + +// LoadSessionLink reads and validates a link file written by Save. +func LoadSessionLink(path string) (*SessionLink, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var l SessionLink + if err := json.Unmarshal(data, &l); err != nil { + return nil, fmt.Errorf("remote: decode session link: %w", err) + } + if err := l.Validate(); err != nil { + return nil, err + } + return &l, nil +} + +// atomicWriteFile writes data to path via a temp file in the same directory, +// chmod-ed to perm, then renamed into place. +func atomicWriteFile(path string, data []byte, perm os.FileMode) error { + dir := filepath.Dir(path) + tmp, err := os.CreateTemp(dir, ".link-*") + if err != nil { + return err + } + tmpName := tmp.Name() + cleanup := true + defer func() { + if cleanup { + _ = os.Remove(tmpName) + } + }() + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Chmod(perm); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + if err := os.Rename(tmpName, path); err != nil { + return err + } + cleanup = false + return nil +} diff --git a/internal/keyring/keyring.go b/internal/keyring/keyring.go new file mode 100644 index 000000000..7f116b50f --- /dev/null +++ b/internal/keyring/keyring.go @@ -0,0 +1,225 @@ +// Package keyring provides a small, dependency-free secret store backed by the +// operating system's native credential tooling: the `security` keychain CLI on +// macOS and `secret-tool` (libsecret) on Linux. It stores a single secret string +// per (service, account). Windows and other platforms report unsupported. +// +// It shells out to the OS tools rather than taking a third-party dependency. On +// macOS the secret is passed to `security` as an argument, so it is briefly +// visible to other processes via the process list; on Linux the secret is passed +// over stdin and is not exposed in the argument vector. Callers that need to keep +// a secret out of the process list on macOS should prefer the file backend. +package keyring + +import ( + "bytes" + "context" + "errors" + "fmt" + "os/exec" + "runtime" + "strings" + "time" +) + +// commandTimeout bounds a single keyring tool invocation. +const commandTimeout = 10 * time.Second + +// ErrUnsupported is returned when no OS keyring backend is available for the +// current platform. +var ErrUnsupported = errors.New("keyring: no OS keyring backend on this platform") + +// runner executes name with args and optional stdin, returning stdout. It is the +// single seam tests replace to drive the platform command logic without touching +// a real keychain. +type runner func(ctx context.Context, name string, stdin []byte, args ...string) ([]byte, error) + +// Keyring is an OS-native secret store. +type Keyring struct { + run runner + goos string +} + +// New returns a Keyring for the current platform. +func New() *Keyring { + return &Keyring{run: execRunner, goos: runtime.GOOS} +} + +// Available reports whether this platform has a supported keyring backend. The +// backing tool (`secret-tool` on Linux) must also be installed; a missing tool +// surfaces as an error from Get/Set/Delete. +func (k *Keyring) Available() bool { + switch k.goos { + case "darwin", "linux": + return true + default: + return false + } +} + +// Set stores secret under (service, account), replacing any existing value. +func (k *Keyring) Set(service, account, secret string) error { + if err := validate(service, account); err != nil { + return err + } + switch k.goos { + case "darwin": + // -U updates the item if it already exists rather than failing. + _, err := k.exec(nil, "security", "add-generic-password", "-U", "-s", service, "-a", account, "-w", secret) + return wrap("set", err) + case "linux": + // secret-tool reads the secret from stdin, keeping it out of the argv. + _, err := k.exec([]byte(secret), "secret-tool", "store", "--label", "zero", "service", service, "account", account) + return wrap("set", err) + default: + return ErrUnsupported + } +} + +// Get returns the secret stored under (service, account). The bool is false when +// no entry exists (which is not an error). +func (k *Keyring) Get(service, account string) (string, bool, error) { + if err := validate(service, account); err != nil { + return "", false, err + } + switch k.goos { + case "darwin": + out, err := k.exec(nil, "security", "find-generic-password", "-s", service, "-a", account, "-w") + if err != nil { + if isNotFound(err, securityNotFoundExit) { + return "", false, nil + } + return "", false, wrap("get", err) + } + return strings.TrimRight(string(out), "\r\n"), true, nil + case "linux": + out, err := k.exec(nil, "secret-tool", "lookup", "service", service, "account", account) + if err != nil { + if isNotFound(err, secretToolNotFoundExit) { + return "", false, nil + } + return "", false, wrap("get", err) + } + value := strings.TrimRight(string(out), "\r\n") + if value == "" { + return "", false, nil + } + return value, true, nil + default: + return "", false, ErrUnsupported + } +} + +// Delete removes the entry under (service, account), reporting whether one +// existed. +func (k *Keyring) Delete(service, account string) (bool, error) { + if err := validate(service, account); err != nil { + return false, err + } + switch k.goos { + case "darwin": + _, err := k.exec(nil, "security", "delete-generic-password", "-s", service, "-a", account) + if err != nil { + if isNotFound(err, securityNotFoundExit) { + return false, nil + } + return false, wrap("delete", err) + } + return true, nil + case "linux": + // `secret-tool clear` always exits 0, so probe existence first. + _, existed, err := k.Get(service, account) + if err != nil { + return false, err + } + if _, err := k.exec(nil, "secret-tool", "clear", "service", service, "account", account); err != nil { + return false, wrap("delete", err) + } + return existed, nil + default: + return false, ErrUnsupported + } +} + +func (k *Keyring) exec(stdin []byte, name string, args ...string) ([]byte, error) { + ctx, cancel := context.WithTimeout(context.Background(), commandTimeout) + defer cancel() + return k.run(ctx, name, stdin, args...) +} + +func execRunner(ctx context.Context, name string, stdin []byte, args ...string) ([]byte, error) { + cmd := exec.CommandContext(ctx, name, args...) + if len(stdin) > 0 { + cmd.Stdin = bytes.NewReader(stdin) + } + var out, errBuf bytes.Buffer + cmd.Stdout = &out + cmd.Stderr = &errBuf + if err := cmd.Run(); err != nil { + return out.Bytes(), &runError{err: err, stderr: strings.TrimSpace(errBuf.String())} + } + return out.Bytes(), nil +} + +// runError carries a tool failure with its stderr and preserves the underlying +// error for errors.As (so exit-code / not-found detection still works). +type runError struct { + err error + stderr string +} + +func (e *runError) Error() string { + if e.stderr != "" { + return e.stderr + } + return e.err.Error() +} + +func (e *runError) Unwrap() error { return e.err } + +// Not-found exit codes for the OS tools: macOS `security` exits 44 +// (errSecItemNotFound) when no matching item exists; `secret-tool` exits 1 when a +// lookup finds nothing. Any other non-zero exit is a real failure. +const ( + securityNotFoundExit = 44 + secretToolNotFoundExit = 1 +) + +// isNotFound reports whether err is a tool exit whose code is one of the given +// "no such entry" codes (as opposed to a missing binary or a genuine failure, +// which must not be masked). It matches on the ExitCode behavior (satisfied by +// *exec.ExitError) so the logic is testable without spawning a real process. +func isNotFound(err error, codes ...int) bool { + var coder interface{ ExitCode() int } + if !errors.As(err, &coder) { + return false + } + code := coder.ExitCode() + for _, c := range codes { + if code == c { + return true + } + } + return false +} + +// wrap adds operation context to a tool error, leaving nil untouched. +func wrap(op string, err error) error { + if err == nil { + return nil + } + var execErr *exec.Error + if errors.As(err, &execErr) { + return fmt.Errorf("keyring: %s: %q not found (install the OS keyring tool): %w", op, execErr.Name, err) + } + return fmt.Errorf("keyring: %s: %w", op, err) +} + +func validate(service, account string) error { + if strings.TrimSpace(service) == "" { + return errors.New("keyring: service is required") + } + if strings.TrimSpace(account) == "" { + return errors.New("keyring: account is required") + } + return nil +} diff --git a/internal/keyring/keyring_test.go b/internal/keyring/keyring_test.go new file mode 100644 index 000000000..01a09ebfc --- /dev/null +++ b/internal/keyring/keyring_test.go @@ -0,0 +1,216 @@ +package keyring + +import ( + "context" + "os/exec" + "strings" + "testing" +) + +// fakeExit is a not-found exit error: it satisfies the ExitCode() seam used by +// isNotFound without spawning a real process. +type fakeExit struct{ code int } + +func (e fakeExit) Error() string { return "exit status" } +func (e fakeExit) ExitCode() int { return e.code } + +// fakeKeyring is an in-memory simulation of the OS tools driven through the +// runner seam. It records the last stdin so tests can assert the secret never +// travels via argv on Linux. +type fakeKeyring struct { + goos string + data map[string]string + lastStdin string + lastArgs []string +} + +func newFake(goos string) *fakeKeyring { + return &fakeKeyring{goos: goos, data: map[string]string{}} +} + +func (f *fakeKeyring) keyring() *Keyring { return &Keyring{run: f.run, goos: f.goos} } + +func flagValue(args []string, flag string) string { + for i := 0; i < len(args)-1; i++ { + if args[i] == flag { + return args[i+1] + } + } + return "" +} + +func attrValue(args []string, attr string) string { + for i := 0; i < len(args)-1; i++ { + if args[i] == attr { + return args[i+1] + } + } + return "" +} + +func key(service, account string) string { return service + "\x00" + account } + +func (f *fakeKeyring) run(_ context.Context, name string, stdin []byte, args ...string) ([]byte, error) { + f.lastStdin = string(stdin) + f.lastArgs = append([]string{name}, args...) + if len(args) == 0 { + return nil, fakeExit{1} + } + switch f.goos { + case "darwin": + svc, acct := flagValue(args, "-s"), flagValue(args, "-a") + switch args[0] { + case "add-generic-password": + f.data[key(svc, acct)] = flagValue(args, "-w") + return nil, nil + case "find-generic-password": + if v, ok := f.data[key(svc, acct)]; ok { + return []byte(v + "\n"), nil // security prints a trailing newline + } + return nil, fakeExit{44} + case "delete-generic-password": + if _, ok := f.data[key(svc, acct)]; ok { + delete(f.data, key(svc, acct)) + return nil, nil + } + return nil, fakeExit{44} + } + case "linux": + svc, acct := attrValue(args, "service"), attrValue(args, "account") + switch args[0] { + case "store": + f.data[key(svc, acct)] = string(stdin) + return nil, nil + case "lookup": + if v, ok := f.data[key(svc, acct)]; ok { + return []byte(v), nil // secret-tool prints no trailing newline + } + return nil, fakeExit{1} + case "clear": + delete(f.data, key(svc, acct)) + return nil, nil + } + } + return nil, fakeExit{1} +} + +func TestKeyringGetSurfacesNonNotFoundError(t *testing.T) { + // On macOS only exit 44 (errSecItemNotFound) means "no entry"; any other + // non-zero exit is a real failure that must surface, not be masked as absent. + k := &Keyring{ + goos: "darwin", + run: func(_ context.Context, _ string, _ []byte, _ ...string) ([]byte, error) { + return nil, fakeExit{1} + }, + } + if _, ok, err := k.Get("zero", "tokens"); err == nil || ok { + t.Fatalf("a non-44 exit must surface as an error, got ok=%v err=%v", ok, err) + } +} + +func TestKeyringRoundTripDarwin(t *testing.T) { + k := newFake("darwin").keyring() + if err := k.Set("zero", "tokens", "blob-AAA"); err != nil { + t.Fatalf("Set: %v", err) + } + got, ok, err := k.Get("zero", "tokens") + if err != nil || !ok { + t.Fatalf("Get: ok=%v err=%v", ok, err) + } + if got != "blob-AAA" { + t.Fatalf("Get = %q, want blob-AAA", got) + } + existed, err := k.Delete("zero", "tokens") + if err != nil || !existed { + t.Fatalf("Delete: existed=%v err=%v", existed, err) + } + if _, ok, _ := k.Get("zero", "tokens"); ok { + t.Fatal("token should be gone after delete") + } +} + +func TestKeyringRoundTripLinuxUsesStdin(t *testing.T) { + f := newFake("linux") + k := f.keyring() + if err := k.Set("zero", "tokens", "blob-BBB"); err != nil { + t.Fatalf("Set: %v", err) + } + // The secret must travel via stdin, never the argument vector. + if f.lastStdin != "blob-BBB" { + t.Fatalf("secret not sent via stdin: stdin=%q", f.lastStdin) + } + for _, a := range f.lastArgs { + if strings.Contains(a, "blob-BBB") { + t.Fatalf("secret leaked into argv: %v", f.lastArgs) + } + } + got, ok, err := k.Get("zero", "tokens") + if err != nil || !ok || got != "blob-BBB" { + t.Fatalf("Get = %q ok=%v err=%v", got, ok, err) + } + existed, err := k.Delete("zero", "tokens") + if err != nil || !existed { + t.Fatalf("Delete: existed=%v err=%v", existed, err) + } +} + +func TestKeyringGetMissingIsNotError(t *testing.T) { + for _, goos := range []string{"darwin", "linux"} { + k := newFake(goos).keyring() + if _, ok, err := k.Get("zero", "absent"); err != nil || ok { + t.Fatalf("[%s] Get(absent) = ok=%v err=%v, want false/nil", goos, ok, err) + } + if existed, err := k.Delete("zero", "absent"); err != nil || existed { + t.Fatalf("[%s] Delete(absent) = existed=%v err=%v, want false/nil", goos, existed, err) + } + } +} + +func TestKeyringUnsupportedPlatform(t *testing.T) { + k := &Keyring{run: newFake("windows").run, goos: "windows"} + if k.Available() { + t.Fatal("windows should report unavailable") + } + if err := k.Set("zero", "tokens", "x"); err == nil { + t.Fatal("Set on unsupported platform should error") + } + if _, _, err := k.Get("zero", "tokens"); err == nil { + t.Fatal("Get on unsupported platform should error") + } + if _, err := k.Delete("zero", "tokens"); err == nil { + t.Fatal("Delete on unsupported platform should error") + } +} + +func TestKeyringValidation(t *testing.T) { + k := newFake("darwin").keyring() + if err := k.Set("", "a", "s"); err == nil { + t.Fatal("empty service should error") + } + if err := k.Set("svc", "", "s"); err == nil { + t.Fatal("empty account should error") + } +} + +func TestKeyringMissingBinaryError(t *testing.T) { + // A missing tool surfaces as a wrapped, descriptive error (not not-found). + k := &Keyring{goos: "linux", run: func(context.Context, string, []byte, ...string) ([]byte, error) { + return nil, &exec.Error{Name: "secret-tool", Err: exec.ErrNotFound} + }} + if err := k.Set("zero", "tokens", "x"); err == nil || !strings.Contains(err.Error(), "secret-tool") { + t.Fatalf("missing-binary Set error = %v, want mention of secret-tool", err) + } + // A missing binary on Get must not be misread as not-found. + if _, ok, err := k.Get("zero", "tokens"); err == nil || ok { + t.Fatalf("missing-binary Get = ok=%v err=%v, want error", ok, err) + } +} + +func TestAvailable(t *testing.T) { + if !(newFake("darwin").keyring().Available()) { + t.Fatal("darwin should be available") + } + if !(newFake("linux").keyring().Available()) { + t.Fatal("linux should be available") + } +} diff --git a/internal/mcp/oauth_store.go b/internal/mcp/oauth_store.go index ad7c25644..78a0f822c 100644 --- a/internal/mcp/oauth_store.go +++ b/internal/mcp/oauth_store.go @@ -6,18 +6,16 @@ import ( "fmt" "os" "path/filepath" - "sort" "strings" - "sync" "time" -) -const ( - tokenStoreSchemaVersion = 1 - tokenStoreLockTimeout = 5 * time.Second - tokenStoreLockRetry = 10 * time.Millisecond + "github.com/Gitlawb/zero/internal/oauth" ) +// tokenStoreSchemaVersion is the schema of the legacy mcp-oauth-tokens.json file, +// retained so migration can recognize a file it understands. +const tokenStoreSchemaVersion = 1 + // StoredToken holds the credentials issued by an OAuth 2.0 authorization server // for a single MCP server. The token fields are sensitive: they are tagged so // the repo's redaction layer masks them, and they must never be written to logs @@ -42,29 +40,37 @@ type TokenStatus struct { Expired bool `json:"expired"` } -// TokenStoreOptions configures where OAuth tokens are persisted. When FilePath -// is empty the path is resolved under the user config dir, mirroring how MCP -// permissions are stored. +// TokenStoreOptions configures the unified token store backing MCP OAuth tokens. +// FilePath overrides the store path (default: the shared oauth store path). +// LegacyPath overrides the pre-unification file migrated on construction; when +// empty it defaults to the conventional mcp-oauth-tokens.json only for the +// default (FilePath-unset) store, so an explicit FilePath never triggers an +// unexpected migration from the real user config. type TokenStoreOptions struct { - FilePath string - Env map[string]string - Now func() time.Time + FilePath string + LegacyPath string + Env map[string]string + Now func() time.Time } -// TokenStore persists OAuth tokens per MCP server in a 0600 JSON file. +// TokenStore persists MCP OAuth tokens in the unified oauth store +// (internal/oauth) under the "mcp:" namespace, sharing one file with provider +// logins. On construction it transparently and non-destructively migrates a +// legacy mcp-oauth-tokens.json into the unified store. type TokenStore struct { - filePath string - now func() time.Time - mu sync.Mutex + store *oauth.Store } +// tokenFile is the legacy on-disk format, retained only to read a +// pre-unification mcp-oauth-tokens.json during migration. type tokenFile struct { SchemaVersion int `json:"schemaVersion"` Tokens map[string]StoredToken `json:"tokens"` } -// ResolveTokenStorePath determines the on-disk location for OAuth tokens, -// honoring an explicit override, XDG_CONFIG_HOME, then the user home dir. +// ResolveTokenStorePath determines the on-disk location of the LEGACY OAuth token +// file, honoring an explicit override, XDG_CONFIG_HOME, then the user home dir. +// It is used to locate a pre-unification file for migration. func ResolveTokenStorePath(env map[string]string) (string, error) { override := strings.TrimSpace(envValue(env, "ZERO_MCP_OAUTH_TOKENS_PATH")) if override != "" { @@ -95,215 +101,168 @@ func ResolveTokenStorePath(env map[string]string) (string, error) { return filepath.Join(configHome, "zero", "mcp-oauth-tokens.json"), nil } -// NewTokenStore builds a file-backed token store. +// NewTokenStore builds the unified-store-backed token store and runs a one-time +// migration from a legacy file when applicable. func NewTokenStore(options TokenStoreOptions) (*TokenStore, error) { - filePath := options.FilePath - var err error - if strings.TrimSpace(filePath) == "" { - filePath, err = ResolveTokenStorePath(options.Env) + now := options.Now + if now == nil { + now = time.Now + } + unified, err := oauth.NewStore(oauth.StoreOptions{ + FilePath: options.FilePath, + Env: options.Env, + Now: now, + }) + if err != nil { + return nil, err + } + store := &TokenStore{store: unified} + + legacyPath := strings.TrimSpace(options.LegacyPath) + if legacyPath == "" && strings.TrimSpace(options.FilePath) == "" { + // Default (production) construction: migrate from the conventional path. + legacyPath, err = ResolveTokenStorePath(options.Env) if err != nil { return nil, err } } - if !filepath.IsAbs(filePath) { - filePath, err = filepath.Abs(filePath) - if err != nil { + if legacyPath != "" { + if err := store.migrateLegacy(legacyPath); err != nil { return nil, err } } - now := options.Now - if now == nil { - now = time.Now - } - return &TokenStore{filePath: filepath.Clean(filePath), now: now}, nil + return store, nil } -// FilePath returns the resolved token store path. +// FilePath returns the resolved unified store path. func (store *TokenStore) FilePath() string { - return store.filePath + return store.store.FilePath() } // Save persists the token for a server, replacing any existing entry. func (store *TokenStore) Save(serverName string, token StoredToken) error { - if err := ValidateServerName(serverName); err != nil { - return err - } - store.mu.Lock() - defer store.mu.Unlock() - unlock, err := store.lockStateFile() + key, err := mcpKey(serverName) if err != nil { return err } - defer unlock() - - state, err := store.readState() - if err != nil { - return err - } - state.Tokens[serverName] = token - return store.writeState(state) + return store.store.Save(key, storedToOAuth(token)) } // Load returns the stored token for a server. The second return value is false // when no token has been stored for the server. func (store *TokenStore) Load(serverName string) (StoredToken, bool, error) { - if err := ValidateServerName(serverName); err != nil { - return StoredToken{}, false, err - } - store.mu.Lock() - defer store.mu.Unlock() - - state, err := store.readState() + key, err := mcpKey(serverName) if err != nil { return StoredToken{}, false, err } - token, ok := state.Tokens[serverName] - return token, ok, nil + token, ok, err := store.store.Load(key) + if err != nil || !ok { + return StoredToken{}, ok, err + } + return tokenToStored(token), true, nil } // Delete removes the stored token for a server. It reports whether an entry was // present before deletion. func (store *TokenStore) Delete(serverName string) (bool, error) { - if err := ValidateServerName(serverName); err != nil { - return false, err - } - store.mu.Lock() - defer store.mu.Unlock() - unlock, err := store.lockStateFile() + key, err := mcpKey(serverName) if err != nil { return false, err } - defer unlock() - - state, err := store.readState() - if err != nil { - return false, err - } - if _, ok := state.Tokens[serverName]; !ok { - return false, nil - } - delete(state.Tokens, serverName) - if err := store.writeState(state); err != nil { - return false, err - } - return true, nil + return store.store.Delete(key) } -// Status returns a redaction-safe summary of every stored token, sorted by +// Status returns a redaction-safe summary of every stored MCP token, sorted by // server name. It never includes the token material. func (store *TokenStore) Status() ([]TokenStatus, error) { - store.mu.Lock() - defer store.mu.Unlock() - - state, err := store.readState() + statuses, err := store.store.Status(oauth.KeyPrefixMCP) if err != nil { return nil, err } - names := make([]string, 0, len(state.Tokens)) - for name := range state.Tokens { - names = append(names, name) - } - sort.Strings(names) - - now := store.now() - statuses := make([]TokenStatus, 0, len(names)) - for _, name := range names { - token := state.Tokens[name] - status := TokenStatus{ - ServerName: name, - HasToken: strings.TrimSpace(token.AccessToken) != "", - HasRefreshToken: strings.TrimSpace(token.RefreshToken) != "", - TokenType: token.TokenType, - Scopes: token.Scopes, - ExpiresAt: token.ExpiresAt, - } - if !token.ExpiresAt.IsZero() && !token.ExpiresAt.After(now) { - status.Expired = true - } - statuses = append(statuses, status) - } - return statuses, nil + out := make([]TokenStatus, 0, len(statuses)) + for _, s := range statuses { + out = append(out, TokenStatus{ + ServerName: strings.TrimPrefix(s.Key, oauth.KeyPrefixMCP), + HasToken: s.HasToken, + HasRefreshToken: s.HasRefreshToken, + TokenType: s.TokenType, + Scopes: s.Scopes, + ExpiresAt: s.ExpiresAt, + Expired: s.Expired, + }) + } + return out, nil } -func (store *TokenStore) readState() (tokenFile, error) { - data, err := os.ReadFile(store.filePath) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - return emptyTokenFile(), nil - } - return tokenFile{}, err - } - var state tokenFile - if err := json.Unmarshal(data, &state); err != nil { - return tokenFile{}, fmt.Errorf("invalid MCP OAuth token file at %s: %w", store.filePath, err) - } - if state.SchemaVersion != tokenStoreSchemaVersion { - return tokenFile{}, fmt.Errorf("invalid MCP OAuth token file at %s: unsupported schemaVersion", store.filePath) - } - if state.Tokens == nil { - state.Tokens = map[string]StoredToken{} - } - for serverName := range state.Tokens { - if err := ValidateServerName(serverName); err != nil { - return tokenFile{}, fmt.Errorf("invalid MCP OAuth token file at %s: %w", store.filePath, err) +// migrateLegacy imports tokens from a legacy mcp-oauth-tokens.json into the +// unified store (under "mcp:" keys), then renames the legacy file to a +// ".migrated" backup. It is non-destructive and idempotent: a newer unified +// entry is never overwritten, a missing/unreadable/foreign-schema legacy file is +// left untouched, and the rename ensures it is imported at most once. +func (store *TokenStore) migrateLegacy(legacyPath string) error { + legacyPath = filepath.Clean(legacyPath) + // FilePath() is an absolute path; resolve a relative LegacyPath so the + // same-file guard can't be bypassed by a relative spelling of the same file. + if !filepath.IsAbs(legacyPath) { + abs, err := filepath.Abs(legacyPath) + if err != nil { + return err } + legacyPath = filepath.Clean(abs) } - return state, nil -} - -func (store *TokenStore) writeState(state tokenFile) error { - if err := os.MkdirAll(filepath.Dir(store.filePath), 0o700); err != nil { - return err + if legacyPath == store.store.FilePath() { + return nil // legacy and unified resolve to the same file; nothing to migrate } - data, err := json.MarshalIndent(state, "", " ") + data, err := os.ReadFile(legacyPath) if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } return err } - tempPath := fmt.Sprintf("%s.tmp-%d-%d", store.filePath, os.Getpid(), store.now().UnixNano()) - if err := os.WriteFile(tempPath, append(data, '\n'), 0o600); err != nil { - return err + var legacy tokenFile + if err := json.Unmarshal(data, &legacy); err != nil { + return nil // unreadable legacy file: leave it in place, don't block startup } - if err := os.Rename(tempPath, store.filePath); err != nil { - _ = os.Remove(tempPath) - return err + if legacy.SchemaVersion != tokenStoreSchemaVersion { + return nil // unknown legacy schema: leave it untouched } - return nil -} - -func (store *TokenStore) lockStateFile() (func(), error) { - lockPath := store.filePath + ".lockfile" - if err := os.MkdirAll(filepath.Dir(store.filePath), 0o700); err != nil { - return nil, err - } - file, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0o600) - if err != nil { - return nil, err - } - deadline := time.Now().Add(tokenStoreLockTimeout) - for { - locked, err := tryLockPermissionFile(file) + for serverName, token := range legacy.Tokens { + key, err := mcpKey(serverName) if err != nil { - _ = file.Close() - return nil, err + continue // a name that cannot form a valid unified key is skipped } - if locked { - return func() { - _ = unlockPermissionFile(file) - _ = file.Close() - }, nil + if _, ok, loadErr := store.store.Load(key); loadErr == nil && ok { + continue // a unified entry already exists; never overwrite } - if time.Now().After(deadline) { - _ = file.Close() - return nil, fmt.Errorf("timed out waiting for MCP OAuth token lock at %s", lockPath) + if err := store.store.Save(key, storedToOAuth(token)); err != nil { + return err } - time.Sleep(tokenStoreLockRetry) } + return os.Rename(legacyPath, legacyPath+".migrated") +} + +// mcpKey builds and validates the unified store key for an MCP server token. +func mcpKey(serverName string) (string, error) { + if err := ValidateServerName(serverName); err != nil { + return "", err + } + key := oauth.KeyPrefixMCP + strings.TrimSpace(serverName) + if err := oauth.ValidateKey(key); err != nil { + return "", err + } + return key, nil } -func emptyTokenFile() tokenFile { - return tokenFile{ - SchemaVersion: tokenStoreSchemaVersion, - Tokens: map[string]StoredToken{}, +// storedToOAuth converts an MCP StoredToken to the shared oauth.Token (the +// inverse of tokenToStored). MCP never sets the oauth Account field. +func storedToOAuth(s StoredToken) oauth.Token { + return oauth.Token{ + AccessToken: s.AccessToken, + RefreshToken: s.RefreshToken, + TokenType: s.TokenType, + Scopes: s.Scopes, + ExpiresAt: s.ExpiresAt, } } diff --git a/internal/mcp/oauth_store_test.go b/internal/mcp/oauth_store_test.go index 8751d98f3..1c5b0fc8b 100644 --- a/internal/mcp/oauth_store_test.go +++ b/internal/mcp/oauth_store_test.go @@ -158,6 +158,101 @@ func TestTokenStoreStatusReportsPresenceWithoutToken(t *testing.T) { } } +func TestTokenStoreMigratesLegacyFile(t *testing.T) { + dir := t.TempDir() + legacy := filepath.Join(dir, "mcp-oauth-tokens.json") + unified := filepath.Join(dir, "oauth-tokens.json") + legacyData := `{"schemaVersion":1,"tokens":{"demo":{"access_token":"a","refresh_token":"r","token_type":"Bearer"}}}` + if err := os.WriteFile(legacy, []byte(legacyData), 0o600); err != nil { + t.Fatal(err) + } + + store, err := NewTokenStore(TokenStoreOptions{FilePath: unified, LegacyPath: legacy}) + if err != nil { + t.Fatalf("NewTokenStore: %v", err) + } + + tok, ok, err := store.Load("demo") + if err != nil || !ok { + t.Fatalf("migrated token not loadable: ok=%v err=%v", ok, err) + } + if tok.AccessToken != "a" || tok.RefreshToken != "r" { + t.Fatalf("migrated token = %#v", tok) + } + + // The unified file keys the token under the mcp: namespace. + raw, err := os.ReadFile(unified) + if err != nil { + t.Fatalf("read unified: %v", err) + } + if !contains(string(raw), "mcp:demo") { + t.Fatalf("unified file should key under mcp: namespace:\n%s", raw) + } + + // The legacy file is renamed to a .migrated backup (non-destructive, one-time). + if _, err := os.Stat(legacy); !os.IsNotExist(err) { + t.Fatalf("legacy file should be renamed away; stat err = %v", err) + } + if _, err := os.Stat(legacy + ".migrated"); err != nil { + t.Fatalf("legacy backup missing: %v", err) + } + + // Idempotent: a second construction (legacy now absent) keeps the token. + store2, err := NewTokenStore(TokenStoreOptions{FilePath: unified, LegacyPath: legacy}) + if err != nil { + t.Fatalf("NewTokenStore#2: %v", err) + } + if _, ok, _ := store2.Load("demo"); !ok { + t.Fatal("token lost after second construction") + } +} + +func TestTokenStoreMigrationPreservesNewerUnified(t *testing.T) { + dir := t.TempDir() + legacy := filepath.Join(dir, "mcp-oauth-tokens.json") + unified := filepath.Join(dir, "oauth-tokens.json") + if err := os.WriteFile(legacy, []byte(`{"schemaVersion":1,"tokens":{"demo":{"access_token":"OLD"}}}`), 0o600); err != nil { + t.Fatal(err) + } + // Pre-seed the unified store with a newer token (no migration: FilePath set, no LegacyPath). + pre, err := NewTokenStore(TokenStoreOptions{FilePath: unified}) + if err != nil { + t.Fatal(err) + } + if err := pre.Save("demo", StoredToken{AccessToken: "NEW"}); err != nil { + t.Fatal(err) + } + // Migrating must not overwrite the newer unified entry. + store, err := NewTokenStore(TokenStoreOptions{FilePath: unified, LegacyPath: legacy}) + if err != nil { + t.Fatal(err) + } + tok, _, _ := store.Load("demo") + if tok.AccessToken != "NEW" { + t.Fatalf("migration overwrote a newer token: %q", tok.AccessToken) + } +} + +func TestTokenStoreNamespacedFromProvider(t *testing.T) { + // An MCP token and a provider login of the same name coexist in one file. + dir := t.TempDir() + unified := filepath.Join(dir, "oauth-tokens.json") + mcpStore, err := NewTokenStore(TokenStoreOptions{FilePath: unified}) + if err != nil { + t.Fatal(err) + } + if err := mcpStore.Save("shared", StoredToken{AccessToken: "mcp-token"}); err != nil { + t.Fatal(err) + } + statuses, err := mcpStore.Status() + if err != nil { + t.Fatal(err) + } + if len(statuses) != 1 || statuses[0].ServerName != "shared" { + t.Fatalf("status = %#v, want one entry for 'shared'", statuses) + } +} + func TestResolveTokenStorePathUsesXDG(t *testing.T) { // Use a real temp dir so the base is absolute on every OS (a literal // "/tmp/..." isn't absolute on Windows, where ResolveTokenStorePath would diff --git a/internal/notify/webhook_wire.go b/internal/notify/webhook_wire.go new file mode 100644 index 000000000..910b0fdb2 --- /dev/null +++ b/internal/notify/webhook_wire.go @@ -0,0 +1,40 @@ +package notify + +import "strings" + +// Webhook delivery is configured entirely from the environment. A webhook URL +// typically embeds a secret token, so sourcing it from the environment keeps it +// out of any on-disk config file. The sink is strictly opt-in: with +// EnvWebhookURL unset the wiring helper attaches nothing and the notifier +// behaves exactly as before. +const ( + // EnvWebhookURL holds the destination webhook/Slack URL. Empty disables it. + EnvWebhookURL = "ZERO_NOTIFY_WEBHOOK_URL" + // EnvWebhookSummary is an optional one-line run summary attached to every + // payload (for example "nightly audit run"). + EnvWebhookSummary = "ZERO_NOTIFY_WEBHOOK_SUMMARY" +) + +// MaybeAddWebhookSink attaches a webhook sink to n when a webhook URL is present +// in the environment, and is otherwise a no-op. env resolves an environment +// variable (pass os.Getenv); logf records one redacted line per failed delivery +// (pass nil to stay silent — for example a TUI that owns the screen). It is safe +// to call unconditionally: configuration alone decides whether the sink exists. +// +// The attached sink is still subject to the notifier's Mode/focus policy, so a +// webhook only delivers when notifications are enabled (for example +// `--notify both`), matching the rest of the notification surface. +func MaybeAddWebhookSink(n *Notifier, env func(string) string, logf func(format string, args ...any)) { + if n == nil || env == nil { + return + } + url := strings.TrimSpace(env(EnvWebhookURL)) + if url == "" { + return + } + n.AddSink(NewWebhookSink(WebhookConfig{ + URL: url, + Summary: strings.TrimSpace(env(EnvWebhookSummary)), + Logf: logf, + })) +} diff --git a/internal/notify/webhook_wire_test.go b/internal/notify/webhook_wire_test.go new file mode 100644 index 000000000..3263f1264 --- /dev/null +++ b/internal/notify/webhook_wire_test.go @@ -0,0 +1,71 @@ +package notify + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" +) + +// envFunc builds a deterministic env resolver from a map for the wiring tests. +func envFunc(values map[string]string) func(string) string { + return func(key string) string { return values[key] } +} + +func TestMaybeAddWebhookSinkAttachesAndDelivers(t *testing.T) { + var hits int32 + var gotSummary string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&hits, 1) + var payload webhookPayload + _ = json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&payload) + gotSummary = payload.Summary + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + n := New(&bytes.Buffer{}, Config{Mode: ModeBell, FocusMode: FocusAlways}) + MaybeAddWebhookSink(n, envFunc(map[string]string{ + EnvWebhookURL: server.URL, + EnvWebhookSummary: "nightly audit", + }), nil) + + if got := len(n.sinks); got != 1 { + t.Fatalf("expected 1 sink attached, got %d", got) + } + + n.Notify(Completion, "Zero: ready") + if got := atomic.LoadInt32(&hits); got != 1 { + t.Fatalf("webhook hit %d times, want 1", got) + } + if gotSummary != "nightly audit" { + t.Fatalf("summary = %q, want %q", gotSummary, "nightly audit") + } +} + +func TestMaybeAddWebhookSinkNoopWhenURLBlank(t *testing.T) { + n := New(&bytes.Buffer{}, Config{Mode: ModeBell, FocusMode: FocusAlways}) + + // Unset. + MaybeAddWebhookSink(n, envFunc(nil), nil) + // Set but blank / whitespace only. + MaybeAddWebhookSink(n, envFunc(map[string]string{EnvWebhookURL: " "}), nil) + + if got := len(n.sinks); got != 0 { + t.Fatalf("expected no sink attached, got %d", got) + } +} + +func TestMaybeAddWebhookSinkNilGuards(t *testing.T) { + // Must not panic on a nil notifier or nil env resolver. + MaybeAddWebhookSink(nil, envFunc(map[string]string{EnvWebhookURL: "https://example.test"}), nil) + + n := New(&bytes.Buffer{}, Config{Mode: ModeBell, FocusMode: FocusAlways}) + MaybeAddWebhookSink(n, nil, nil) + if got := len(n.sinks); got != 0 { + t.Fatalf("nil env must attach nothing, got %d sinks", got) + } +} diff --git a/internal/oauth/encrypt_test.go b/internal/oauth/encrypt_test.go index 3a4c6f4d4..47a2e3687 100644 --- a/internal/oauth/encrypt_test.go +++ b/internal/oauth/encrypt_test.go @@ -108,6 +108,30 @@ func newEncryptedStore(t *testing.T) (*Store, string) { return s, path } +// The unified Storage="encrypted-file" selector must encrypt at rest, the same as +// the legacy Encrypted:true alias (the file/keyring/encrypted-file merge). +func TestNewStoreEncryptedFileStorageSelector(t *testing.T) { + path := filepath.Join(t.TempDir(), "oauth-tokens.json") + s, err := NewStore(StoreOptions{FilePath: path, Storage: "encrypted-file"}) + if err != nil { + t.Fatalf("NewStore(encrypted-file): %v", err) + } + if err := s.Save(ProviderKey("demo"), Token{AccessToken: "super-secret-access"}); err != nil { + t.Fatalf("Save: %v", err) + } + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if strings.Contains(string(raw), "super-secret-access") || strings.Contains(string(raw), "schemaVersion") { + t.Fatalf("Storage=encrypted-file did not encrypt at rest:\n%s", raw) + } + got, ok, err := s.Load(ProviderKey("demo")) + if err != nil || !ok || got.AccessToken != "super-secret-access" { + t.Fatalf("Load = %+v ok=%v err=%v", got, ok, err) + } +} + func TestEncryptedStoreRoundTripAndCiphertextOnDisk(t *testing.T) { s, path := newEncryptedStore(t) tok := Token{AccessToken: "super-secret-access", RefreshToken: "super-secret-refresh", Account: "me@x"} diff --git a/internal/oauth/store.go b/internal/oauth/store.go index 18ecc1255..46f75d7b7 100644 --- a/internal/oauth/store.go +++ b/internal/oauth/store.go @@ -1,16 +1,20 @@ package oauth import ( + "encoding/base64" "encoding/json" "errors" "fmt" "os" "path/filepath" "regexp" + "runtime" "sort" "strings" "sync" "time" + + "github.com/Gitlawb/zero/internal/keyring" ) const ( @@ -54,20 +58,41 @@ type StoreOptions struct { FilePath string Env map[string]string Now func() time.Time - // Encrypted selects the AES-256-GCM encrypted-at-rest backend (a per-user - // secret is created beside the token file). Default (false) writes the 0600 - // plaintext JSON unchanged. + // Storage selects the backend: "" / "file" => a 0600 JSON file (default); + // "encrypted-file" => an AES-256-GCM encrypted file; "keyring" => the OS + // keyring. When empty it falls back to ZERO_OAUTH_STORAGE. + Storage string + // Encrypted is a legacy alias for Storage=="encrypted-file" (AES-256-GCM at + // rest). Ignored when Storage is set. Encrypted bool + // Keyring is the client used when Storage=="keyring"; nil => keyring.New(). + // Injected by tests to avoid touching a real keychain. + Keyring KeyringClient +} + +// KeyringClient is the minimal OS-keyring surface the store needs. *keyring.Keyring +// satisfies it; tests inject a fake. +type KeyringClient interface { + Get(service, account string) (string, bool, error) + Set(service, account, secret string) error + Delete(service, account string) (bool, error) } -// Store persists OAuth tokens (provider + MCP namespaces) in a 0600 file, -// guarded by a cross-process lock and written atomically. The file is plaintext -// JSON by default, or AES-256-GCM ciphertext when the encrypted backend is on. +// Keyring storage stores the whole token blob under one fixed entry. +const ( + keyringService = "zero" + keyringAccount = "oauth-tokens" +) + +// Store persists OAuth tokens (provider + MCP namespaces) as one JSON blob, +// written atomically through a pluggable backend (a 0600 file guarded by a +// cross-process lock, or the OS keyring). When crypter is non-nil the file blob +// is AES-256-GCM ciphertext at rest. type Store struct { - filePath string - crypter *aesGCMCrypter // nil => plaintext backend - now func() time.Time - mu sync.Mutex + blob blobStore + crypter *aesGCMCrypter // nil => plaintext blob + now func() time.Time + mu sync.Mutex } type storeFile struct { @@ -105,33 +130,80 @@ func ResolveStorePath(env map[string]string) (string, error) { return filepath.Join(configHome, "zero", "oauth-tokens.json"), nil } -// NewStore builds a file-backed token store. +// NewStore builds a token store with the configured backend (file by default, +// or the OS keyring when Storage/ZERO_OAUTH_STORAGE selects it). func NewStore(options StoreOptions) (*Store, error) { + now := options.Now + if now == nil { + now = time.Now + } + storage := strings.TrimSpace(options.Storage) + if storage == "" { + storage = strings.TrimSpace(envValue(options.Env, "ZERO_OAUTH_STORAGE")) + } + if storage == "" && options.Encrypted { + storage = "encrypted-file" // legacy alias + } + switch storage { + case "", "file": + path, err := resolveStoreFilePath(options) + if err != nil { + return nil, err + } + return &Store{blob: fileBlob{path: path}, now: now}, nil + case "encrypted-file": + path, err := resolveStoreFilePath(options) + if err != nil { + return nil, err + } + // The file blob holds AES-256-GCM ciphertext; the per-user secret lives in + // a sibling ".secret" file (see encrypt.go). + return &Store{blob: fileBlob{path: path}, crypter: newAESGCMCrypter(path + ".secret"), now: now}, nil + case "keyring": + kr := options.Keyring + if kr == nil { + osKeyring := keyring.New() + if !osKeyring.Available() { + return nil, fmt.Errorf("oauth: keyring storage requested but not available on %s; use file storage", runtime.GOOS) + } + kr = osKeyring + } + // Serialize the keyring's read-modify-write across processes with a lock + // file beside where the file backend would live. Best-effort: if no config + // location resolves, fall back to in-process serialization only. + lockPath := "" + if storePath, perr := ResolveStorePath(options.Env); perr == nil { + lockPath = filepath.Join(filepath.Dir(storePath), "oauth-keyring.lockfile") + } + return &Store{blob: keyringBlob{kr: kr, service: keyringService, account: keyringAccount, lockPath: lockPath}, now: now}, nil + default: + return nil, fmt.Errorf("oauth: unknown storage %q (want \"file\", \"encrypted-file\", or \"keyring\")", storage) + } +} + +// resolveStoreFilePath resolves the absolute file path for the file backend. +func resolveStoreFilePath(options StoreOptions) (string, error) { filePath := options.FilePath var err error if strings.TrimSpace(filePath) == "" { filePath, err = ResolveStorePath(options.Env) if err != nil { - return nil, err + return "", err } } if !filepath.IsAbs(filePath) { filePath, err = filepath.Abs(filePath) if err != nil { - return nil, err + return "", err } } - now := options.Now - if now == nil { - now = time.Now - } - store := &Store{filePath: filepath.Clean(filePath), now: now} - if options.Encrypted { - store.crypter = newAESGCMCrypter(store.filePath + ".secret") - } - return store, nil + return filepath.Clean(filePath), nil } +// FilePath returns the resolved token store location (a path for the file +// backend, or a "keyring:..." identifier for the keyring backend). +func (s *Store) FilePath() string { return s.blob.location() } + // Save persists a token under key, replacing any existing entry. func (s *Store) Save(key string, token Token) error { if err := ValidateKey(key); err != nil { @@ -139,17 +211,14 @@ func (s *Store) Save(key string, token Token) error { } s.mu.Lock() defer s.mu.Unlock() - unlock, err := acquireFileLock(s.filePath+".lockfile", s.now) - if err != nil { - return err - } - defer unlock() - state, err := s.readState() - if err != nil { - return err - } - state.Tokens[key] = token - return s.writeState(state) + return s.blob.withLock(s.now, func() error { + state, err := s.readState() + if err != nil { + return err + } + state.Tokens[key] = token + return s.writeState(state) + }) } // Load returns the token for key; the bool is false when none is stored. @@ -174,23 +243,20 @@ func (s *Store) Delete(key string) (bool, error) { } s.mu.Lock() defer s.mu.Unlock() - unlock, err := acquireFileLock(s.filePath+".lockfile", s.now) - if err != nil { - return false, err - } - defer unlock() - state, err := s.readState() - if err != nil { - return false, err - } - if _, ok := state.Tokens[key]; !ok { - return false, nil - } - delete(state.Tokens, key) - if err := s.writeState(state); err != nil { - return false, err - } - return true, nil + var removed bool + err := s.blob.withLock(s.now, func() error { + state, err := s.readState() + if err != nil { + return err + } + if _, ok := state.Tokens[key]; !ok { + return nil + } + delete(state.Tokens, key) + removed = true + return s.writeState(state) + }) + return removed, err } // Status returns redaction-safe summaries of every stored token, sorted by key. @@ -228,68 +294,159 @@ func (s *Store) Status(prefix string) ([]Status, error) { } func (s *Store) readState() (storeFile, error) { - data, err := os.ReadFile(s.filePath) + data, ok, err := s.blob.read() if err != nil { - if errors.Is(err, os.ErrNotExist) { - return emptyStoreFile(), nil - } return storeFile{}, err } + if !ok { + return emptyStoreFile(), nil + } if s.crypter != nil { + // Encrypted backend: the blob is AES-256-GCM ciphertext, not JSON. data, err = s.crypter.open(data) if err != nil { - return storeFile{}, err + return storeFile{}, fmt.Errorf("oauth: decrypt token store at %s: %w", s.blob.location(), err) } } var state storeFile if err := json.Unmarshal(data, &state); err != nil { - return storeFile{}, fmt.Errorf("oauth: invalid token file at %s: %w", s.filePath, err) + return storeFile{}, fmt.Errorf("oauth: invalid token store at %s: %w", s.blob.location(), err) } if state.SchemaVersion != storeSchemaVersion { - return storeFile{}, fmt.Errorf("oauth: invalid token file at %s: unsupported schemaVersion", s.filePath) + return storeFile{}, fmt.Errorf("oauth: invalid token store at %s: unsupported schemaVersion", s.blob.location()) } if state.Tokens == nil { state.Tokens = map[string]Token{} } for key := range state.Tokens { if err := ValidateKey(key); err != nil { - return storeFile{}, fmt.Errorf("oauth: invalid token file at %s: %w", s.filePath, err) + return storeFile{}, fmt.Errorf("oauth: invalid token store at %s: %w", s.blob.location(), err) } } return state, nil } func (s *Store) writeState(state storeFile) error { - if err := os.MkdirAll(filepath.Dir(s.filePath), 0o700); err != nil { - return err - } data, err := json.MarshalIndent(state, "", " ") if err != nil { return err } + // Plaintext keeps the trailing newline for a tidy file; the encrypted backend + // writes opaque ciphertext instead. payload := append(data, '\n') if s.crypter != nil { - // Encrypted backend: the on-disk file is opaque ciphertext, not JSON. payload, err = s.crypter.seal(data) if err != nil { return err } } - tempPath := fmt.Sprintf("%s.tmp-%d-%d", s.filePath, os.Getpid(), s.now().UnixNano()) - if err := os.WriteFile(tempPath, payload, 0o600); err != nil { + return s.blob.write(payload) +} + +func emptyStoreFile() storeFile { + return storeFile{SchemaVersion: storeSchemaVersion, Tokens: map[string]Token{}} +} + +// blobStore abstracts the persistence of the whole token blob behind the Store, +// so the same store logic backs either a 0600 file or the OS keyring. +type blobStore interface { + // read returns the stored blob; ok is false when nothing is stored yet. + read() (data []byte, ok bool, err error) + // write replaces the stored blob. + write(data []byte) error + // withLock runs fn under whatever cross-process exclusion the backend offers + // (a lock file for the file backend; none for the keyring, which is the + // authoritative store and is serialized within the process by Store.mu). + withLock(now func() time.Time, fn func() error) error + // location is a human-readable identifier for diagnostics/errors. + location() string +} + +// fileBlob persists the blob as a 0600 JSON file, written atomically and guarded +// by a cross-process lock file. Behavior matches the original file store. +type fileBlob struct{ path string } + +func (b fileBlob) read() ([]byte, bool, error) { + data, err := os.ReadFile(b.path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, false, nil + } + return nil, false, err + } + return data, true, nil +} + +func (b fileBlob) write(data []byte) error { + if err := os.MkdirAll(filepath.Dir(b.path), 0o700); err != nil { + return err + } + tempPath := fmt.Sprintf("%s.tmp-%d-%d", b.path, os.Getpid(), time.Now().UnixNano()) + if err := os.WriteFile(tempPath, data, 0o600); err != nil { return err } - if err := os.Rename(tempPath, s.filePath); err != nil { + if err := os.Rename(tempPath, b.path); err != nil { _ = os.Remove(tempPath) return err } return nil } -func emptyStoreFile() storeFile { - return storeFile{SchemaVersion: storeSchemaVersion, Tokens: map[string]Token{}} +func (b fileBlob) withLock(now func() time.Time, fn func() error) error { + unlock, err := acquireFileLock(b.path+".lockfile", now) + if err != nil { + return err + } + defer unlock() + return fn() +} + +func (b fileBlob) location() string { return b.path } + +// keyringBlob persists the blob in the OS keyring as a single base64 entry +// (base64 keeps the multi-line JSON a single, control-character-free value). +type keyringBlob struct { + kr KeyringClient + service string + account string + // lockPath, when set, is a cross-process lock file serializing the keyring's + // read-modify-write so concurrent processes don't clobber each other's tokens. + lockPath string +} + +func (b keyringBlob) read() ([]byte, bool, error) { + enc, ok, err := b.kr.Get(b.service, b.account) + if err != nil || !ok { + return nil, ok, err + } + data, err := base64.StdEncoding.DecodeString(strings.TrimSpace(enc)) + if err != nil { + return nil, false, fmt.Errorf("oauth: decode keyring token blob: %w", err) + } + return data, true, nil } +func (b keyringBlob) write(data []byte) error { + return b.kr.Set(b.service, b.account, base64.StdEncoding.EncodeToString(data)) +} + +// withLock serializes the keyring's read-modify-write. Store.mu covers the +// in-process case; lockPath (when set) adds cross-process exclusion so two +// processes can't both read the blob, modify, and write — dropping a token. +func (b keyringBlob) withLock(now func() time.Time, fn func() error) error { + if b.lockPath == "" { + return fn() + } + unlock, err := acquireFileLock(b.lockPath, now) + if err != nil { + return err + } + defer unlock() + return fn() +} + +func (b keyringBlob) location() string { return "keyring:" + b.service + "/" + b.account } + // FormatStatuses renders a human-readable status table without leaking token // material. func FormatStatuses(statuses []Status) string { diff --git a/internal/oauth/store_keyring_test.go b/internal/oauth/store_keyring_test.go new file mode 100644 index 000000000..8931dc6de --- /dev/null +++ b/internal/oauth/store_keyring_test.go @@ -0,0 +1,113 @@ +package oauth + +import ( + "strings" + "testing" +) + +// fakeKR is an in-memory KeyringClient for exercising the keyring backend +// without touching a real OS keychain. +type fakeKR struct{ data map[string]string } + +func newFakeKR() *fakeKR { return &fakeKR{data: map[string]string{}} } + +func (f *fakeKR) Get(service, account string) (string, bool, error) { + v, ok := f.data[service+"/"+account] + return v, ok, nil +} +func (f *fakeKR) Set(service, account, secret string) error { + f.data[service+"/"+account] = secret + return nil +} +func (f *fakeKR) Delete(service, account string) (bool, error) { + key := service + "/" + account + _, ok := f.data[key] + delete(f.data, key) + return ok, nil +} + +func TestStoreKeyringBackendRoundTrip(t *testing.T) { + // Keep the cross-process keyring lock file inside a temp config dir. + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + kr := newFakeKR() + s, err := NewStore(StoreOptions{Storage: "keyring", Keyring: kr}) + if err != nil { + t.Fatalf("NewStore(keyring): %v", err) + } + if !strings.HasPrefix(s.FilePath(), "keyring:") { + t.Fatalf("FilePath = %q, want keyring identifier", s.FilePath()) + } + + if err := s.Save(ProviderKey("demo"), Token{AccessToken: "a", RefreshToken: "r"}); err != nil { + t.Fatalf("Save: %v", err) + } + got, ok, err := s.Load(ProviderKey("demo")) + if err != nil || !ok { + t.Fatalf("Load: ok=%v err=%v", ok, err) + } + if got.AccessToken != "a" || got.RefreshToken != "r" { + t.Fatalf("Load = %#v", got) + } + + // The blob is stored base64-encoded, so the raw JSON field names never appear. + raw := kr.data[keyringService+"/"+keyringAccount] + if raw == "" { + t.Fatal("nothing stored in keyring") + } + if strings.Contains(raw, "access_token") { + t.Fatalf("keyring blob is not encoded: %s", raw) + } + + removed, err := s.Delete(ProviderKey("demo")) + if err != nil || !removed { + t.Fatalf("Delete: removed=%v err=%v", removed, err) + } + if _, ok, _ := s.Load(ProviderKey("demo")); ok { + t.Fatal("token still present after delete") + } +} + +func TestNewStoreStorageSelection(t *testing.T) { + // Unknown storage is rejected (fail closed). + if _, err := NewStore(StoreOptions{Storage: "bogus"}); err == nil { + t.Fatal("unknown storage should error") + } + // ZERO_OAUTH_STORAGE selects the keyring (with an injected client). + s, err := NewStore(StoreOptions{ + Env: map[string]string{"ZERO_OAUTH_STORAGE": "keyring"}, + Keyring: newFakeKR(), + }) + if err != nil { + t.Fatalf("NewStore(env keyring): %v", err) + } + if !strings.HasPrefix(s.FilePath(), "keyring:") { + t.Fatalf("env did not select keyring backend: %q", s.FilePath()) + } + // Default is the file backend. + fileStore, err := NewStore(StoreOptions{FilePath: t.TempDir() + "/oauth-tokens.json"}) + if err != nil { + t.Fatalf("NewStore(file): %v", err) + } + if strings.HasPrefix(fileStore.FilePath(), "keyring:") { + t.Fatalf("default backend should be file, got %q", fileStore.FilePath()) + } +} + +func TestStoreKeyringStatus(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + kr := newFakeKR() + s, err := NewStore(StoreOptions{Storage: "keyring", Keyring: kr}) + if err != nil { + t.Fatal(err) + } + if err := s.Save(ProviderKey("demo"), Token{AccessToken: "a"}); err != nil { + t.Fatal(err) + } + statuses, err := s.Status(KeyPrefixProvider) + if err != nil { + t.Fatal(err) + } + if len(statuses) != 1 || statuses[0].Key != ProviderKey("demo") || !statuses[0].HasToken { + t.Fatalf("status = %#v", statuses) + } +} diff --git a/internal/providers/gemini/provider.go b/internal/providers/gemini/provider.go index dc98355bc..c39312db6 100644 --- a/internal/providers/gemini/provider.go +++ b/internal/providers/gemini/provider.go @@ -19,10 +19,6 @@ import ( const defaultBaseURL = "https://generativelanguage.googleapis.com" const defaultMaxTokens = 8192 -// providerName tags reasoning signatures this adapter binds to a tool call so -// only it replays them. -const providerName = "gemini" - // thinkingBudgetForEffort maps a requested reasoning effort to a Gemini thinking // token budget, capped at 24576 (the lowest per-model ceiling among 2.5 models). // 0 means "no thinking config" (leave the request unchanged). diff --git a/internal/swarm/schedule_tool.go b/internal/swarm/schedule_tool.go new file mode 100644 index 000000000..d4e4bf56b --- /dev/null +++ b/internal/swarm/schedule_tool.go @@ -0,0 +1,241 @@ +package swarm + +import ( + "context" + "errors" + "fmt" + "math" + "sort" + "strconv" + "strings" + "time" + + "github.com/Gitlawb/zero/internal/tools" +) + +// ScheduleToolName is the recurring-spawn scheduler tool. +const ScheduleToolName = "swarm_schedule" + +// ---- swarm_schedule -------------------------------------------------------- + +type scheduleTool struct { + sw *Swarm + now func() time.Time // injectable for tests; nil => time.Now +} + +func (t *scheduleTool) Name() string { return ScheduleToolName } +func (t *scheduleTool) Description() string { + return "Manage recurring swarm spawns. action=add schedules an agent to spawn on an interval (every, e.g. \"30m\") or daily (daily_at \"HH:MM\"); action=list shows active schedules; action=cancel stops one by job_id. Each fire spawns a fresh member; a fire is skipped while the job's previous spawn is still running." +} + +func (t *scheduleTool) Parameters() tools.Schema { + return tools.Schema{ + Type: "object", + Properties: map[string]tools.PropertySchema{ + "action": {Type: "string", Description: "add (default), list, or cancel."}, + "agent_type": {Type: "string", Description: "add: roster agent type to spawn each fire."}, + "task": {Type: "string", Description: "add: the task/briefing handed to each spawned member."}, + "team": {Type: "string", Description: "Team to spawn into. Defaults to \"default\"."}, + "every": {Type: "string", Description: "add: interval between fires as a Go duration (e.g. \"30m\", \"2h\"). Minimum 1s. Mutually exclusive with daily_at."}, + "daily_at": {Type: "string", Description: "add: local time of day \"HH:MM\" to fire once per day. Mutually exclusive with every."}, + "first_delay": {Type: "string", Description: "add: optional delay before the first fire (Go duration). Ignored with daily_at."}, + "max_runs": {Type: "integer", Description: "add: stop after this many successful spawns. 0 or omitted => unbounded."}, + "job_id": {Type: "string", Description: "cancel: the schedule id to stop."}, + }, + Required: []string{}, + AdditionalProperties: false, + } +} + +func (t *scheduleTool) Safety() tools.Safety { + return tools.Safety{ + // Adding a schedule spawns members repeatedly over time, so the tool is + // classified like swarm_spawn (prompts) even though list/cancel are light. + SideEffect: tools.SideEffectShell, + Permission: tools.PermissionPrompt, + Reason: "Schedules recurring swarm member spawns under the orchestrator's sandbox and policy.", + AdvertiseInAuto: true, + } +} + +func (t *scheduleTool) Run(ctx context.Context, args map[string]any) tools.Result { + return t.RunWithOptions(ctx, args, tools.RunOptions{}) +} + +func (t *scheduleTool) RunWithOptions(_ context.Context, args map[string]any, options tools.RunOptions) tools.Result { + action := strings.ToLower(swarmStr(args, "action")) + if action == "" { + action = "add" + } + switch action { + case "add": + return t.add(args, options) + case "list": + return t.list() + case "cancel": + return t.cancel(args) + default: + return errResult("swarm_schedule: unknown action %q (want add, list, or cancel)", action) + } +} + +func (t *scheduleTool) add(args map[string]any, options tools.RunOptions) tools.Result { + agentType := swarmStr(args, "agent_type") + task := swarmStr(args, "task") + if agentType == "" { + return errResult("swarm_schedule add requires agent_type") + } + if task == "" { + return errResult("swarm_schedule add requires task") + } + + every := swarmStr(args, "every") + dailyAt := swarmStr(args, "daily_at") + if every == "" && dailyAt == "" { + return errResult("swarm_schedule add requires every or daily_at") + } + if every != "" && dailyAt != "" { + return errResult("swarm_schedule add: every and daily_at are mutually exclusive") + } + + var sch Schedule + switch { + case dailyAt != "": + hour, minute, err := parseClock(dailyAt) + if err != nil { + return errResult("%v", err) + } + // Daily mode: Every=24h satisfies validation/display, but the scheduler + // recomputes each fire from Hour:Minute so the wall-clock time holds across + // DST instead of drifting by a fixed 24h. + sch.Every = 24 * time.Hour + sch.Daily = true + sch.Hour = hour + sch.Minute = minute + sch.FirstDelay = nextDailyDelay(t.clock(), hour, minute) + default: + d, err := time.ParseDuration(every) + if err != nil { + return errResult("swarm_schedule add: invalid every %q: %v", every, err) + } + sch.Every = d + if fd := swarmStr(args, "first_delay"); fd != "" { + delay, err := time.ParseDuration(fd) + if err != nil { + return errResult("swarm_schedule add: invalid first_delay %q: %v", fd, err) + } + if delay < 0 { + return errResult("swarm_schedule add: first_delay must be >= 0") + } + sch.FirstDelay = delay + } + } + if mr, ok := swarmInt(args, "max_runs"); ok { + sch.MaxRuns = mr + } + + team := swarmStr(args, "team") + id, err := t.sw.Scheduler().Add(policyFrom(options), team, agentType, task, options.Cwd, sch) + if err != nil { + if errors.Is(err, ErrUnknownAgentType) { + return errResult("%v; available agent types: %s", err, strings.Join(t.sw.Registry().AgentTypes(), ", ")) + } + return errResult("%v", err) + } + cadence := "every " + sch.Every.String() + if dailyAt != "" { + cadence = "daily at " + dailyAt + } + out := fmt.Sprintf("Scheduled %s as %s on team %s (%s).", agentType, id, displayTeam(team), cadence) + res := okResult(out, "swarm", out) + res.Meta = map[string]string{"job_id": id, "team": sanitizeName(team), "agent_type": agentType} + return res +} + +func (t *scheduleTool) list() tools.Result { + jobs := t.sw.Scheduler().List() + sort.Slice(jobs, func(i, j int) bool { return jobs[i].ID < jobs[j].ID }) + var b strings.Builder + fmt.Fprintf(&b, "Scheduled jobs: %d\n", len(jobs)) + for _, j := range jobs { + maxRuns := "unbounded" + if j.MaxRuns > 0 { + maxRuns = strconv.Itoa(j.MaxRuns) + } + fmt.Fprintf(&b, " - %s [%s/%s] every %s, runs %d (max %s), skipped %d: %s\n", + j.ID, j.AgentType, j.Team, j.Every, j.Runs, maxRuns, j.Skipped, collapse(j.Task)) + } + out := strings.TrimRight(b.String(), "\n") + return okResult(out, "swarm", fmt.Sprintf("%d scheduled job(s)", len(jobs))) +} + +func (t *scheduleTool) cancel(args map[string]any) tools.Result { + id := swarmStr(args, "job_id") + if id == "" { + return errResult("swarm_schedule cancel requires job_id") + } + if !t.sw.Scheduler().Cancel(id) { + return errResult("swarm_schedule: no such job %q", id) + } + out := fmt.Sprintf("Cancelled scheduled job %s.", id) + return okResult(out, "swarm", out) +} + +func (t *scheduleTool) clock() time.Time { + if t.now != nil { + return t.now() + } + return time.Now() +} + +// parseClock parses a 24-hour "HH:MM" local time of day. +func parseClock(s string) (hour, minute int, err error) { + parts := strings.Split(strings.TrimSpace(s), ":") + if len(parts) != 2 { + return 0, 0, fmt.Errorf("swarm_schedule: invalid daily_at %q (want HH:MM)", s) + } + hour, err = strconv.Atoi(strings.TrimSpace(parts[0])) + if err != nil || hour < 0 || hour > 23 { + return 0, 0, fmt.Errorf("swarm_schedule: invalid hour in daily_at %q (want 00-23)", s) + } + minute, err = strconv.Atoi(strings.TrimSpace(parts[1])) + if err != nil || minute < 0 || minute > 59 { + return 0, 0, fmt.Errorf("swarm_schedule: invalid minute in daily_at %q (want 00-59)", s) + } + return hour, minute, nil +} + +// swarmInt reads an integer argument, accepting JSON numbers or numeric strings. +func swarmInt(args map[string]any, key string) (int, bool) { + if args == nil { + return 0, false + } + v, ok := args[key] + if !ok { + return 0, false + } + switch n := v.(type) { + case float64: + // Reject non-integer / non-finite JSON numbers so e.g. max_runs=1.9 is an + // error rather than silently truncating to 1. + if math.IsNaN(n) || math.IsInf(n, 0) || math.Trunc(n) != n { + return 0, false + } + return int(n), true + case int: + return n, true + case int64: + return int(n), true + case string: + s := strings.TrimSpace(n) + if s == "" { + return 0, false + } + i, err := strconv.Atoi(s) + if err != nil { + return 0, false + } + return i, true + } + return 0, false +} diff --git a/internal/swarm/schedule_tool_test.go b/internal/swarm/schedule_tool_test.go new file mode 100644 index 000000000..f7d05c2a0 --- /dev/null +++ b/internal/swarm/schedule_tool_test.go @@ -0,0 +1,119 @@ +package swarm + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/Gitlawb/zero/internal/tools" +) + +func TestScheduleToolAddListCancel(t *testing.T) { + reg, sw := newToolSwarm(t, newLauncher(okFor)) + ctx := context.Background() + grant := tools.RunOptions{PermissionGranted: true, Model: "m1", Cwd: "/work"} + + // add (24h interval so the real timer never fires during the test) + res := reg.RunWithOptions(ctx, ScheduleToolName, map[string]any{ + "agent_type": "teammate", + "task": "nightly sweep", + "team": "alpha", + "every": "24h", + "max_runs": float64(3), + }, grant) + if res.Status != tools.StatusOK { + t.Fatalf("add status = %v, output=%q", res.Status, res.Output) + } + jobID := res.Meta["job_id"] + if jobID == "" { + t.Fatal("add must return a job_id in Meta") + } + + // list shows it + res = reg.RunWithOptions(ctx, ScheduleToolName, map[string]any{"action": "list"}, grant) + if res.Status != tools.StatusOK || !strings.Contains(res.Output, jobID) { + t.Fatalf("list missing job %q: status=%v output=%q", jobID, res.Status, res.Output) + } + + // cancel removes it + res = reg.RunWithOptions(ctx, ScheduleToolName, map[string]any{"action": "cancel", "job_id": jobID}, grant) + if res.Status != tools.StatusOK { + t.Fatalf("cancel status = %v, output=%q", res.Status, res.Output) + } + if got := len(sw.Scheduler().List()); got != 0 { + t.Fatalf("after cancel, %d jobs remain, want 0", got) + } + + // cancel again is an error + res = reg.RunWithOptions(ctx, ScheduleToolName, map[string]any{"action": "cancel", "job_id": jobID}, grant) + if res.Status != tools.StatusError { + t.Fatalf("cancel of gone job should error, got %v", res.Status) + } +} + +func TestScheduleToolDailyAt(t *testing.T) { + reg, sw := newToolSwarm(t, newLauncher(okFor)) + // Make the daily-time math deterministic by injecting a fixed clock into the + // tool. The registry already holds a scheduleTool; replace it with one whose + // clock is fixed so first_delay is computed against a known "now". + st := &scheduleTool{sw: sw, now: func() time.Time { return time.Date(2026, 6, 15, 8, 0, 0, 0, time.UTC) }} + reg.Register(st) + + res := reg.RunWithOptions(context.Background(), ScheduleToolName, map[string]any{ + "agent_type": "teammate", + "task": "report", + "daily_at": "23:30", + }, tools.RunOptions{PermissionGranted: true}) + if res.Status != tools.StatusOK { + t.Fatalf("daily_at add status = %v, output=%q", res.Status, res.Output) + } + jobs := sw.Scheduler().List() + if len(jobs) != 1 { + t.Fatalf("want 1 scheduled job, got %d", len(jobs)) + } + if jobs[0].Every != 24*time.Hour { + t.Fatalf("daily_at should schedule a 24h interval, got %s", jobs[0].Every) + } +} + +func TestScheduleToolValidation(t *testing.T) { + reg, _ := newToolSwarm(t, newLauncher(okFor)) + ctx := context.Background() + grant := tools.RunOptions{PermissionGranted: true} + + cases := []struct { + name string + args map[string]any + }{ + {"missing agent_type", map[string]any{"task": "t", "every": "1h"}}, + {"missing task", map[string]any{"agent_type": "teammate", "every": "1h"}}, + {"neither every nor daily_at", map[string]any{"agent_type": "teammate", "task": "t"}}, + {"both every and daily_at", map[string]any{"agent_type": "teammate", "task": "t", "every": "1h", "daily_at": "10:00"}}, + {"invalid every", map[string]any{"agent_type": "teammate", "task": "t", "every": "nope"}}, + {"sub-second every", map[string]any{"agent_type": "teammate", "task": "t", "every": "500ms"}}, + {"unknown agent_type", map[string]any{"agent_type": "ghost", "task": "t", "every": "1h"}}, + {"invalid daily_at", map[string]any{"agent_type": "teammate", "task": "t", "daily_at": "99:99"}}, + {"unknown action", map[string]any{"action": "frobnicate"}}, + {"cancel without id", map[string]any{"action": "cancel"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + res := reg.RunWithOptions(ctx, ScheduleToolName, tc.args, grant) + if res.Status != tools.StatusError { + t.Fatalf("expected error, got status=%v output=%q", res.Status, res.Output) + } + }) + } +} + +func TestScheduleToolRequiresPermission(t *testing.T) { + reg, _ := newToolSwarm(t, newLauncher(okFor)) + // swarm_schedule is a prompt tool: without a grant the registry refuses it. + res := reg.RunWithOptions(context.Background(), ScheduleToolName, map[string]any{ + "agent_type": "teammate", "task": "t", "every": "1h", + }, tools.RunOptions{}) + if res.Status != tools.StatusError { + t.Fatalf("schedule without permission should error, got %v", res.Status) + } +} diff --git a/internal/swarm/scheduler.go b/internal/swarm/scheduler.go new file mode 100644 index 000000000..3cdaccb8d --- /dev/null +++ b/internal/swarm/scheduler.go @@ -0,0 +1,345 @@ +package swarm + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" +) + +// minScheduleInterval is the floor for a recurring spawn. Scheduling is opt-in +// and additive, but a tight loop could still flood a team's queue, so the +// shortest permitted interval is one second. +const minScheduleInterval = time.Second + +// Schedule describes when a scheduled job fires. Scheduling is interval-based +// ("wakeup"): the job first fires after FirstDelay (or Every when FirstDelay is +// zero), then every Every interval, until MaxRuns successful spawns is reached +// or the job/scheduler is stopped. A daily "cron" time sets Daily with Hour/Minute +// (and Every=24h for display/validation); the run loop then recomputes the delay +// to the next local HH:MM each cycle so it holds across DST (see the +// swarm_schedule tool's daily_at handling). +type Schedule struct { + // Every is the interval between fires. Required, must be >= minScheduleInterval. + Every time.Duration + // FirstDelay delays the first fire. Zero => the first fire happens after Every. + FirstDelay time.Duration + // MaxRuns bounds successful spawns. Zero => unbounded (until cancelled). + MaxRuns int + // Daily, when set, recomputes each fire as the next local Hour:Minute rather + // than adding a fixed Every, so a wall-clock daily time does not drift across + // DST transitions. + Daily bool + Hour int + Minute int +} + +func (sch Schedule) validate() error { + if sch.Every < minScheduleInterval { + return fmt.Errorf("swarm: schedule interval must be >= %s", minScheduleInterval) + } + if sch.MaxRuns < 0 { + return errors.New("swarm: schedule max_runs must be >= 0") + } + return nil +} + +// tickerFunc returns a channel that delivers one tick after d, plus a stop func +// that releases the underlying timer. Production uses realTicker; tests inject a +// controllable source. It mirrors time.NewTimer's one-shot semantics: the run +// loop requests a fresh ticker for each interval. +type tickerFunc func(d time.Duration) (<-chan time.Time, func()) + +func realTicker(d time.Duration) (<-chan time.Time, func()) { + t := time.NewTimer(d) + return t.C, func() { t.Stop() } +} + +// JobStatus is a read-only snapshot of a scheduled job for listing. +type JobStatus struct { + ID string + Team string + AgentType string + Task string + Every time.Duration + MaxRuns int + Runs int + // Skipped counts fires that did not spawn — either because the job's previous + // spawn was still running (non-overlap) or a spawn attempt errored. + Skipped int +} + +// scheduledJob is one recurring spawn. Its goroutine owns the timing loop; mu +// guards the mutable counters and the last-spawned task id used for non-overlap. +type scheduledJob struct { + id string + schedule Schedule + policy Policy + team string + agentType string + task string + cwd string + + ctx context.Context + cancel context.CancelFunc + + mu sync.Mutex + runs int + skipped int + lastTask string +} + +func (j *scheduledJob) snapshot() JobStatus { + j.mu.Lock() + defer j.mu.Unlock() + return JobStatus{ + ID: j.id, + Team: j.team, + AgentType: j.agentType, + Task: j.task, + Every: j.schedule.Every, + MaxRuns: j.schedule.MaxRuns, + Runs: j.runs, + Skipped: j.skipped, + } +} + +// Scheduler fires recurring swarm spawns. It is opt-in: nothing runs unless a +// job is explicitly added via Add, and Close stops every job. Each job spawns a +// fresh member per interval through the same Swarm.Spawn path, so members +// inherit the recorded policy and are bounded by the team's slot cap and queue. +type Scheduler struct { + sw *Swarm + newTicker tickerFunc + // now is the wall clock used to recompute a daily job's next local fire each + // cycle (so HH:MM holds across DST). Defaults to time.Now; tests override it. + now func() time.Time + + ctx context.Context + cancel context.CancelFunc + + mu sync.Mutex + jobs map[string]*scheduledJob + closed bool + seq atomic.Uint64 + wg sync.WaitGroup +} + +// newScheduler builds a Scheduler bound to sw. Its context derives from the +// Swarm's base context, so closing the Swarm also stops every scheduled job. +func newScheduler(sw *Swarm) *Scheduler { + ctx, cancel := context.WithCancel(sw.baseCtx) + return &Scheduler{ + sw: sw, + newTicker: realTicker, + now: time.Now, + ctx: ctx, + cancel: cancel, + jobs: map[string]*scheduledJob{}, + } +} + +// clock returns the scheduler's wall clock, defaulting to time.Now. +func (s *Scheduler) clock() time.Time { + if s.now != nil { + return s.now() + } + return time.Now() +} + +// Add registers a recurring spawn and starts its timing loop. It validates the +// schedule and the agent type up front (fail fast) so a bad job never starts. +// The returned id identifies the job for List/Cancel. +func (s *Scheduler) Add(pol Policy, teamName, agentType, task, cwd string, sch Schedule) (string, error) { + if err := sch.validate(); err != nil { + return "", err + } + if _, err := s.sw.registry.Lookup(agentType); err != nil { + return "", err + } + task = strings.TrimSpace(task) + if task == "" { + return "", errors.New("swarm: schedule requires a task") + } + + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return "", errors.New("swarm: scheduler is closed") + } + // A scheduler whose parent context is already canceled (e.g. after + // Swarm.Close) is effectively closed: a new job's loop would exit on the + // first select, so reject Add rather than reporting a job that never runs. + select { + case <-s.ctx.Done(): + return "", errors.New("swarm: scheduler is closed") + default: + } + id := fmt.Sprintf("sched-%d", s.seq.Add(1)) + ctx, cancel := context.WithCancel(s.ctx) + job := &scheduledJob{ + id: id, + schedule: sch, + policy: pol, + team: sanitizeName(teamName), + agentType: agentType, + task: task, + cwd: cwd, + ctx: ctx, + cancel: cancel, + } + s.jobs[id] = job + s.wg.Add(1) + go s.run(job) + return id, nil +} + +// Cancel stops a scheduled job by id. It reports whether a job was found. +func (s *Scheduler) Cancel(id string) bool { + s.mu.Lock() + job, ok := s.jobs[id] + if ok { + delete(s.jobs, id) + } + s.mu.Unlock() + if ok { + job.cancel() + } + return ok +} + +// List returns a snapshot of every active scheduled job. +func (s *Scheduler) List() []JobStatus { + s.mu.Lock() + jobs := make([]*scheduledJob, 0, len(s.jobs)) + for _, j := range s.jobs { + jobs = append(jobs, j) + } + s.mu.Unlock() + out := make([]JobStatus, 0, len(jobs)) + for _, j := range jobs { + out = append(out, j.snapshot()) + } + return out +} + +// Close cancels every job and waits for their loops to exit. Safe to call more +// than once. +func (s *Scheduler) Close() { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.closed = true + s.cancel() + s.mu.Unlock() + s.wg.Wait() +} + +// run is one job's timing loop. It requests a fresh one-shot ticker per interval +// and fires until cancelled or MaxRuns is reached. +func (s *Scheduler) run(job *scheduledJob) { + defer s.wg.Done() + defer s.forget(job.id) + + delay := job.schedule.FirstDelay + if delay <= 0 { + delay = job.schedule.Every + } + for { + ch, stop := s.newTicker(delay) + select { + case <-job.ctx.Done(): + stop() + return + case <-ch: + stop() + } + // A tick and a cancel can be ready together; the select above may pick the + // tick, so re-check cancellation before firing to avoid one extra spawn + // after Cancel/Close. + select { + case <-job.ctx.Done(): + return + default: + } + // Daily jobs recompute the delay to the next local HH:MM each cycle so the + // wall-clock time holds across DST; interval jobs use the fixed Every. + if job.schedule.Daily { + delay = nextDailyDelay(s.clock(), job.schedule.Hour, job.schedule.Minute) + } else { + delay = job.schedule.Every + } + if !s.fireIfIdle(job) { + continue + } + runs := job.incRuns() + if max := job.schedule.MaxRuns; max > 0 && runs >= max { + return + } + } +} + +// fireIfIdle spawns a fresh member unless the job's previous spawn is still +// running (non-overlap). It returns whether a spawn occurred. A spawn error is +// treated like a skip so a transient failure never tears down the loop. +func (s *Scheduler) fireIfIdle(job *scheduledJob) bool { + job.mu.Lock() + last := job.lastTask + job.mu.Unlock() + + if last != "" { + if t, ok := s.sw.coord.Get(last); ok && !t.Status.terminal() { + job.incSkipped() + return false + } + } + + id, err := s.sw.Spawn(job.policy, job.team, job.agentType, job.task, job.cwd) + if err != nil { + job.incSkipped() + return false + } + job.mu.Lock() + job.lastTask = id + job.mu.Unlock() + return true +} + +func (j *scheduledJob) incRuns() int { + j.mu.Lock() + defer j.mu.Unlock() + j.runs++ + return j.runs +} + +func (j *scheduledJob) incSkipped() { + j.mu.Lock() + j.skipped++ + j.mu.Unlock() +} + +func (s *Scheduler) forget(id string) { + s.mu.Lock() + delete(s.jobs, id) + s.mu.Unlock() +} + +// nextDailyDelay returns the duration from now until the next occurrence of the +// given local hour:minute (today if still ahead, otherwise tomorrow). It backs +// the swarm_schedule tool's daily_at ("cron"-style) option. +func nextDailyDelay(now time.Time, hour, minute int) time.Duration { + next := time.Date(now.Year(), now.Month(), now.Day(), hour, minute, 0, 0, now.Location()) + if !next.After(now) { + // Roll to the same wall-clock time on the next calendar day via day+1 + // (not Add(24h)): time.Date normalizes the date and applies the correct + // local offset, so the HH:MM holds across DST (a spring-forward/fall-back + // day is 23h/25h, and a fixed 24h would fire an hour early/late). + next = time.Date(now.Year(), now.Month(), now.Day()+1, hour, minute, 0, 0, now.Location()) + } + return next.Sub(now) +} diff --git a/internal/swarm/scheduler_test.go b/internal/swarm/scheduler_test.go new file mode 100644 index 000000000..b29e11910 --- /dev/null +++ b/internal/swarm/scheduler_test.go @@ -0,0 +1,267 @@ +package swarm + +import ( + "math" + "testing" + "time" +) + +// testTicker returns a ticker factory backed by a single unbounded-handshake +// channel: each send on ticks unblocks exactly one run-loop iteration, so a test +// drives fires deterministically with no real time. +func testTicker(ticks chan time.Time) tickerFunc { + return func(time.Duration) (<-chan time.Time, func()) { + return ticks, func() {} + } +} + +func findJob(jobs []JobStatus, id string) (JobStatus, bool) { + for _, j := range jobs { + if j.ID == id { + return j, true + } + } + return JobStatus{}, false +} + +func TestSchedulerFiresAndCountsRuns(t *testing.T) { + l := newLauncher(okFor) // members complete immediately + sw := newSwarmFor(t, l) + sched := sw.Scheduler() + ticks := make(chan time.Time) + sched.newTicker = testTicker(ticks) + + id, err := sched.Add(Policy{Model: "m"}, "team", "teammate", "ping", "", Schedule{Every: time.Hour, MaxRuns: 3}) + if err != nil { + t.Fatalf("Add: %v", err) + } + + for i := 0; i < 3; i++ { + ticks <- time.Time{} + want := i + 1 + waitFor(t, "task completed", func() bool { return sw.Coordinator().Summarize().Done == want }) + } + + // After MaxRuns the job retires itself. + waitFor(t, "job retired", func() bool { _, ok := findJob(sched.List(), id); return !ok }) + if got := len(l.recorded()); got != 3 { + t.Fatalf("spawned %d members, want 3", got) + } + if done := sw.Coordinator().Summarize().Done; done != 3 { + t.Fatalf("done = %d, want 3", done) + } +} + +func TestSchedulerSkipsWhilePreviousRuns(t *testing.T) { + gate := make(chan struct{}) + l := newLauncher(okFor) + l.gate = gate // members block until released + sw := newSwarmFor(t, l) + sched := sw.Scheduler() + ticks := make(chan time.Time) + sched.newTicker = testTicker(ticks) + + id, err := sched.Add(Policy{Model: "m"}, "team", "teammate", "ping", "", Schedule{Every: time.Hour}) + if err != nil { + t.Fatalf("Add: %v", err) + } + + // Fire 1: spawns and the member stays running (gated). + ticks <- time.Time{} + waitFor(t, "first spawn", func() bool { return len(l.recorded()) == 1 }) + + // Fire 2: previous still running => skipped, no new spawn. + ticks <- time.Time{} + waitFor(t, "skip recorded", func() bool { + j, ok := findJob(sched.List(), id) + return ok && j.Skipped == 1 + }) + if got := len(l.recorded()); got != 1 { + t.Fatalf("a skip must not spawn: recorded %d, want 1", got) + } + + // Release the first member, then fire 3: previous terminal => spawns again. + close(gate) + waitFor(t, "first done", func() bool { return sw.Coordinator().Summarize().Done == 1 }) + ticks <- time.Time{} + waitFor(t, "second spawn", func() bool { return len(l.recorded()) == 2 }) + + j, ok := findJob(sched.List(), id) + if !ok { + t.Fatal("job should still be active (unbounded)") + } + if j.Runs != 2 || j.Skipped != 1 { + t.Fatalf("job runs=%d skipped=%d, want 2/1", j.Runs, j.Skipped) + } +} + +func TestSchedulerAddValidation(t *testing.T) { + sw := newSwarmFor(t, newLauncher(okFor)) + sched := sw.Scheduler() + + if _, err := sched.Add(Policy{}, "team", "teammate", "t", "", Schedule{Every: time.Millisecond}); err == nil { + t.Fatal("sub-second interval should be rejected") + } + if _, err := sched.Add(Policy{}, "team", "nope", "t", "", Schedule{Every: time.Hour}); err == nil { + t.Fatal("unknown agent type should be rejected") + } + if _, err := sched.Add(Policy{}, "team", "teammate", " ", "", Schedule{Every: time.Hour}); err == nil { + t.Fatal("empty task should be rejected") + } + if _, err := sched.Add(Policy{}, "team", "teammate", "t", "", Schedule{Every: time.Hour, MaxRuns: -1}); err == nil { + t.Fatal("negative max_runs should be rejected") + } +} + +func TestSchedulerCancel(t *testing.T) { + sw := newSwarmFor(t, newLauncher(okFor)) + sched := sw.Scheduler() + ticks := make(chan time.Time) + sched.newTicker = testTicker(ticks) // never ticked => never fires + + id, err := sched.Add(Policy{}, "team", "teammate", "t", "", Schedule{Every: time.Hour}) + if err != nil { + t.Fatalf("Add: %v", err) + } + if !sched.Cancel(id) { + t.Fatal("Cancel of an active job should return true") + } + if _, ok := findJob(sched.List(), id); ok { + t.Fatal("cancelled job should be gone from List") + } + if sched.Cancel(id) { + t.Fatal("Cancel of an unknown job should return false") + } +} + +func TestSchedulerCloseStopsJobs(t *testing.T) { + sw, err := New(Options{BaseDir: t.TempDir(), Launcher: newLauncher(okFor)}) + if err != nil { + t.Fatalf("New: %v", err) + } + sched := sw.Scheduler() + ticks := make(chan time.Time) + sched.newTicker = testTicker(ticks) + for i := 0; i < 3; i++ { + if _, err := sched.Add(Policy{}, "team", "teammate", "t", "", Schedule{Every: time.Hour}); err != nil { + t.Fatalf("Add: %v", err) + } + } + sw.Close() // must return (wg waits) and leave no active jobs + if got := len(sched.List()); got != 0 { + t.Fatalf("after Close, %d jobs remain, want 0", got) + } + if _, err := sched.Add(Policy{}, "team", "teammate", "t", "", Schedule{Every: time.Hour}); err == nil { + t.Fatal("Add after Close should fail") + } +} + +func TestSchedulerAddRejectedAfterContextCancel(t *testing.T) { + sw := newSwarmFor(t, newLauncher(okFor)) + sched := sw.Scheduler() + // Parent context canceled but s.closed not yet set: Add must still refuse so it + // never reports a job whose loop exits immediately. + sched.cancel() + if _, err := sched.Add(Policy{}, "team", "teammate", "t", "", Schedule{Every: time.Hour}); err == nil { + t.Fatal("Add after the scheduler context is canceled should fail") + } +} + +func TestSchedulerDailyRecomputesNextDelay(t *testing.T) { + l := newLauncher(okFor) + sw := newSwarmFor(t, l) + sched := sw.Scheduler() + ticks := make(chan time.Time) + var delays []time.Duration + sched.newTicker = func(d time.Duration) (<-chan time.Time, func()) { + delays = append(delays, d) + return ticks, func() {} + } + // Freeze the clock at 10:00 local so the next 11:00 is deterministically 1h — + // not the fixed 24h Every. This is what guards against DST drift. + base := time.Date(2026, 1, 1, 10, 0, 0, 0, time.Local) + sched.now = func() time.Time { return base } + + if _, err := sched.Add(Policy{Model: "m"}, "team", "teammate", "ping", "", + Schedule{Every: 24 * time.Hour, Daily: true, Hour: 11, Minute: 0, FirstDelay: nextDailyDelay(base, 11, 0), MaxRuns: 2}); err != nil { + t.Fatalf("Add: %v", err) + } + for i := 0; i < 2; i++ { + ticks <- time.Time{} + waitFor(t, "task completed", func() bool { return sw.Coordinator().Summarize().Done == i+1 }) + } + if len(delays) < 2 { + t.Fatalf("expected at least two requested delays, got %v", delays) + } + // The second iteration's delay is recomputed from the clock (1h), not Every (24h). + if delays[1] != time.Hour { + t.Fatalf("recomputed daily delay = %s, want 1h (next 11:00 from frozen 10:00)", delays[1]) + } +} + +func TestNextDailyDelay(t *testing.T) { + loc := time.UTC + now := time.Date(2026, 6, 15, 10, 0, 0, 0, loc) + if got := nextDailyDelay(now, 11, 0); got != time.Hour { + t.Fatalf("ahead-today delay = %s, want 1h", got) + } + if got := nextDailyDelay(now, 9, 0); got != 23*time.Hour { + t.Fatalf("passed-today delay = %s, want 23h", got) + } + // Exactly now => next is tomorrow (strictly-after guard). + if got := nextDailyDelay(now, 10, 0); got != 24*time.Hour { + t.Fatalf("equal-now delay = %s, want 24h", got) + } +} + +func TestNextDailyDelayHoldsWallClockAcrossDST(t *testing.T) { + loc, err := time.LoadLocation("America/New_York") + if err != nil { + t.Skipf("tzdata unavailable: %v", err) + } + // US spring-forward 2026-03-08: clocks jump 02:00 -> 03:00 (a 23-hour day). + // At 06:00 EDT the 01:30 target has passed, so the next fire must be 01:30 + // local tomorrow; a fixed Add(24h) would land at 02:30 instead. + now := time.Date(2026, 3, 8, 6, 0, 0, 0, loc) + got := now.Add(nextDailyDelay(now, 1, 30)) + if got.Day() != 9 || got.Hour() != 1 || got.Minute() != 30 { + t.Fatalf("next fire = %s, want 2026-03-09 01:30 local (DST-safe)", got.Format("2006-01-02 15:04 MST")) + } +} + +func TestParseClock(t *testing.T) { + h, m, err := parseClock(" 09:30 ") + if err != nil || h != 9 || m != 30 { + t.Fatalf("parseClock(09:30) = %d,%d,%v", h, m, err) + } + for _, bad := range []string{"24:00", "10:60", "9", "x:y", "10:", ":30", "-1:00"} { + if _, _, err := parseClock(bad); err == nil { + t.Fatalf("parseClock(%q) should error", bad) + } + } +} + +func TestSwarmInt(t *testing.T) { + if v, ok := swarmInt(map[string]any{"n": float64(5)}, "n"); !ok || v != 5 { + t.Fatalf("float64 => %d,%v", v, ok) + } + if v, ok := swarmInt(map[string]any{"n": "7"}, "n"); !ok || v != 7 { + t.Fatalf("string => %d,%v", v, ok) + } + if _, ok := swarmInt(map[string]any{"n": "x"}, "n"); ok { + t.Fatal("non-numeric string should not parse") + } + if _, ok := swarmInt(nil, "n"); ok { + t.Fatal("nil args should not parse") + } + // A non-integer JSON number must be rejected, not truncated to an int. + if v, ok := swarmInt(map[string]any{"n": 1.9}, "n"); ok { + t.Fatalf("non-integer float should not parse, got %d", v) + } + if _, ok := swarmInt(map[string]any{"n": math.Inf(1)}, "n"); ok { + t.Fatal("infinity should not parse") + } + if _, ok := swarmInt(map[string]any{"n": math.NaN()}, "n"); ok { + t.Fatal("NaN should not parse") + } +} diff --git a/internal/swarm/team.go b/internal/swarm/team.go index a6ece1a90..fc3c607d4 100644 --- a/internal/swarm/team.go +++ b/internal/swarm/team.go @@ -64,10 +64,11 @@ type Swarm struct { baseCtx context.Context cancel context.CancelFunc - mu sync.Mutex - teams map[string]*Team - taskCwd map[string]string // taskID -> cwd, for handoff/adoption relaunch - idSeq atomic.Uint64 + mu sync.Mutex + teams map[string]*Team + taskCwd map[string]string // taskID -> cwd, for handoff/adoption relaunch + scheduler *Scheduler // lazily created by Scheduler(); nil until first use + idSeq atomic.Uint64 } // Team is a named set of concurrently-running members with a bounded slot count @@ -136,13 +137,31 @@ func New(opts Options) (*Swarm, error) { } // Close cancels every running member's context and releases resources. It is -// safe to call more than once. +// safe to call more than once. The scheduler is closed first so no new spawn +// fires after shutdown begins. func (s *Swarm) Close() { + s.mu.Lock() + sched := s.scheduler + s.mu.Unlock() + if sched != nil { + sched.Close() + } if s.cancel != nil { s.cancel() } } +// Scheduler returns the swarm's recurring-spawn scheduler, creating it on first +// use. Scheduling is opt-in: until a job is added the scheduler does nothing. +func (s *Swarm) Scheduler() *Scheduler { + s.mu.Lock() + defer s.mu.Unlock() + if s.scheduler == nil { + s.scheduler = newScheduler(s) + } + return s.scheduler +} + // rememberCwd records a task's working dir so a handoff/adoption relaunch keeps it. func (s *Swarm) rememberCwd(taskID, cwd string) { s.mu.Lock() diff --git a/internal/swarm/tools.go b/internal/swarm/tools.go index c6d254daa..5a5b2f422 100644 --- a/internal/swarm/tools.go +++ b/internal/swarm/tools.go @@ -30,6 +30,7 @@ func RegisterTools(registry *tools.Registry, sw *Swarm) { registry.Register(&statusTool{sw: sw}) registry.Register(&handoffTool{sw: sw}) registry.Register(&collectTool{sw: sw}) + registry.Register(&scheduleTool{sw: sw}) } // policyFrom derives the member-inheritance policy from the live tool options so diff --git a/internal/swarm/tools_test.go b/internal/swarm/tools_test.go index fb4b883d4..9922aa0af 100644 --- a/internal/swarm/tools_test.go +++ b/internal/swarm/tools_test.go @@ -30,7 +30,7 @@ func newToolSwarm(t *testing.T, l MemberLauncher) (*tools.Registry, *Swarm) { func TestRegisterToolsRegistersAll(t *testing.T) { reg, _ := newToolSwarm(t, newLauncher(okFor)) - for _, name := range []string{SpawnToolName, SendToolName, InboxToolName, StatusToolName, HandoffToolName, CollectToolName} { + for _, name := range []string{SpawnToolName, SendToolName, InboxToolName, StatusToolName, HandoffToolName, CollectToolName, ScheduleToolName} { if _, ok := reg.Get(name); !ok { t.Fatalf("tool %q not registered", name) } diff --git a/internal/tui/assistant_markdown.go b/internal/tui/assistant_markdown.go index 11bfdb29f..f67fd3ea1 100644 --- a/internal/tui/assistant_markdown.go +++ b/internal/tui/assistant_markdown.go @@ -666,10 +666,6 @@ func renderMarkdownStandaloneLine(line string) string { return strings.TrimRight(line, " ") } -func stripMarkdownInline(text string) string { - return markdownInlinePlain(text) -} - type markdownInlineSegment struct { text string bold bool diff --git a/internal/tui/autocomplete.go b/internal/tui/autocomplete.go index 458c29f17..addbbb2a0 100644 --- a/internal/tui/autocomplete.go +++ b/internal/tui/autocomplete.go @@ -333,16 +333,6 @@ func completePathQueryWithTrailingSpace(value string, cursorPos int, selectedPat return string(out), query.StartIndex + len(replacement) } -func removeTrailingAtToken(value string) string { - if _, ok := trailingAtToken(value); !ok { - return value - } - if i := strings.LastIndexAny(value, " \t\n"); i >= 0 { - return value[:i+1] - } - return "" -} - // maxFileWalk bounds how many filesystem entries the "@file" picker visits per // keystroke so a large workspace tree can't stall the TUI. const maxFileWalk = 4000 diff --git a/internal/tui/command_center.go b/internal/tui/command_center.go index e6a99df4c..bf9532e31 100644 --- a/internal/tui/command_center.go +++ b/internal/tui/command_center.go @@ -343,12 +343,6 @@ func (defaultModelSwitchCompactionPolicy) BeforeModelSwitch(request modelSwitchC } } -type noopModelSwitchCompactionPolicy struct{} - -func (noopModelSwitchCompactionPolicy) BeforeModelSwitch(modelSwitchCompactionRequest) modelSwitchCompactionDecision { - return modelSwitchCompactionDecision{} -} - var modelSwitchCompactionGuard modelSwitchCompactionPolicy = defaultModelSwitchCompactionPolicy{} // sanitizeCardField strips the card protocol's separator bytes from diff --git a/internal/tui/composer.go b/internal/tui/composer.go index 49c32e270..80e312871 100644 --- a/internal/tui/composer.go +++ b/internal/tui/composer.go @@ -72,16 +72,6 @@ func deleteComposerWordAfter(state composerState) composerState { return deleteComposerRange(state, state.cursor, end) } -func deleteComposerLineBefore(state composerState) composerState { - state = normalizeComposerState(state) - return deleteComposerRange(state, composerLineStart(state), state.cursor) -} - -func deleteComposerLineAfter(state composerState) composerState { - state = normalizeComposerState(state) - return deleteComposerRange(state, state.cursor, composerLineEnd(state)) -} - func moveComposerWordBefore(state composerState) composerState { state = normalizeComposerState(state) runes := []rune(state.text) @@ -580,14 +570,6 @@ func shouldInsertCommandArgumentSpace(state composerState, text string) bool { return commandArgumentHintForInput(state.text) != "" } -func deleteCompletedFileMentionBefore(state composerState) (composerState, bool) { - start, end, ok := completedFileMentionRangeBefore(state) - if !ok { - return state, false - } - return deleteComposerRange(state, start, end), true -} - func completedFileMentionRangeBefore(state composerState) (int, int, bool) { state = normalizeComposerState(state) runes := []rune(state.text) diff --git a/internal/tui/mcp_view.go b/internal/tui/mcp_view.go index 8f1f94a4a..fc4061722 100644 --- a/internal/tui/mcp_view.go +++ b/internal/tui/mcp_view.go @@ -113,19 +113,6 @@ func renderMCPView(state MCPViewState, width int) string { return fitMCPManagerLines(lines, width) } -func mcpViewStatus(state MCPViewState) commandStatus { - if !hasMCPOperationalContent(state) { - return commandStatusWarning - } - for _, server := range state.Servers { - status := strings.ToLower(strings.TrimSpace(server.State)) - if strings.Contains(status, "error") || strings.Contains(status, "failed") { - return commandStatusBlocked - } - } - return commandStatusOK -} - func hasMCPViewContent(state MCPViewState) bool { return hasMCPOperationalContent(state) || hasMCPPermissionActivity(state.Permissions) @@ -146,31 +133,6 @@ func hasMCPPermissionActivity(summary MCPPermissionSummary) bool { len(summary.Grants) > 0 } -func mcpServerLines(servers []MCPServerView) []string { - lines := make([]string, 0, len(servers)) - for _, server := range servers { - name := displayValue(strings.TrimSpace(server.Name), "unnamed") - transport := displayValue(strings.TrimSpace(server.Transport), "unknown") - state := displayValue(strings.TrimSpace(server.State), "configured") - - line := fmt.Sprintf("%s [%s] %s", name, transport, state) - if auth := strings.TrimSpace(server.Auth); auth != "" { - line += " " + auth - } - if server.ToolCount > 0 { - line += " " + pluralCount(server.ToolCount, "tool") - } - if target := strings.TrimSpace(server.Target); target != "" { - line += " - " + target - } - lines = append(lines, commandBullet(line)) - if actions := mcpServerActionLine(server); actions != "" { - lines = append(lines, actions) - } - } - return lines -} - func mcpManagerServerLines(servers []MCPServerView) []string { lines := make([]string, 0, len(servers)*3) for index, server := range servers { @@ -380,18 +342,6 @@ func dedupeStrings(values []string) []string { return out } -func fitCommandOutput(output commandOutput, width int) string { - rendered := renderCommandOutput(output) - if width <= 0 { - return rendered - } - lines := strings.Split(rendered, "\n") - for index, line := range lines { - lines[index] = fitStyledLine(line, width) - } - return strings.Join(lines, "\n") -} - func fitMCPManagerLines(lines []string, width int) string { if width <= 0 { return strings.Join(lines, "\n") diff --git a/internal/tui/model.go b/internal/tui/model.go index 4232c8912..c9b730216 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -384,6 +384,10 @@ func newModel(ctx context.Context, options Options) model { Mode: notify.Mode(strings.TrimSpace(options.Notify.Mode)), FocusMode: notify.FocusMode(strings.TrimSpace(options.Notify.FocusMode)), }) + // Opt-in webhook fan-out (ZERO_NOTIFY_WEBHOOK_URL). Delivery failures stay + // silent here: the TUI owns the alt-screen, so writing to stderr would + // corrupt the display. + notify.MaybeAddWebhookSink(notifier, os.Getenv, nil) notifier.SetFocused(true) m := model{ diff --git a/internal/tui/mouse.go b/internal/tui/mouse.go index 2ae9cabaa..d98cf3cea 100644 --- a/internal/tui/mouse.go +++ b/internal/tui/mouse.go @@ -170,28 +170,28 @@ func (m model) syncMouseCapture() (model, tea.Cmd) { return m, tea.DisableMouse } -// Bubble Tea's Type field is deprecated, but its parser still populates it for -// compatibility cases such as left-button drag events. Keep these helpers -// tolerant of both the current Button/Action pair and legacy Type values. +// Mouse classification uses the current Button/Action pair only. Bubble Tea's +// parser always populates Button+Action and merely derives the deprecated Type +// field from them, so checking Type adds nothing — and a left-button drag is +// Action==Motion (which mouseMotion already covers), not a press. func mouseLeftPress(msg tea.MouseMsg) bool { - return msg.Button == tea.MouseButtonLeft && msg.Action == tea.MouseActionPress || - msg.Type == tea.MouseLeft && msg.Action == tea.MouseActionPress + return msg.Button == tea.MouseButtonLeft && msg.Action == tea.MouseActionPress } func mouseMotion(msg tea.MouseMsg) bool { - return msg.Action == tea.MouseActionMotion || msg.Type == tea.MouseMotion + return msg.Action == tea.MouseActionMotion } func mouseRelease(msg tea.MouseMsg) bool { - return msg.Action == tea.MouseActionRelease || msg.Type == tea.MouseRelease + return msg.Action == tea.MouseActionRelease } func mouseWheelUp(msg tea.MouseMsg) bool { - return msg.Button == tea.MouseButtonWheelUp || msg.Type == tea.MouseWheelUp + return msg.Button == tea.MouseButtonWheelUp } func mouseWheelDown(msg tea.MouseMsg) bool { - return msg.Button == tea.MouseButtonWheelDown || msg.Type == tea.MouseWheelDown + return msg.Button == tea.MouseButtonWheelDown } func (m model) mouseOverComposer(msg tea.MouseMsg) bool { diff --git a/internal/tui/mouse_test.go b/internal/tui/mouse_test.go index 2cd0fcf07..7e4671220 100644 --- a/internal/tui/mouse_test.go +++ b/internal/tui/mouse_test.go @@ -348,12 +348,11 @@ func TestTranscriptSelectionLeftDragDoesNotResetAnchor(t *testing.T) { Y: 1, }) m = updated.(model) - // Bubble Tea marks left-button drag motion as Type MouseLeft for backward - // compatibility; this must update the cursor without resetting the anchor. + // A left-button drag is Action==Motion with Button==Left; this must update the + // cursor without resetting the selection anchor. updated, _ = m.Update(tea.MouseMsg{ Button: tea.MouseButtonLeft, Action: tea.MouseActionMotion, - Type: tea.MouseLeft, X: 8, Y: 1, }) diff --git a/internal/tui/picker.go b/internal/tui/picker.go index 8cbf75cd6..76b0ef93c 100644 --- a/internal/tui/picker.go +++ b/internal/tui/picker.go @@ -85,11 +85,6 @@ func (p *commandPicker) deleteQueryRune() { p.applyQuery() } -func (p *commandPicker) clearQuery() { - p.query = "" - p.applyQuery() -} - func (p *commandPicker) applyQuery() { source := p.allItems if len(source) == 0 { diff --git a/internal/tui/theme.go b/internal/tui/theme.go index 880e9bc3c..893eb7db8 100644 --- a/internal/tui/theme.go +++ b/internal/tui/theme.go @@ -177,11 +177,6 @@ func (t tuiTheme) onUserPrompt(style lipgloss.Style) lipgloss.Style { return style.Background(lipgloss.Color(colorPromptBg)) } -// onPanel2 paints on the header/picker-row surface. -func (t tuiTheme) onPanel2(style lipgloss.Style) lipgloss.Style { - return style.Background(lipgloss.Color(colorPanel2)) -} - // onSel paints on the selected-row tint. func (t tuiTheme) onSel(style lipgloss.Style) lipgloss.Style { return style.Background(lipgloss.Color(colorSelBg)) diff --git a/internal/tui/view.go b/internal/tui/view.go index 5581ea854..4ab613cea 100644 --- a/internal/tui/view.go +++ b/internal/tui/view.go @@ -839,15 +839,6 @@ func firstArgValue(raw string, keys []string) string { return "" } -func indentText(text string, spaces int) string { - prefix := strings.Repeat(" ", spaces) - lines := strings.Split(text, "\n") - for index, line := range lines { - lines[index] = prefix + line - } - return strings.Join(lines, "\n") -} - // looksLikeDiff reports whether output should be rendered as a diff card: a // real hunk header, or both old/new file headers. A single line starting with // "---" (a Markdown rule, YAML document marker, log separator…) must NOT