go-ralph-diff-driven-cursor-wiring

Answer

Gap: GetDeltas existed but only fed the display endpoint; GetCursor/UpdateCursor existed but were never called. The diff-driven loop had no durable position, so every pass re-processed all deltas and the agent had no delta context.

Fix: Introduce a cursor store + delta summary layer (internal/controller/delta.go), resolve the cursor at the top of RunLoop and scheduler_executor, inject the summary into the OpenCode prompt, and advance the cursor only after a successful commit in handleSessionComplete.

1. internal/controller/delta.go — cursor + summary primitives

package controller

import (
    "context"
    "fmt"
    "sort"
    "strings"
)

type DeltaStatus string

const (
    StatusPending DeltaStatus = "pending"
    StatusApplied DeltaStatus = "applied"
    StatusFailed  DeltaStatus = "failed"
    StatusSkipped DeltaStatus = "skipped"
)

// Delta is a single change-set produced by GetDeltas.
type Delta struct {
    ID     string            `json:"id"`
    Commit string            `json:"commit,omitempty"`
    Offset int64             `json:"offset"`
    Files  []string          `json:"files"`
    Status DeltaStatus       `json:"status"`
    Meta   map[string]string `json:"meta,omitempty"`
}

// DeltaCursor is the durable position of the loop. Deltas with
// From <= Offset < To are in scope for the next agent pass.
type DeltaCursor struct {
    From              int64  `json:"from"`
    To                int64  `json:"to"`
    LastAppliedCommit string `json:"lastAppliedCommit,omitempty"`
}

func (c DeltaCursor) IsZero() bool {
    return c.From == 0 && c.To == 0 && c.LastAppliedCommit == ""
}

// CursorStore is shared by RunLoop and scheduler_executor so
// scheduled work resumes exactly where the loop stopped.
type CursorStore interface {
    GetCursor(ctx context.Context) (DeltaCursor, error)
    UpdateCursor(ctx context.Context, c DeltaCursor) error
}

// ScopePaths are the read/write roots from IOConfig that constrain
// which deltas are relevant to the agent.
type ScopePaths struct {
    Reads  []string `json:"reads"`
    Writes []string `json:"writes"`
}

func (s ScopePaths) String() string {
    parts := make([]string, 0, len(s.Reads)+len(s.Writes))
    for _, p := range s.Reads {
        parts = append(parts, "read:"+p)
    }
    for _, p := range s.Writes {
        parts = append(parts, "write:"+p)
    }
    sort.Strings(parts)
    return strings.Join(parts, ",")
}

// DeltaSummary is injected verbatim into the OpenCode agent prompt.
type DeltaSummary struct {
    Cursor   DeltaCursor            `json:"cursor"`
    Files    []string               `json:"files"`
    Statuses map[DeltaStatus]int    `json:"statuses"`
    Scope    ScopePaths             `json:"scope"`
    DeltaIDs []string               `json:"deltaIds"`
}

func (s DeltaSummary) Render() string {
    var b strings.Builder
    fmt.Fprintf(&b, "## Delta window\n")
    fmt.Fprintf(&b, "cursor range: [%d, %d)\n", s.Cursor.From, s.Cursor.To)
    fmt.Fprintf(&b, "files (%d): %s\n", len(s.Files), strings.Join(s.Files, ", "))
    fmt.Fprintf(&b, "statuses: pending=%d applied=%d failed=%d skipped=%d\n",
        s.Statuses[StatusPending], s.Statuses[StatusApplied],
        s.Statuses[StatusFailed], s.Statuses[StatusSkipped])
    fmt.Fprintf(&b, "scope paths: %s\n", s.Scope.String())
    return b.String()
}

// ResolveCursor reads the persisted cursor and clamps it to the
// available delta offsets. First run (zero cursor) starts at 0.
func ResolveCursor(ctx context.Context, store CursorStore, deltas []Delta) (DeltaCursor, error) {
    c, err := store.GetCursor(ctx)
    if err != nil {
        return DeltaCursor{}, fmt.Errorf("resolve cursor: %w", err)
    }
    cur := c
    if cur.IsZero() {
        cur.From, cur.To = 0, 0
    }
    // Never leave the valid offset space.
    if n := int64(len(deltas)); cur.To > n {
        cur.To = n
    }
    if cur.From > cur.To {
        cur.From = cur.To
    }
    return cur, nil
}

