harness loop: self-hosting daemon — Redmine SoR drives its own turns
`harness loop` polls the /issues.json intake on an interval (default
120s, --once for cron-style single scans), runs one bounded turn per
new/updated issue — deduped by issue id + updated_on in an append-only
loop.jsonl — then writes the REPORT back as a Redmine journal note,
transitions status per [redmine.status_map] (In Progress -> Done by
name, resolved via /issue_statuses.json), and refreshes the dedup marker
to the post-writeback updated_on so its own notes never re-trigger it.
Turns are sequential (v0); failed turns are recorded, not retried, so a
down proxy cannot hot-loop the poll. The optional Gitea REPORT commit
([gitea] commit_reports, off) rides the same flow. No slot files, no
doorbell screens, no queue scripts — the bash middle layer is replaced,
not wrapped.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
This commit is contained in:
@@ -0,0 +1,250 @@
|
||||
// Self-host daemon: `harness loop` polls the Redmine intake on an interval,
|
||||
// runs ONE bounded turn per new/updated issue (sequentially, v0), notes the
|
||||
// REPORT back on the issue, transitions status per the config map, and
|
||||
// optionally commits the REPORT to gitea. Redmine is the SoR; this loop is
|
||||
// the worker. It replaces the crossfeed bash stack: no slot files, no
|
||||
// doorbell screens, no queue scripts.
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"ukrrs.com/mopac/harness/internal/models"
|
||||
"ukrrs.com/mopac/harness/internal/task"
|
||||
"ukrrs.com/mopac/harness/internal/writeback"
|
||||
)
|
||||
|
||||
// LoopOpts controls the daemon.
|
||||
type LoopOpts struct {
|
||||
Interval time.Duration // poll interval; 0 = [loop] poll_interval_secs
|
||||
Once bool // single scan then return (cron-able)
|
||||
DryRun bool // scan + print what would dispatch; no turns, no state writes
|
||||
}
|
||||
|
||||
// RunLoop is the daemon body. It returns nil on context cancel (SIGINT),
|
||||
// and a non-nil error only for startup/config failures.
|
||||
func (c *Conductor) RunLoop(ctx context.Context, opts LoopOpts) error {
|
||||
rc := c.cfg.Redmine
|
||||
if rc.URL == "" {
|
||||
return fmt.Errorf("loop: [redmine] url is required (the loop is Redmine-driven; use `harness once --demo` for the demo issue)")
|
||||
}
|
||||
|
||||
interval := opts.Interval
|
||||
if interval <= 0 {
|
||||
interval = time.Duration(c.cfg.Loop.PollIntervalSecs) * time.Second
|
||||
}
|
||||
|
||||
var state *loopState
|
||||
var err error
|
||||
if opts.DryRun {
|
||||
// Dry-run: dedup awareness without leaving any trace on disk.
|
||||
state, err = openLoopStateReadOnly(c.cfg.Loop.StateDir)
|
||||
} else {
|
||||
state, err = openLoopState(c.cfg.Loop.StateDir)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer state.Close()
|
||||
|
||||
key, err := c.keys.Resolve(ctx, rc.KeyRef)
|
||||
if err != nil {
|
||||
return fmt.Errorf("loop: redmine key: %w", err)
|
||||
}
|
||||
writer := writeback.NewRedmineWriter(rc, key)
|
||||
|
||||
var gitea *writeback.GiteaWriter
|
||||
if c.cfg.Gitea.CommitReports {
|
||||
gkey, err := c.keys.Resolve(ctx, c.cfg.Gitea.KeyRef)
|
||||
if err != nil {
|
||||
return fmt.Errorf("loop: gitea key: %w", err)
|
||||
}
|
||||
gitea = writeback.NewGiteaWriter(c.cfg.Gitea, gkey)
|
||||
}
|
||||
|
||||
fmt.Fprintf(c.out, "harness: loop: vertical=%s poll=%s state=%s redmine=%s (SIGINT to stop)\n",
|
||||
c.cfg.Vertical, interval, c.cfg.Loop.StateDir, c.redmineHost())
|
||||
if gitea != nil {
|
||||
fmt.Fprintf(c.out, "harness: loop: gitea report commit on (%s/%s)\n", c.cfg.Gitea.Owner, c.cfg.Gitea.Repo)
|
||||
}
|
||||
|
||||
for {
|
||||
if err := c.scanOnce(ctx, state, writer, gitea, opts.DryRun); err != nil {
|
||||
// Scan failures (redmine down, decode hiccups) are logged and
|
||||
// retried on the next tick; only ctx cancel ends the daemon.
|
||||
fmt.Fprintf(c.out, "harness: loop: scan error: %v (retrying next interval)\n", err)
|
||||
_ = state.log(loopEvent{Type: evError, Detail: "scan: " + err.Error()})
|
||||
}
|
||||
if opts.Once {
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Fprintf(c.out, "harness: loop: stopped\n")
|
||||
return nil
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// scanOnce runs one intake scan: every task whose (id, updated_on) has not
|
||||
// been processed gets one bounded turn + writebacks, strictly sequentially.
|
||||
func (c *Conductor) scanOnce(ctx context.Context, state *loopState, writer *writeback.RedmineWriter, gitea *writeback.GiteaWriter, dryRun bool) error {
|
||||
tasks, err := c.intake(ctx, OnceOpts{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fresh := 0
|
||||
for _, t := range tasks {
|
||||
if !state.processed(t.ID, t.UpdatedOn) {
|
||||
fresh++
|
||||
}
|
||||
}
|
||||
fmt.Fprintf(c.out, "harness: loop: scan: %d task(s) in scope, %d new/updated\n", len(tasks), fresh)
|
||||
|
||||
for _, t := range tasks {
|
||||
if state.processed(t.ID, t.UpdatedOn) {
|
||||
continue
|
||||
}
|
||||
if dryRun {
|
||||
fmt.Fprintf(c.out, "harness: loop: would dispatch %s (updated %s): %q\n", t.ID, t.UpdatedOn, t.Subject)
|
||||
continue
|
||||
}
|
||||
c.dispatchTask(ctx, state, writer, gitea, t)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// dispatchTask runs one turn end-to-end for t: mark -> turn -> REPORT ->
|
||||
// note -> status -> gitea -> refresh. Every step logs one stdout line plus
|
||||
// one JSONL event; a failed step never kills the loop.
|
||||
func (c *Conductor) dispatchTask(ctx context.Context, state *loopState, writer *writeback.RedmineWriter, gitea *writeback.GiteaWriter, t task.Task) {
|
||||
// Mark BEFORE the turn: a failing turn must not hot-loop the poll.
|
||||
_ = state.log(loopEvent{Type: evDispatch, TaskID: t.ID, UpdatedOn: t.UpdatedOn, Subject: t.Subject})
|
||||
fmt.Fprintf(c.out, "harness: loop: dispatch %s (updated %s): %q\n", t.ID, t.UpdatedOn, t.Subject)
|
||||
|
||||
run, err := c.runTask(ctx, t)
|
||||
if run != nil && run.turn != nil && run.reportPath != "" {
|
||||
_ = state.log(loopEvent{
|
||||
Type: evReport, TaskID: t.ID, Model: run.decision.Model,
|
||||
ReportPath: run.reportPath, StopReason: run.turn.StopReason,
|
||||
})
|
||||
}
|
||||
if err != nil {
|
||||
_ = state.log(loopEvent{Type: evError, TaskID: t.ID, Detail: "turn: " + err.Error()})
|
||||
fmt.Fprintf(c.out, "harness: loop: task %s turn failed: %v (issue left for PMO; update the issue to retry)\n", t.ID, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Gitea commit (optional): the REPORT file into the repo.
|
||||
if gitea != nil && run.reportPath != "" {
|
||||
if err := gitea.CommitFile(ctx, repoPath(run.reportPath), run.reportBody,
|
||||
fmt.Sprintf("harness: REPORT for %s task %s (%s)", c.cfg.Vertical, t.ID, run.decision.Model)); err != nil {
|
||||
_ = state.log(loopEvent{Type: evError, TaskID: t.ID, Detail: "gitea commit: " + err.Error()})
|
||||
fmt.Fprintf(c.out, "harness: loop: gitea commit failed for %s: %v\n", t.ID, err)
|
||||
} else {
|
||||
_ = state.log(loopEvent{Type: evCommit, TaskID: t.ID, ReportPath: run.reportPath})
|
||||
fmt.Fprintf(c.out, "harness: loop: gitea: committed %s\n", repoPath(run.reportPath))
|
||||
}
|
||||
}
|
||||
|
||||
// SoR writeback: note the REPORT on the issue, then transition status.
|
||||
if to, ok := c.cfg.Redmine.StatusMap[t.Status]; ok {
|
||||
if err := writer.SetStatus(ctx, t.ID, to); err != nil {
|
||||
_ = state.log(loopEvent{Type: evError, TaskID: t.ID, Detail: "status: " + err.Error()})
|
||||
fmt.Fprintf(c.out, "harness: loop: status transition failed for %s: %v\n", t.ID, err)
|
||||
} else {
|
||||
_ = state.log(loopEvent{Type: evStatus, TaskID: t.ID, StatusFrom: t.Status, StatusTo: to})
|
||||
fmt.Fprintf(c.out, "harness: loop: status %s -> %s on #%s\n", t.Status, to, t.ID)
|
||||
}
|
||||
}
|
||||
if err := writer.AddNote(ctx, t.ID, run.reportBody); err != nil {
|
||||
_ = state.log(loopEvent{Type: evError, TaskID: t.ID, Detail: "note: " + err.Error()})
|
||||
fmt.Fprintf(c.out, "harness: loop: note writeback failed for %s: %v\n", t.ID, err)
|
||||
} else {
|
||||
_ = state.log(loopEvent{Type: evNote, TaskID: t.ID, ReportPath: run.reportPath})
|
||||
fmt.Fprintf(c.out, "harness: loop: noted REPORT on #%s\n", t.ID)
|
||||
}
|
||||
|
||||
// Our own writebacks bump updated_on; refresh the dedup marker to the
|
||||
// post-writeback value so the next scan does not re-trigger itself.
|
||||
updated, err := writer.UpdatedOn(ctx, t.ID)
|
||||
if err != nil {
|
||||
_ = state.log(loopEvent{Type: evError, TaskID: t.ID, Detail: "refresh: " + err.Error()})
|
||||
fmt.Fprintf(c.out, "harness: loop: refresh failed for %s: %v (a re-dispatch may follow)\n", t.ID, err)
|
||||
return
|
||||
}
|
||||
_ = state.log(loopEvent{Type: evRefresh, TaskID: t.ID, UpdatedOn: updated})
|
||||
}
|
||||
|
||||
// taskRun is one completed (or failed) turn with everything the writebacks
|
||||
// need: the routing decision, the turn result, and the REPORT file path +
|
||||
// rendered body.
|
||||
type taskRun struct {
|
||||
task task.Task
|
||||
decision models.Decision
|
||||
turn *TurnResult
|
||||
reportPath string
|
||||
reportBody string
|
||||
}
|
||||
|
||||
// runTask is the shared per-task core of `once` and `loop`: routing ->
|
||||
// bounded turn -> REPORT file. The returned error is the turn error (a
|
||||
// partial REPORT may still exist); setup failures (routing, keys) return a
|
||||
// nil taskRun.
|
||||
func (c *Conductor) runTask(ctx context.Context, t task.Task) (*taskRun, error) {
|
||||
decision, err := c.router.Resolve(t.Class)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fmt.Fprintf(c.out, "harness: task %s (%s): %q class=%s -> %s -> %s\n",
|
||||
t.ID, t.Source, t.Subject, t.Class, decision.Tier, decision.Model)
|
||||
|
||||
client, err := c.llmClient()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
start := time.Now()
|
||||
turn, err := c.turn(ctx, client, t, decision.Model)
|
||||
run := &taskRun{task: t, decision: decision, turn: turn}
|
||||
if err != nil {
|
||||
if turn != nil && turn.Content != "" {
|
||||
path, body, werr := c.writeReport(t, decision, turn, start, err)
|
||||
if werr == nil {
|
||||
run.reportPath, run.reportBody = path, body
|
||||
}
|
||||
}
|
||||
return run, fmt.Errorf("%w: %v", ErrLLM, err)
|
||||
}
|
||||
path, body, werr := c.writeReport(t, decision, turn, start, nil)
|
||||
if werr != nil {
|
||||
return run, werr
|
||||
}
|
||||
run.reportPath, run.reportBody = path, body
|
||||
return run, nil
|
||||
}
|
||||
|
||||
// repoPath maps a local REPORT path to its in-repo path (slash-separated,
|
||||
// relative to the process CWD when possible, else the base name).
|
||||
func repoPath(local string) string {
|
||||
cwd, err := os.Getwd()
|
||||
if err == nil {
|
||||
if rel, err := filepath.Rel(cwd, local); err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
|
||||
return filepath.ToSlash(rel)
|
||||
}
|
||||
}
|
||||
return filepath.ToSlash(filepath.Base(local))
|
||||
}
|
||||
|
||||
func (c *Conductor) redmineHost() string {
|
||||
u := c.cfg.Redmine.URL
|
||||
if i := strings.Index(u, "://"); i >= 0 {
|
||||
u = u[i+3:]
|
||||
}
|
||||
return strings.TrimSuffix(u, "/")
|
||||
}
|
||||
@@ -0,0 +1,415 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"ukrrs.com/mopac/harness/internal/config"
|
||||
)
|
||||
|
||||
// fakeRedmine is a mutable in-memory Redmine: /issues.json scope listing,
|
||||
// PUT /issues/{id}.json (note + status), /issue_statuses.json, and
|
||||
// GET /issues/{id}.json (updated_on). Notes bump updated_on like the real
|
||||
// SoR, so the loop's refresh logic is exercised for real.
|
||||
type fakeRedmine struct {
|
||||
mu sync.Mutex
|
||||
issues map[string]struct {
|
||||
subject string
|
||||
status string
|
||||
updatedOn string
|
||||
notes []string
|
||||
}
|
||||
statusIDs map[string]int
|
||||
noted []string // issue ids that got a note, in order
|
||||
srv *httptest.Server
|
||||
}
|
||||
|
||||
func newFakeRedmine(t *testing.T) *fakeRedmine {
|
||||
t.Helper()
|
||||
f := &fakeRedmine{
|
||||
issues: map[string]struct {
|
||||
subject string
|
||||
status string
|
||||
updatedOn string
|
||||
notes []string
|
||||
}{},
|
||||
statusIDs: map[string]int{"New": 1, "In Progress": 2, "Done": 3},
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("GET /issues.json", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
var b strings.Builder
|
||||
b.WriteString(`{"issues":[`)
|
||||
first := true
|
||||
for id, is := range f.issues {
|
||||
if !first {
|
||||
b.WriteString(",")
|
||||
}
|
||||
first = false
|
||||
fmt.Fprintf(&b, `{"id":%s,"subject":%q,"description":"do the thing","updated_on":%q,"status":{"name":%q},"custom_fields":[{"name":"Class","value":"primary"}]}`,
|
||||
id, is.subject, is.updatedOn, is.status)
|
||||
}
|
||||
b.WriteString(`]}`)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Write([]byte(b.String()))
|
||||
})
|
||||
mux.HandleFunc("PUT /issues/", func(w http.ResponseWriter, r *http.Request) {
|
||||
id := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/issues/"), ".json")
|
||||
var body struct {
|
||||
Issue struct {
|
||||
Notes string `json:"notes"`
|
||||
StatusID int `json:"status_id"`
|
||||
} `json:"issue"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
http.Error(w, "bad json", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
is, ok := f.issues[id]
|
||||
if !ok {
|
||||
http.Error(w, "not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
if body.Issue.Notes != "" {
|
||||
is.notes = append(is.notes, body.Issue.Notes)
|
||||
f.noted = append(f.noted, id)
|
||||
}
|
||||
if body.Issue.StatusID != 0 {
|
||||
for name, sid := range f.statusIDs {
|
||||
if sid == body.Issue.StatusID {
|
||||
is.status = name
|
||||
}
|
||||
}
|
||||
}
|
||||
is.updatedOn = time.Now().UTC().Format(time.RFC3339)
|
||||
f.issues[id] = is
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
})
|
||||
mux.HandleFunc("GET /issue_statuses.json", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
var b strings.Builder
|
||||
b.WriteString(`{"issue_statuses":[`)
|
||||
first := true
|
||||
for name, id := range f.statusIDs {
|
||||
if !first {
|
||||
b.WriteString(",")
|
||||
}
|
||||
first = false
|
||||
fmt.Fprintf(&b, `{"id":%d,"name":%q}`, id, name)
|
||||
}
|
||||
b.WriteString(`]}`)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Write([]byte(b.String()))
|
||||
})
|
||||
mux.HandleFunc("GET /issues/", func(w http.ResponseWriter, r *http.Request) {
|
||||
id := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/issues/"), ".json")
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
is, ok := f.issues[id]
|
||||
if !ok {
|
||||
http.Error(w, "not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
fmt.Fprintf(w, `{"issue":{"id":%s,"updated_on":%q}}`, id, is.updatedOn)
|
||||
})
|
||||
f.srv = httptest.NewServer(mux)
|
||||
t.Cleanup(f.srv.Close)
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *fakeRedmine) addIssue(id, subject, status, updatedOn string) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.issues[id] = struct {
|
||||
subject string
|
||||
status string
|
||||
updatedOn string
|
||||
notes []string
|
||||
}{subject: subject, status: status, updatedOn: updatedOn}
|
||||
}
|
||||
|
||||
func (f *fakeRedmine) setUpdatedOn(id, updatedOn string) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if is, ok := f.issues[id]; ok {
|
||||
is.updatedOn = updatedOn
|
||||
f.issues[id] = is
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeRedmine) status(id string) string {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.issues[id].status
|
||||
}
|
||||
|
||||
func (f *fakeRedmine) noteCount(id string) int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.issues[id].notes)
|
||||
}
|
||||
|
||||
func loopTestConfig(t *testing.T, redmineURL, llmURL string) *config.Config {
|
||||
t.Helper()
|
||||
cfg := testConfig(t, llmURL)
|
||||
cfg.Redmine.URL = redmineURL
|
||||
cfg.Redmine.KeyRef = "literal:rm-key"
|
||||
cfg.Redmine.ScopeQuery = "project=mopac"
|
||||
cfg.Redmine.StatusMap = map[string]string{"In Progress": "Done"}
|
||||
cfg.Loop.StateDir = filepath.Join(t.TempDir(), "state-loop")
|
||||
cfg.Loop.PollIntervalSecs = 3600 // between-scan sleep; tests use --once scans
|
||||
return cfg
|
||||
}
|
||||
|
||||
// runOneScan runs the daemon in --once mode against the given servers.
|
||||
func runOneScan(t *testing.T, cfg *config.Config, dryRun bool) string {
|
||||
t.Helper()
|
||||
var out strings.Builder
|
||||
cond, err := New(cfg, &out)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := cond.RunLoop(context.Background(), LoopOpts{Once: true, DryRun: dryRun}); err != nil {
|
||||
t.Fatalf("RunLoop: %v (output:\n%s)", err, out.String())
|
||||
}
|
||||
return out.String()
|
||||
}
|
||||
|
||||
func TestLoopDispatchNoteStatusDedup(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "Self-host: write the note", "In Progress", "2026-08-28T21:30:00Z")
|
||||
f := newFakeLLM(t, textMsg("Report body: the turn ran."))
|
||||
cfg := loopTestConfig(t, rm.srv.URL, f.srv.URL)
|
||||
|
||||
out := runOneScan(t, cfg, false)
|
||||
|
||||
// One LLM turn, routed through the class map.
|
||||
if f.requestCount() != 1 {
|
||||
t.Fatalf("LLM calls = %d, want 1", f.requestCount())
|
||||
}
|
||||
for _, want := range []string{
|
||||
"dispatch 421",
|
||||
"noted REPORT on #421",
|
||||
"status In Progress -> Done on #421",
|
||||
} {
|
||||
if !strings.Contains(out, want) {
|
||||
t.Errorf("output missing %q:\n%s", want, out)
|
||||
}
|
||||
}
|
||||
if rm.status("421") != "Done" {
|
||||
t.Errorf("issue status = %q, want Done", rm.status("421"))
|
||||
}
|
||||
if rm.noteCount("421") != 1 {
|
||||
t.Fatalf("notes on #421 = %d, want 1", rm.noteCount("421"))
|
||||
}
|
||||
if body := rm.issues["421"].notes[0]; !strings.Contains(body, "Report body: the turn ran.") ||
|
||||
!strings.Contains(body, "# REPORT - teststack - 421") {
|
||||
t.Errorf("note body wrong:\n%s", body)
|
||||
}
|
||||
|
||||
// REPORT file landed too.
|
||||
reports, err := filepath.Glob(filepath.Join(cfg.ReportDir, "REPORT-teststack-421-*.md"))
|
||||
if err != nil || len(reports) != 1 {
|
||||
t.Fatalf("REPORT files = %v err=%v", reports, err)
|
||||
}
|
||||
|
||||
// Second scan: our own note bumped updated_on, but the refresh must
|
||||
// have advanced the dedup marker -> no re-dispatch.
|
||||
f2 := newFakeLLM(t, textMsg("must not run"))
|
||||
cfg.LiteLLM.BaseURL = f2.srv.URL
|
||||
out2 := runOneScan(t, cfg, false)
|
||||
if f2.requestCount() != 0 {
|
||||
t.Fatalf("second scan re-dispatched (LLM calls = %d)", f2.requestCount())
|
||||
}
|
||||
if !strings.Contains(out2, "0 new/updated") {
|
||||
t.Errorf("second scan should find nothing new:\n%s", out2)
|
||||
}
|
||||
if rm.noteCount("421") != 1 {
|
||||
t.Errorf("second scan added notes: %d", rm.noteCount("421"))
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopRedispatchOnIssueUpdate(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "First edit", "In Progress", "2026-08-28T21:30:00Z")
|
||||
f := newFakeLLM(t, textMsg("first reply"))
|
||||
cfg := loopTestConfig(t, rm.srv.URL, f.srv.URL)
|
||||
|
||||
runOneScan(t, cfg, false)
|
||||
if f.requestCount() != 1 {
|
||||
t.Fatalf("first scan LLM calls = %d", f.requestCount())
|
||||
}
|
||||
|
||||
// PMO updates the issue (bumps updated_on): the loop re-releases it.
|
||||
rm.setUpdatedOn("421", "2026-08-28T22:30:00Z")
|
||||
out := runOneScan(t, cfg, false)
|
||||
if !strings.Contains(out, "1 new/updated") {
|
||||
t.Errorf("updated issue should be new:\n%s", out)
|
||||
}
|
||||
if f.requestCount() != 2 {
|
||||
t.Errorf("LLM calls after update = %d, want 2", f.requestCount())
|
||||
}
|
||||
if rm.noteCount("421") != 2 {
|
||||
t.Errorf("notes after update = %d, want 2", rm.noteCount("421"))
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopTurnFailureRecordedNoHotLoop(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "Broken task", "In Progress", "2026-08-28T21:30:00Z")
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "proxy down", http.StatusBadGateway)
|
||||
}))
|
||||
defer srv.Close()
|
||||
cfg := loopTestConfig(t, rm.srv.URL, srv.URL)
|
||||
|
||||
out := runOneScan(t, cfg, false)
|
||||
if !strings.Contains(out, "turn failed") {
|
||||
t.Errorf("output should record the failed turn:\n%s", out)
|
||||
}
|
||||
// No note, no status change on a failed turn...
|
||||
if rm.noteCount("421") != 0 || rm.status("421") != "In Progress" {
|
||||
t.Errorf("failed turn must not note/status: notes=%d status=%s", rm.noteCount("421"), rm.status("421"))
|
||||
}
|
||||
// ...and the next scan must NOT retry it (updated_on marker consumed).
|
||||
out2 := runOneScan(t, cfg, false)
|
||||
if !strings.Contains(out2, "0 new/updated") {
|
||||
t.Errorf("failed turn hot-looped:\n%s", out2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopDryRunDispatchesNothing(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "Would-be task", "In Progress", "2026-08-28T21:30:00Z")
|
||||
f := newFakeLLM(t, textMsg("unused"))
|
||||
cfg := loopTestConfig(t, rm.srv.URL, f.srv.URL)
|
||||
|
||||
out := runOneScan(t, cfg, true)
|
||||
if f.requestCount() != 0 {
|
||||
t.Errorf("dry-run made %d LLM calls", f.requestCount())
|
||||
}
|
||||
if !strings.Contains(out, "would dispatch 421") {
|
||||
t.Errorf("dry-run should print the would-be dispatch:\n%s", out)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(cfg.Loop.StateDir, "loop.jsonl")); !os.IsNotExist(err) {
|
||||
t.Errorf("dry-run must not write loop state")
|
||||
}
|
||||
// A dry-run must not consume the task: a real scan still dispatches.
|
||||
runOneScan(t, cfg, false)
|
||||
if f.requestCount() != 1 {
|
||||
t.Errorf("LLM calls after real scan = %d, want 1", f.requestCount())
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopStateSurvivesRestart(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "Persisted", "In Progress", "2026-08-28T21:30:00Z")
|
||||
f := newFakeLLM(t, textMsg("reply"))
|
||||
cfg := loopTestConfig(t, rm.srv.URL, f.srv.URL)
|
||||
|
||||
runOneScan(t, cfg, false)
|
||||
|
||||
// New conductor = process restart: state reloads from loop.jsonl, and
|
||||
// the refresh line (post-note updated_on) is what dedups.
|
||||
f2 := newFakeLLM(t, textMsg("must not run"))
|
||||
cfg.LiteLLM.BaseURL = f2.srv.URL
|
||||
runOneScan(t, cfg, false)
|
||||
if f2.requestCount() != 0 {
|
||||
t.Errorf("restart re-dispatched a processed issue")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopStatusMapMissLeavesStatus(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "Unmapped status", "Review", "2026-08-28T21:30:00Z")
|
||||
f := newFakeLLM(t, textMsg("reply"))
|
||||
cfg := loopTestConfig(t, rm.srv.URL, f.srv.URL)
|
||||
|
||||
out := runOneScan(t, cfg, false)
|
||||
if strings.Contains(out, "status Review ->") {
|
||||
t.Errorf("unmapped status must not transition:\n%s", out)
|
||||
}
|
||||
if rm.status("421") != "Review" {
|
||||
t.Errorf("status changed: %q", rm.status("421"))
|
||||
}
|
||||
if rm.noteCount("421") != 1 {
|
||||
t.Errorf("note should still land, got %d", rm.noteCount("421"))
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopRequiresRedmineConfig(t *testing.T) {
|
||||
cfg := testConfig(t, "http://unused")
|
||||
cond, err := New(cfg, os.Stdout)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = cond.RunLoop(context.Background(), LoopOpts{Once: true})
|
||||
if err == nil || !strings.Contains(err.Error(), "[redmine] url is required") {
|
||||
t.Fatalf("err = %v, want missing-redmine error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoopGiteaCommitStep(t *testing.T) {
|
||||
rm := newFakeRedmine(t)
|
||||
rm.addIssue("421", "Commit me", "In Progress", "2026-08-28T21:30:00Z")
|
||||
f := newFakeLLM(t, textMsg("reply with body"))
|
||||
|
||||
var mu sync.Mutex
|
||||
var commits []map[string]any
|
||||
var authHeaders []string
|
||||
gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
authHeaders = append(authHeaders, r.Header.Get("Authorization"))
|
||||
if r.Method == http.MethodGet {
|
||||
http.Error(w, `{"message":"Not Found"}`, http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
var body map[string]any
|
||||
json.NewDecoder(r.Body).Decode(&body)
|
||||
commits = append(commits, body)
|
||||
w.Write([]byte(`{"content":{"path":"x"}}`))
|
||||
}))
|
||||
defer gitea.Close()
|
||||
|
||||
cfg := loopTestConfig(t, rm.srv.URL, f.srv.URL)
|
||||
cfg.Gitea = config.GiteaConfig{
|
||||
URL: gitea.URL, KeyRef: "literal:gt-key",
|
||||
Owner: "ukrrs", Repo: "reports", Branch: "main",
|
||||
CommitReports: true,
|
||||
}
|
||||
out := runOneScan(t, cfg, false)
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(commits) != 1 {
|
||||
t.Fatalf("gitea commits = %d, want 1", len(commits))
|
||||
}
|
||||
if authHeaders[0] != "token gt-key" {
|
||||
t.Errorf("gitea auth = %q", authHeaders[0])
|
||||
}
|
||||
path, _ := commits[0]["message"].(string)
|
||||
if !strings.Contains(out, "gitea: committed") {
|
||||
t.Errorf("output missing gitea commit line:\n%s", out)
|
||||
}
|
||||
if msg, _ := commits[0]["message"].(string); !strings.Contains(msg, "task 421") {
|
||||
t.Errorf("commit message = %q", msg)
|
||||
}
|
||||
_ = path
|
||||
if commits[0]["branch"] != "main" {
|
||||
t.Errorf("commit branch = %v", commits[0]["branch"])
|
||||
}
|
||||
}
|
||||
+14
-10
@@ -1,6 +1,7 @@
|
||||
// Package loop is the conductor: one bounded iteration per call
|
||||
// (intake -> gate -> bounded turn -> REPORT writeback). Callers chain
|
||||
// iterations by re-invoking; there is no daemon.
|
||||
// (intake -> gate -> bounded turn -> REPORT writeback) plus the
|
||||
// self-hosting daemon (`harness loop`) that drives iterations off the
|
||||
// Redmine intake poll.
|
||||
package loop
|
||||
|
||||
import (
|
||||
@@ -32,6 +33,7 @@ type Conductor struct {
|
||||
cfg *config.Config
|
||||
router *models.Router
|
||||
bash *tools.BashTool
|
||||
keys *config.KeyResolver
|
||||
out io.Writer
|
||||
}
|
||||
|
||||
@@ -55,7 +57,7 @@ func New(cfg *config.Config, out io.Writer) (*Conductor, error) {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &Conductor{cfg: cfg, router: router, bash: bash, out: out}, nil
|
||||
return &Conductor{cfg: cfg, router: router, bash: bash, keys: config.NewKeyResolver(cfg), out: out}, nil
|
||||
}
|
||||
|
||||
// OnceOpts controls a single iteration.
|
||||
@@ -127,11 +129,11 @@ func (c *Conductor) Once(ctx context.Context, opts OnceOpts) (*OnceResult, error
|
||||
if err != nil {
|
||||
// Keep whatever content the turn produced before failing.
|
||||
if turn != nil && turn.Content != "" {
|
||||
_, _ = c.writeReport(t, decision, turn, start, err)
|
||||
_, _, _ = c.writeReport(t, decision, turn, start, err)
|
||||
}
|
||||
return res, fmt.Errorf("%w: %v", ErrLLM, err)
|
||||
}
|
||||
path, err := c.writeReport(t, decision, turn, start, nil)
|
||||
path, _, err := c.writeReport(t, decision, turn, start, nil)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
@@ -227,7 +229,7 @@ func (c *Conductor) intake(ctx context.Context, opts OnceOpts) ([]task.Task, err
|
||||
if rc.URL == "" {
|
||||
return nil, fmt.Errorf("%w: no [redmine] url configured (use --demo for the demo issue)", ErrIntake)
|
||||
}
|
||||
key, err := config.ResolveKeyRef(rc.KeyRef)
|
||||
key, err := c.keys.Resolve(ctx, rc.KeyRef)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: redmine key: %v", ErrIntake, err)
|
||||
}
|
||||
@@ -241,7 +243,7 @@ func (c *Conductor) intake(ctx context.Context, opts OnceOpts) ([]task.Task, err
|
||||
}
|
||||
|
||||
func (c *Conductor) llmClient() (*llm.Client, error) {
|
||||
key, err := config.ResolveKeyRef(c.cfg.LiteLLM.KeyRef)
|
||||
key, err := c.keys.Resolve(context.Background(), c.cfg.LiteLLM.KeyRef)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("litellm key: %w", err)
|
||||
}
|
||||
@@ -253,7 +255,9 @@ func (c *Conductor) llmClient() (*llm.Client, error) {
|
||||
), nil
|
||||
}
|
||||
|
||||
func (c *Conductor) writeReport(t task.Task, d models.Decision, turn *TurnResult, start time.Time, turnErr error) (string, error) {
|
||||
// writeReport persists the turn as a REPORT file and returns its path plus
|
||||
// the rendered body (the loop re-uses the body for the Redmine note).
|
||||
func (c *Conductor) writeReport(t task.Task, d models.Decision, turn *TurnResult, start time.Time, turnErr error) (string, string, error) {
|
||||
r := writeback.Report{
|
||||
Time: time.Now().UTC(),
|
||||
Vertical: c.cfg.Vertical,
|
||||
@@ -276,10 +280,10 @@ func (c *Conductor) writeReport(t task.Task, d models.Decision, turn *TurnResult
|
||||
}
|
||||
path, err := writeback.Write(c.cfg.ReportDir, r)
|
||||
if err != nil {
|
||||
return "", err
|
||||
return "", "", err
|
||||
}
|
||||
fmt.Fprintf(c.out, "harness: REPORT %s\n", path)
|
||||
return path, nil
|
||||
return path, r.Render(), nil
|
||||
}
|
||||
|
||||
func (c *Conductor) printPlan(t task.Task, d models.Decision) {
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
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()
|
||||
}
|
||||
Reference in New Issue
Block a user