package loop import ( "bufio" "encoding/json" "fmt" "os" "path/filepath" "sync" "time" ) // Loop event types (one JSON line per action in loop.jsonl). const ( evDispatch = "dispatch" // turn started for task_id at updated_on evReport = "report" // REPORT file written (or partial on failure) evNote = "note" // REPORT noted back on the Redmine issue evStatus = "status" // issue status transitioned per the config map evCommit = "commit" // REPORT committed to gitea (optional step) evRefresh = "refresh" // dedup marker advanced to post-writeback updated_on evError = "error" // a step failed; loop continues ) // loopEvent is one line of the append-only loop log. It doubles as the // dedup index source: dispatch/refresh lines carry (task_id, updated_on) // pairs; the latest one per task wins. type loopEvent struct { TS time.Time `json:"ts"` Type string `json:"type"` TaskID string `json:"task_id,omitempty"` UpdatedOn string `json:"updated_on,omitempty"` Subject string `json:"subject,omitempty"` Model string `json:"model,omitempty"` ReportPath string `json:"report_path,omitempty"` StopReason string `json:"stop_reason,omitempty"` StatusFrom string `json:"status_from,omitempty"` StatusTo string `json:"status_to,omitempty"` Detail string `json:"detail,omitempty"` // human context; never secrets } // loopState is the append-only loop log + dedup index: stateDir/loop.jsonl, // one event per line. The index maps task id -> the updated_on it was last // processed at; it is rebuilt from the file at open so restarts keep the // exactly-once-reaction semantics (a torn tail line is skipped, not fatal). type loopState struct { mu sync.Mutex f *os.File path string seen map[string]string readOnly bool // dry-run: dedup checks work, writes are refused } func openLoopState(stateDir string) (*loopState, error) { if err := os.MkdirAll(stateDir, 0o700); err != nil { return nil, fmt.Errorf("loop state dir: %w", err) } path := filepath.Join(stateDir, "loop.jsonl") f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) if err != nil { return nil, fmt.Errorf("open loop log: %w", err) } s := &loopState{path: path, f: f, seen: make(map[string]string)} if err := s.loadIndex(); err != nil { f.Close() return nil, err } return s, nil } // openLoopStateReadOnly builds the dedup index without creating or // touching the log (the dry-run path must leave no trace). func openLoopStateReadOnly(stateDir string) (*loopState, error) { s := &loopState{ path: filepath.Join(stateDir, "loop.jsonl"), seen: make(map[string]string), readOnly: true, } if err := s.loadIndex(); err != nil { return nil, err } return s, nil } func (s *loopState) loadIndex() error { rf, err := os.Open(s.path) if err != nil { if os.IsNotExist(err) { return nil } return fmt.Errorf("read loop log: %w", err) } defer rf.Close() sc := bufio.NewScanner(rf) sc.Buffer(make([]byte, 0, 64*1024), 1<<20) for sc.Scan() { var ev loopEvent if err := json.Unmarshal(sc.Bytes(), &ev); err != nil { continue // torn/malformed tail line; dedup stays conservative } if ev.Type == evDispatch || ev.Type == evRefresh { s.seen[ev.TaskID] = ev.UpdatedOn } } return sc.Err() } // processed reports whether taskID was already handled at this updated_on. // Empty updated_on (demo/local tasks) dedups on the id alone. func (s *loopState) processed(taskID, updatedOn string) bool { s.mu.Lock() defer s.mu.Unlock() return s.seen[taskID] == updatedOn } // log appends one event line and, for dispatch/refresh events, advances the // dedup marker. In read-only mode (dry-run) it is a no-op. func (s *loopState) log(ev loopEvent) error { if s.readOnly { return nil } ev.TS = time.Now().UTC() s.mu.Lock() defer s.mu.Unlock() line, err := json.Marshal(ev) if err != nil { return fmt.Errorf("encode loop event: %w", err) } if _, err := s.f.Write(append(line, '\n')); err != nil { return fmt.Errorf("append loop log: %w", err) } if ev.Type == evDispatch || ev.Type == evRefresh { s.seen[ev.TaskID] = ev.UpdatedOn } return nil } func (s *loopState) Close() error { s.mu.Lock() defer s.mu.Unlock() if s.f == nil { return nil } return s.f.Close() }