| 1 | package sessioninbox |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "sort" |
| 10 | "strings" |
| 11 | "time" |
| 12 | |
| 13 | "reasonix/internal/fileutil" |
| 14 | "reasonix/internal/store" |
| 15 | ) |
| 16 | |
| 17 | func (s *Store) mutableLocked() error { |
| 18 | if s.closed { |
| 19 | return ErrClosed |
| 20 | } |
| 21 | if s.readonly { |
| 22 | return ErrSchemaReadonly |
| 23 | } |
| 24 | return nil |
| 25 | } |
| 26 | |
| 27 | func (s *Store) blobPath(blobName string) (string, error) { |
| 28 | if !validBlobStem(blobName) { |
| 29 | return "", fmt.Errorf("sessioninbox: invalid blob name") |
| 30 | } |
| 31 | base := filepath.Join(s.dir, blobsDirName) |
| 32 | path := filepath.Join(base, blobName+blobSuffix) |
| 33 | rel, err := filepath.Rel(base, path) |
| 34 | if err != nil || rel == "." || !filepath.IsLocal(rel) { |
| 35 | return "", fmt.Errorf("sessioninbox: blob path escapes inbox") |
| 36 | } |
| 37 | return path, nil |
| 38 | } |
| 39 | |
| 40 | // blobNameFor returns the on-disk blob stem for a meta entry. |
| 41 | func blobNameFor(meta InboxItemMeta) string { |
| 42 | if name := strings.TrimSpace(meta.BlobName); name != "" { |
| 43 | return name |
| 44 | } |
| 45 | return meta.ID |
| 46 | } |
| 47 | |
| 48 | func (s *Store) writeBlobLocked(blobName string, data []byte) error { |
| 49 | if err := ensurePrivateDir(s.dir); err != nil { |
| 50 | return err |
| 51 | } |
| 52 | if err := ensurePrivateDir(filepath.Join(s.dir, blobsDirName)); err != nil { |
| 53 | return fmt.Errorf("sessioninbox: blobs dir: %w", err) |
| 54 | } |
| 55 | path, err := s.blobPath(blobName) |
| 56 | if err != nil { |
| 57 | return err |
| 58 | } |
| 59 | fileutil.Crash("inbox-blob-write", path) |
| 60 | if err := fileutil.AtomicWriteFileStrict(path, data, 0o600); err != nil { |
| 61 | return fmt.Errorf("sessioninbox: write blob: %w", err) |
| 62 | } |
| 63 | fileutil.Crash("inbox-blob-rename", path) |
| 64 | return nil |
| 65 | } |
| 66 | |
| 67 | func (s *Store) readBlobLocked(blobName, wantChecksum string) (PromptEnvelope, error) { |
| 68 | if err := validatePrivateDir(filepath.Join(s.dir, blobsDirName)); err != nil { |
| 69 | return PromptEnvelope{}, fmt.Errorf("sessioninbox: blobs dir: %w", err) |
| 70 | } |
| 71 | path, err := s.blobPath(blobName) |
| 72 | if err != nil { |
| 73 | return PromptEnvelope{}, err |
| 74 | } |
| 75 | data, err := readRegularFile(path, s.limits.MaxItemBytes) |
| 76 | if err != nil { |
| 77 | return PromptEnvelope{}, fmt.Errorf("sessioninbox: read blob: %w", err) |
| 78 | } |
| 79 | got := sha256Hex(data) |
| 80 | if wantChecksum != "" && got != wantChecksum { |
| 81 | return PromptEnvelope{}, fmt.Errorf("sessioninbox: blob checksum mismatch") |
| 82 | } |
| 83 | var env PromptEnvelope |
| 84 | if err := json.Unmarshal(data, &env); err != nil { |
| 85 | return PromptEnvelope{}, fmt.Errorf("sessioninbox: decode blob: %w", err) |
| 86 | } |
| 87 | return env, nil |
| 88 | } |
| 89 | |
| 90 | func ensurePrivateDir(path string) error { |
| 91 | if err := os.MkdirAll(path, 0o700); err != nil { |
| 92 | return err |
| 93 | } |
| 94 | return validatePrivateDir(path) |
| 95 | } |
| 96 | |
| 97 | func validatePrivateDir(path string) error { |
| 98 | info, err := os.Lstat(path) |
| 99 | if err != nil { |
| 100 | return err |
| 101 | } |
| 102 | if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() { |
| 103 | return fmt.Errorf("refusing non-directory or symlink") |
| 104 | } |
| 105 | return nil |
| 106 | } |
| 107 | |
| 108 | func readRegularFile(path string, maxBytes int64) ([]byte, error) { |
| 109 | before, err := os.Lstat(path) |
| 110 | if err != nil { |
| 111 | return nil, err |
| 112 | } |
| 113 | if before.Mode()&os.ModeSymlink != 0 || !before.Mode().IsRegular() { |
| 114 | return nil, fmt.Errorf("refusing non-regular file") |
| 115 | } |
| 116 | if maxBytes > 0 && before.Size() > maxBytes { |
| 117 | return nil, fmt.Errorf("file exceeds %d bytes", maxBytes) |
| 118 | } |
| 119 | f, err := os.Open(path) |
| 120 | if err != nil { |
| 121 | return nil, err |
| 122 | } |
| 123 | defer f.Close() |
| 124 | opened, err := f.Stat() |
| 125 | if err != nil { |
| 126 | return nil, err |
| 127 | } |
| 128 | if !opened.Mode().IsRegular() || !os.SameFile(before, opened) { |
| 129 | return nil, fmt.Errorf("file changed while opening") |
| 130 | } |
| 131 | reader := io.Reader(f) |
| 132 | if maxBytes > 0 { |
| 133 | reader = io.LimitReader(f, maxBytes+1) |
| 134 | } |
| 135 | data, err := io.ReadAll(reader) |
| 136 | if err != nil { |
| 137 | return nil, err |
| 138 | } |
| 139 | if maxBytes > 0 && int64(len(data)) > maxBytes { |
| 140 | return nil, fmt.Errorf("file exceeds %d bytes", maxBytes) |
| 141 | } |
| 142 | return data, nil |
| 143 | } |
| 144 | |
| 145 | func (s *Store) commitManifestLocked(next *manifest) error { |
| 146 | if next == nil { |
| 147 | return fmt.Errorf("sessioninbox: nil manifest") |
| 148 | } |
| 149 | if err := validateManifest(next, false); err != nil { |
| 150 | return fmt.Errorf("sessioninbox: invalid manifest: %w", err) |
| 151 | } |
| 152 | if err := ensurePrivateDir(s.dir); err != nil { |
| 153 | return fmt.Errorf("sessioninbox: mkdir: %w", err) |
| 154 | } |
| 155 | next.SchemaVersion = SchemaVersion |
| 156 | next.RunID = s.runID |
| 157 | next.Revision++ |
| 158 | next.UpdatedAt = time.Now().UTC() |
| 159 | data, err := json.MarshalIndent(next, "", " ") |
| 160 | if err != nil { |
| 161 | return err |
| 162 | } |
| 163 | data = append(data, '\n') |
| 164 | path := filepath.Join(s.dir, manifestName) |
| 165 | fileutil.Crash("inbox-manifest-write", path) |
| 166 | if err := fileutil.AtomicWriteFileStrict(path, data, 0o600); err != nil { |
| 167 | return fmt.Errorf("sessioninbox: write manifest: %w", err) |
| 168 | } |
| 169 | fileutil.Crash("inbox-manifest-commit", path) |
| 170 | // Best-effort directory fsync for durability of the rename. |
| 171 | if d, err := os.Open(s.dir); err == nil { |
| 172 | _ = d.Sync() |
| 173 | _ = d.Close() |
| 174 | } |
| 175 | s.man = next |
| 176 | return nil |
| 177 | } |
| 178 | |
| 179 | func (s *Store) quarantineFileLocked(path, tag string) error { |
| 180 | qdir := filepath.Join(s.dir, quarantineName) |
| 181 | if err := ensurePrivateDir(qdir); err != nil { |
| 182 | return err |
| 183 | } |
| 184 | base := filepath.Base(path) + "." + tag + "." + fmt.Sprintf("%d", time.Now().UnixNano()) |
| 185 | return os.Rename(path, filepath.Join(qdir, base)) |
| 186 | } |
| 187 | |
| 188 | func (s *Store) gcOrphansLocked() { |
| 189 | bdir := filepath.Join(s.dir, blobsDirName) |
| 190 | if err := validatePrivateDir(bdir); err != nil { |
| 191 | return |
| 192 | } |
| 193 | entries, err := os.ReadDir(bdir) |
| 194 | if err != nil { |
| 195 | return |
| 196 | } |
| 197 | live := make(map[string]struct{}, len(s.man.Items)) |
| 198 | for _, it := range s.man.Items { |
| 199 | live[blobNameFor(it)] = struct{}{} |
| 200 | } |
| 201 | qdir := filepath.Join(s.dir, quarantineName) |
| 202 | for _, e := range entries { |
| 203 | if e.IsDir() { |
| 204 | continue |
| 205 | } |
| 206 | name := e.Name() |
| 207 | if !strings.HasSuffix(name, blobSuffix) { |
| 208 | _ = s.quarantineUnknownLocked(filepath.Join(bdir, name)) |
| 209 | continue |
| 210 | } |
| 211 | stem := strings.TrimSuffix(name, blobSuffix) |
| 212 | if _, ok := live[stem]; ok { |
| 213 | continue |
| 214 | } |
| 215 | // Orphan blob → quarantine (do not delete silently: crash recovery). |
| 216 | _ = ensurePrivateDir(qdir) |
| 217 | _ = os.Rename(filepath.Join(bdir, name), filepath.Join(qdir, name+"."+fmt.Sprintf("%d", time.Now().UnixNano()))) |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | // salvageOrphanBlobsLocked rebuilds uncertain meta rows from blob files after a |
| 222 | // corrupt-manifest quarantine. Bodies stay on disk; the user reviews before resume. |
| 223 | func (s *Store) salvageOrphanBlobsLocked() []InboxItemMeta { |
| 224 | bdir := filepath.Join(s.dir, blobsDirName) |
| 225 | if err := validatePrivateDir(bdir); err != nil { |
| 226 | return nil |
| 227 | } |
| 228 | entries, err := os.ReadDir(bdir) |
| 229 | if err != nil { |
| 230 | return nil |
| 231 | } |
| 232 | now := time.Now().UTC() |
| 233 | var out []InboxItemMeta |
| 234 | for _, e := range entries { |
| 235 | if e.IsDir() || e.Type()&os.ModeSymlink != 0 || !strings.HasSuffix(e.Name(), blobSuffix) { |
| 236 | continue |
| 237 | } |
| 238 | stem := strings.TrimSuffix(e.Name(), blobSuffix) |
| 239 | if !validBlobStem(stem) { |
| 240 | continue |
| 241 | } |
| 242 | data, err := readRegularFile(filepath.Join(bdir, e.Name()), s.limits.MaxItemBytes) |
| 243 | if err != nil { |
| 244 | continue |
| 245 | } |
| 246 | var env PromptEnvelope |
| 247 | if err := json.Unmarshal(data, &env); err != nil { |
| 248 | continue |
| 249 | } |
| 250 | preview := PreviewText(firstNonEmpty(env.DisplayText, env.SubmitText, env.RawText), DefaultPreviewRunes) |
| 251 | if preview == "" { |
| 252 | preview = "(salvaged body)" |
| 253 | } |
| 254 | // The corrupt manifest no longer provides a revision-to-item mapping. |
| 255 | // Use the complete blob stem as a collision-free recovered item ID. |
| 256 | itemID := stem |
| 257 | out = append(out, InboxItemMeta{ |
| 258 | ID: itemID, |
| 259 | Intent: IntentFollowup, |
| 260 | State: StateUncertain, |
| 261 | BlobName: stem, |
| 262 | CreatedAt: now, |
| 263 | UpdatedAt: now, |
| 264 | Preview: preview, |
| 265 | ByteSize: int64(len(data)), |
| 266 | Checksum: sha256Hex(data), |
| 267 | RunID: s.runID, |
| 268 | BlockReason: "salvaged after corrupt manifest", |
| 269 | }) |
| 270 | } |
| 271 | return out |
| 272 | } |
| 273 | |
| 274 | func (s *Store) quarantineUnknownLocked(path string) error { |
| 275 | qdir := filepath.Join(s.dir, quarantineName) |
| 276 | if err := ensurePrivateDir(qdir); err != nil { |
| 277 | return err |
| 278 | } |
| 279 | return os.Rename(path, filepath.Join(qdir, filepath.Base(path)+"."+fmt.Sprintf("%d", time.Now().UnixNano()))) |
| 280 | } |
| 281 | |
| 282 | func (s *Store) notifyLocked(snap InboxSnapshot) { |
| 283 | listeners := append([]func(InboxSnapshot){}, s.listeners...) |
| 284 | // Unlock is held; notify asynchronously so listeners can re-enter. |
| 285 | go func() { |
| 286 | for _, fn := range listeners { |
| 287 | fn(snap) |
| 288 | } |
| 289 | }() |
| 290 | } |
| 291 | |
| 292 | func encodeEnvelope(env PromptEnvelope) (data []byte, checksum string, size int64, err error) { |
| 293 | data, err = json.Marshal(env) |
| 294 | if err != nil { |
| 295 | return nil, "", 0, err |
| 296 | } |
| 297 | return data, sha256Hex(data), int64(len(data)), nil |
| 298 | } |
| 299 | |
| 300 | // idempotencyRequestHash fingerprints stable client intent. Enqueue-time |
| 301 | // reference materialization is deliberately excluded so a network retry does |
| 302 | // not conflict merely because the referenced workspace changed meanwhile. |
| 303 | func idempotencyRequestHash(env PromptEnvelope) (string, error) { |
| 304 | if env.FingerprintVersion != 0 { |
| 305 | if env.FingerprintVersion != 1 || len(env.RequestFingerprint) != 64 { |
| 306 | return "", fmt.Errorf("unsupported or invalid inbox request fingerprint; automatic replay is disabled") |
| 307 | } |
| 308 | return env.RequestFingerprint, nil |
| 309 | } |
| 310 | type stableInvocation struct { |
| 311 | Name string `json:"name,omitempty"` |
| 312 | Args map[string]string `json:"args,omitempty"` |
| 313 | Display string `json:"display,omitempty"` |
| 314 | } |
| 315 | storedInvocations := append([]StructuredInvocation(nil), env.Invocations...) |
| 316 | if len(storedInvocations) == 0 && env.Invocation != nil { |
| 317 | storedInvocations = []StructuredInvocation{*env.Invocation} |
| 318 | } |
| 319 | sort.SliceStable(storedInvocations, func(i, j int) bool { |
| 320 | return storedInvocations[i].Offset < storedInvocations[j].Offset |
| 321 | }) |
| 322 | invocations := make([]stableInvocation, 0, len(storedInvocations)) |
| 323 | for _, invocation := range storedInvocations { |
| 324 | invocations = append(invocations, stableInvocation{ |
| 325 | Name: invocation.Name, Args: invocation.Args, Display: invocation.Display, |
| 326 | }) |
| 327 | } |
| 328 | stable := struct { |
| 329 | DisplayText string `json:"displayText"` |
| 330 | RawText string `json:"rawText"` |
| 331 | SubmitText string `json:"submitText"` |
| 332 | Invocations []stableInvocation `json:"invocations,omitempty"` |
| 333 | Format string `json:"format,omitempty"` |
| 334 | Attachments []string `json:"attachments,omitempty"` |
| 335 | AttachmentIdentities []string `json:"attachmentIdentities,omitempty"` |
| 336 | ExplicitRefs []string `json:"explicitRefs,omitempty"` |
| 337 | Source string `json:"source,omitempty"` |
| 338 | Extra map[string]string `json:"extra,omitempty"` |
| 339 | }{ |
| 340 | DisplayText: env.DisplayText, |
| 341 | RawText: env.RawText, |
| 342 | SubmitText: env.SubmitText, |
| 343 | Invocations: invocations, |
| 344 | Format: env.Format, |
| 345 | Attachments: env.Attachments, |
| 346 | AttachmentIdentities: env.AttachmentIdentities, |
| 347 | ExplicitRefs: env.ExplicitRefs, |
| 348 | Source: env.Source, |
| 349 | Extra: env.Extra, |
| 350 | } |
| 351 | data, err := json.Marshal(stable) |
| 352 | if err != nil { |
| 353 | return "", err |
| 354 | } |
| 355 | return sha256Hex(data), nil |
| 356 | } |
| 357 | |
| 358 | func normalizeEnvelope(env PromptEnvelope) PromptEnvelope { |
| 359 | env.DisplayText = strings.TrimSpace(env.DisplayText) |
| 360 | env.RawText = strings.TrimSpace(env.RawText) |
| 361 | env.SubmitText = strings.TrimSpace(env.SubmitText) |
| 362 | env.Format = strings.TrimSpace(env.Format) |
| 363 | env.Idempotency = strings.TrimSpace(env.Idempotency) |
| 364 | env.Source = strings.TrimSpace(env.Source) |
| 365 | return env |
| 366 | } |
| 367 | |
| 368 | func completeEnqueueEnvelope(env PromptEnvelope) PromptEnvelope { |
| 369 | env = normalizeEnvelope(env) |
| 370 | if env.Invocation != nil || len(env.Invocations) > 0 { |
| 371 | return env |
| 372 | } |
| 373 | if env.SubmitText == "" { |
| 374 | env.SubmitText = firstNonEmpty(env.RawText, env.DisplayText) |
| 375 | } |
| 376 | if env.DisplayText == "" { |
| 377 | env.DisplayText = env.SubmitText |
| 378 | } |
| 379 | if env.RawText == "" { |
| 380 | env.RawText = env.SubmitText |
| 381 | } |
| 382 | return env |
| 383 | } |
| 384 | |
| 385 | func refSummaries(refs []RefSnapshot) []RefSummary { |
| 386 | if len(refs) == 0 { |
| 387 | return nil |
| 388 | } |
| 389 | out := make([]RefSummary, 0, len(refs)) |
| 390 | for _, r := range refs { |
| 391 | out = append(out, RefSummary{ |
| 392 | Kind: r.Kind, |
| 393 | Path: firstNonEmpty(r.DisplayPath, r.Path), |
| 394 | Commit: r.Commit, |
| 395 | Bytes: int64(len(r.Content)), |
| 396 | Preview: PreviewText(string(r.Content), 40), |
| 397 | }) |
| 398 | } |
| 399 | return out |
| 400 | } |
| 401 | |
| 402 | func firstNonEmpty(vals ...string) string { |
| 403 | for _, v := range vals { |
| 404 | if strings.TrimSpace(v) != "" { |
| 405 | return strings.TrimSpace(v) |
| 406 | } |
| 407 | } |
| 408 | return "" |
| 409 | } |
| 410 | |
| 411 | func agentBranchID(sessionPath string) string { |
| 412 | base := filepath.Base(sessionPath) |
| 413 | return strings.TrimSuffix(base, ".jsonl") |
| 414 | } |
| 415 | |
| 416 | // RemoveDir deletes the entire inbox directory (clear/delete session). |
| 417 | func RemoveDir(sessionPath string) error { |
| 418 | dir := store.SessionInboxDir(sessionPath) |
| 419 | if dir == "" { |
| 420 | return nil |
| 421 | } |
| 422 | if err := os.RemoveAll(dir); err != nil && !os.IsNotExist(err) { |
| 423 | return err |
| 424 | } |
| 425 | return nil |
| 426 | } |
| 427 | |
| 428 | // MigrateDir renames the inbox directory with a session path change. |
| 429 | func MigrateDir(oldPath, newPath string) error { |
| 430 | oldDir := store.SessionInboxDir(oldPath) |
| 431 | newDir := store.SessionInboxDir(newPath) |
| 432 | if oldDir == "" || newDir == "" { |
| 433 | return nil |
| 434 | } |
| 435 | if err := os.Rename(oldDir, newDir); err != nil && !os.IsNotExist(err) { |
| 436 | return err |
| 437 | } |
| 438 | return nil |
| 439 | } |
| 440 |