// BuildSummary converts the delta window [c.From, c.To) plus IOConfig
// scope into the agent-facing summary. Deltas outside the window are
// excluded so the prompt only ever sees in-scope work.
func BuildSummary(cursor DeltaCursor, deltas []Delta, scope ScopePaths) DeltaSummary {
    s := DeltaSummary{
        Cursor:   cursor,
        Statuses: make(map[DeltaStatus]int),
        Scope:    scope,
    }
    seen := make(map[string]struct{})
    for _, d := range deltas {
        if d.Offset < cursor.From || d.Offset >= cursor.To {
            continue // out of window
        }
        s.DeltaIDs = append(s.DeltaIDs, d.ID)
        s.Statuses[d.Status]++
        for _, f := range d.Files {
            if _, ok := seen[f]; !ok {
                seen[f] = struct{}{}
                s.Files = append(s.Files, f)
            }
        }
    }
    sort.Strings(s.Files)
    return s
}

// AdvanceCursor persists the new position. Called only after a
// successful commit so failed windows are retried on the next pass.
func AdvanceCursor(ctx context.Context, store CursorStore, c DeltaCursor) error {
    if err := store.UpdateCursor(ctx, c); err != nil {
        return fmt.Errorf("advance cursor: %w", err)
    }
    return nil
}

2. RunLoop wiring — resolve → summarize → inject → advance

func (l *Loop) RunLoop(ctx context.Context, cfg IOConfig) error {
    deltas, err := GetDeltas(ctx, l.deltaSource)
    if err != nil {
        return fmt.Errorf("get deltas: %w", err)
    }

    // Previously: display-only. Now the driver of the whole pass.
    cursor, err := ResolveCursor(ctx, l.store, deltas)
    if err != nil {
        return err
    }

    scope := ScopePaths{Reads: cfg.Reads, Writes: cfg.Writes}
    summary := BuildSummary(cursor, deltas, scope)

    // Inject the delta summary into the agent prompt.
    prompt := l.tmpl.Render(map[string]any{
        "deltas":      summary.Render(),
        "scope":       scope.String(),
        "deltaCount":  len(summary.DeltaIDs),
    })

    result, err := l.agent.Run(ctx, prompt)
    if err != nil {
        l.log.Warn("agent pass failed; cursor held for retry", "err", err)
        return err
    }

    // Successful commit -> persist the advanced cursor.
    if err := l.handleSessionComplete(ctx, result, summary); err != nil {
        return err
    }
    return nil
}

3. scheduler_executor wiring — resume from the same cursor

func (s *Scheduler) Execute(ctx context.Context, store CursorStore) error {
    // Reuse the exact same resolution path as RunLoop: scheduled work
    // starts at the persisted cursor and never reprocesses committed deltas.
    deltas, err := GetDeltas(ctx, s.deltaSource)
    if err != nil {
        return err
    }
    cursor, err := ResolveCursor(ctx, store, deltas)
    if err != nil {
        return err
    }
    tasks := s.index.Scan(cursor.From, cursor.To) // offset-scoped task window
    return s.runTasks(ctx, tasks, cursor)
}

4. handleSessionComplete — advance only on success

func (l *Loop) handleSessionComplete(ctx context.Context, result AgentResult, s DeltaSummary) error {
    if !result.CommitSucceeded() {
        // Keep the cursor: the failed window is retried idempotently.
        l.log.Info("no commit; cursor unchanged", "range", s.Cursor)
        return nil
    }
    next := DeltaCursor{
        From:              s.Cursor.To,               // move past everything handled
        To:                s.Cursor.To,               // next pass resolves the new window
        LastAppliedCommit: result.CommitHash,
    }
    if err := AdvanceCursor(ctx, l.store, next); err != nil {
        return err
    }
    l.log.Info("cursor advanced", "from", s.Cursor.From, "to", next.From,
        "commit", next.LastAppliedCommit)
    return nil
}

5. Prompt template addition

You are operating on a diff-driven loop.
{deltas}
Scope is limited to the read/write paths above; do not touch files outside them.

({deltas} renders DeltaSummary.Render(): cursor range, deduped file list, per-status counts, and IOConfig scope paths.)