Add artifact workdir layout and locking

2026-05-02 10:56:42 -05:00
parent 6caecb1baa
commit fb659e0d3d
8 changed files with 615 additions and 46 deletions
--- a/internal/artifacts/local.go
+++ b/internal/artifacts/local.go
@@ -1,31 +1,291 @@
 package artifacts

 import (
-	"context"
+	"errors"
 	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
 )

-// LocalStore is a placeholder local filesystem artifact store.
+// ErrLockConflict is returned when a session lock already exists.
+var ErrLockConflict = errors.New("session workdir is already locked")
+
+// LockHandle tracks a held lock for a session work directory.
+type LockHandle struct {
+	path string
+	file *os.File
+}
+
+// LocalStore stores artifacts and session working state on the local filesystem.
 type LocalStore struct {
-	RootDir string
+	WorkspaceRoot string
 }

-// WriteLocal returns a not-yet-implemented error in the scaffold.
-func (s *LocalStore) WriteLocal(_ context.Context, _ Ref, _ []byte) error {
-	return fmt.Errorf("artifacts local write: not yet implemented")
+// NewLocalStore constructs a local store rooted at workspaceRoot.
+func NewLocalStore(workspaceRoot string) *LocalStore {
+	return &LocalStore{WorkspaceRoot: workspaceRoot}
 }

-// ReadLocal returns a not-yet-implemented error in the scaffold.
-func (s *LocalStore) ReadLocal(_ context.Context, _ Ref) ([]byte, error) {
-	return nil, fmt.Errorf("artifacts local read: not yet implemented")
+// SessionPaths resolves canonical paths for a session workdir.
+func (s *LocalStore) SessionPaths(sessionID string) SessionPaths {
+	return buildSessionPaths(s.WorkspaceRoot, sessionID)
 }

-// ExistsLocal returns a not-yet-implemented error in the scaffold.
-func (s *LocalStore) ExistsLocal(_ context.Context, _ Ref) (bool, error) {
-	return false, fmt.Errorf("artifacts local exists: not yet implemented")
+// EnsureLayout creates and verifies the canonical session workdir directory layout.
+func (s *LocalStore) EnsureLayout(sessionID string) (SessionPaths, error) {
+	if strings.TrimSpace(s.WorkspaceRoot) == "" {
+		return SessionPaths{}, fmt.Errorf("workspace root is required")
+	}
+	if strings.TrimSpace(sessionID) == "" {
+		return SessionPaths{}, fmt.Errorf("sessionID is required")
+	}
+
+	paths := s.SessionPaths(sessionID)
+	dirs := []string{
+		paths.Root,
+		paths.InputsDir,
+		paths.AudioDir,
+		paths.TranscriptsDir,
+		paths.TranscriptsRawDir,
+		paths.TranscriptsNormalizedDir,
+		paths.ArtifactsDir,
+		paths.ConfigDir,
+		paths.LogsDir,
+	}
+
+	for _, dir := range dirs {
+		if err := os.MkdirAll(dir, 0o755); err != nil {
+			return SessionPaths{}, fmt.Errorf("ensure layout: create %q: %w", dir, err)
+		}
+	}
+
+	return paths, nil
 }

-// Upload returns a not-yet-implemented error in the scaffold.
-func (s *LocalStore) Upload(_ context.Context, _ Ref) (string, error) {
-	return "", fmt.Errorf("artifacts local upload: not yet implemented")
+// CopyInput copies an input file into the session workdir under destRelativePath.
+func (s *LocalStore) CopyInput(sessionID, srcPath, destRelativePath string) (Ref, error) {
+	paths, err := s.EnsureLayout(sessionID)
+	if err != nil {
+		return Ref{}, err
+	}
+
+	destAbs, err := resolveInRoot(paths.Root, destRelativePath)
+	if err != nil {
+		return Ref{}, fmt.Errorf("copy input: %w", err)
+	}
+
+	if err := copyFileAtomic(srcPath, destAbs, 0o644); err != nil {
+		return Ref{}, fmt.Errorf("copy input %q -> %q: %w", srcPath, destAbs, err)
+	}
+
+	checksum, err := s.Checksum(destAbs)
+	if err != nil {
+		return Ref{}, fmt.Errorf("copy input checksum %q: %w", destAbs, err)
+	}
+
+	return Ref{
+		Kind:         "input",
+		Category:     "inputs",
+		SessionID:    sessionID,
+		RelativePath: filepath.Clean(destRelativePath),
+		AbsolutePath: destAbs,
+		Checksum:     checksum,
+	}, nil
+}
+
+// Exists reports whether path exists.
+func (s *LocalStore) Exists(path string) (bool, error) {
+	_, err := os.Stat(path)
+	if err == nil {
+		return true, nil
+	}
+	if errors.Is(err, os.ErrNotExist) {
+		return false, nil
+	}
+	return false, fmt.Errorf("exists %q: %w", path, err)
+}
+
+// ExistsRef reports whether ref.AbsolutePath exists.
+func (s *LocalStore) ExistsRef(ref Ref) (bool, error) {
+	if strings.TrimSpace(ref.AbsolutePath) == "" {
+		return false, fmt.Errorf("exists ref: absolute path is required")
+	}
+	return s.Exists(ref.AbsolutePath)
+}
+
+// WriteFileAtomic writes a file via temp-file + rename within the same directory.
+func (s *LocalStore) WriteFileAtomic(path string, data []byte, perm os.FileMode) error {
+	if strings.TrimSpace(path) == "" {
+		return fmt.Errorf("write file atomic: path is required")
+	}
+
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return fmt.Errorf("write file atomic: create parent dir %q: %w", dir, err)
+	}
+
+	base := filepath.Base(path)
+	tmp, err := os.CreateTemp(dir, "."+base+".tmp-*")
+	if err != nil {
+		return fmt.Errorf("write file atomic: create temp file: %w", err)
+	}
+	tmpName := tmp.Name()
+	removeTmp := true
+	defer func() {
+		if removeTmp {
+			_ = os.Remove(tmpName)
+		}
+	}()
+
+	if _, err := tmp.Write(data); err != nil {
+		_ = tmp.Close()
+		return fmt.Errorf("write file atomic: write temp file: %w", err)
+	}
+	if err := tmp.Sync(); err != nil {
+		_ = tmp.Close()
+		return fmt.Errorf("write file atomic: sync temp file: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return fmt.Errorf("write file atomic: close temp file: %w", err)
+	}
+
+	if err := os.Chmod(tmpName, perm); err != nil {
+		return fmt.Errorf("write file atomic: chmod temp file: %w", err)
+	}
+	if err := os.Rename(tmpName, path); err != nil {
+		return fmt.Errorf("write file atomic: rename temp file: %w", err)
+	}
+	removeTmp = false
+
+	return nil
+}
+
+// Checksum computes SHA-256 for a file path.
+func (s *LocalStore) Checksum(path string) (string, error) {
+	digest, err := SHA256File(path)
+	if err != nil {
+		return "", fmt.Errorf("checksum %q: %w", path, err)
+	}
+	return digest, nil
+}
+
+// AcquireSessionLock acquires an exclusive lock file for a session workdir.
+func (s *LocalStore) AcquireSessionLock(sessionID string) (*LockHandle, error) {
+	paths, err := s.EnsureLayout(sessionID)
+	if err != nil {
+		return nil, err
+	}
+
+	f, err := os.OpenFile(paths.LockPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
+	if err != nil {
+		if errors.Is(err, os.ErrExist) {
+			return nil, fmt.Errorf("%w: %s", ErrLockConflict, paths.LockPath)
+		}
+		return nil, fmt.Errorf("acquire lock %q: %w", paths.LockPath, err)
+	}
+
+	metadata := "pid=" + strconv.Itoa(os.Getpid()) + "\nacquired_at=" + time.Now().UTC().Format(time.RFC3339Nano) + "\n"
+	if _, err := io.WriteString(f, metadata); err != nil {
+		_ = f.Close()
+		_ = os.Remove(paths.LockPath)
+		return nil, fmt.Errorf("acquire lock %q: write metadata: %w", paths.LockPath, err)
+	}
+	if err := f.Sync(); err != nil {
+		_ = f.Close()
+		_ = os.Remove(paths.LockPath)
+		return nil, fmt.Errorf("acquire lock %q: sync: %w", paths.LockPath, err)
+	}
+
+	return &LockHandle{path: paths.LockPath, file: f}, nil
+}
+
+// ReleaseSessionLock releases a previously acquired session lock.
+func (s *LocalStore) ReleaseSessionLock(lock *LockHandle) error {
+	if lock == nil {
+		return nil
+	}
+
+	var closeErr error
+	if lock.file != nil {
+		closeErr = lock.file.Close()
+		lock.file = nil
+	}
+
+	removeErr := os.Remove(lock.path)
+	if errors.Is(removeErr, os.ErrNotExist) {
+		removeErr = nil
+	}
+
+	if closeErr != nil {
+		return fmt.Errorf("release lock %q: close: %w", lock.path, closeErr)
+	}
+	if removeErr != nil {
+		return fmt.Errorf("release lock %q: remove: %w", lock.path, removeErr)
+	}
+	return nil
+}
+
+func resolveInRoot(root, relative string) (string, error) {
+	rel := filepath.Clean(relative)
+	if rel == "." || rel == "" {
+		return "", fmt.Errorf("relative destination path is required")
+	}
+	if filepath.IsAbs(rel) {
+		return "", fmt.Errorf("relative destination must not be absolute: %q", relative)
+	}
+	if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("relative destination escapes root: %q", relative)
+	}
+	return filepath.Join(root, rel), nil
+}
+
+func copyFileAtomic(srcPath, dstPath string, perm os.FileMode) error {
+	src, err := os.Open(srcPath)
+	if err != nil {
+		return err
+	}
+	defer src.Close()
+
+	if err := os.MkdirAll(filepath.Dir(dstPath), 0o755); err != nil {
+		return err
+	}
+
+	dir := filepath.Dir(dstPath)
+	base := filepath.Base(dstPath)
+	tmp, err := os.CreateTemp(dir, "."+base+".tmp-*")
+	if err != nil {
+		return err
+	}
+	tmpName := tmp.Name()
+	removeTmp := true
+	defer func() {
+		if removeTmp {
+			_ = os.Remove(tmpName)
+		}
+	}()
+
+	if _, err := io.Copy(tmp, src); err != nil {
+		_ = tmp.Close()
+		return err
+	}
+	if err := tmp.Sync(); err != nil {
+		_ = tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	if err := os.Chmod(tmpName, perm); err != nil {
+		return err
+	}
+	if err := os.Rename(tmpName, dstPath); err != nil {
+		return err
+	}
+	removeTmp = false
+
+	return nil
 }