diff --git a/CLAUDE.md b/CLAUDE.md index 3e8308fa9..60f58a824 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -90,6 +90,7 @@ REST API at `/api/v1/`. Key endpoints: - `POST /workspaces/{ws}/items/{slug}/copy` — cross-workspace copy; with `archive_source` it is the move. Same request shape as the preflight. **Never retry it automatically** — there is no idempotency key, so a retry duplicates the item - `GET/POST /workspaces/{ws}/items/{slug}/reminders` — item reminders (IDEA-2641). POST arms one; `remind_at` is an RFC3339 **instant** and a bare date is refused, not read as midnight - `PATCH/DELETE /workspaces/{ws}/reminders/{id}`, `POST /workspaces/{ws}/reminders/{id}/ack` — re-arm (clears both fire marks), disarm, acknowledge. Permission is the ITEM's; the reminder has no separate owner +- `GET /workspaces/{ws}/items/{slug}/decisions` — the item's latest typed-decision answer per question (PLAN-3114 / TASK-3117), each with `current: true` only when it was computed from the item's PRESENT state (a sha256 of the exact state bytes sent to the provider — title, collection, fields, bounded body, last 10 comments) AND for the question as registered now under the model pinned now — rewording a question or changing the model re-asks. Evaluation is asynchronous: item create / update / move and comment create enqueue a coalesced `decision_jobs` row in their own transaction and a server tick does the provider call, so nothing on a write path waits on the network. System fan-out rewrites (title-rename cascade, field migration, attachment remap) deliberately do NOT enqueue — their items read `current: false` until the next direct write — and neither does workspace import, whose items have no decisions until their next direct write. With no provider configured the list is empty and no job row is ever written. The single-item GET (and `pad item show --format json` / `--agent`) carries the same list as `decisions` when non-empty - `GET /workspaces/{ws}/dashboard` — computed project overview (active items, plans, attention, blockers). Also carries `pending_reminders`: fired-but-unacknowledged reminders, which is the delivery path on any instance with no webhook configured - `GET /workspaces/{ws}/activity` — workspace activity feed (enriched with item titles + change details) - `GET/POST/DELETE /workspaces/{ws}/webhooks` — webhook management diff --git a/cmd/pad/cmd_server.go b/cmd/pad/cmd_server.go index 9597bbf66..7bd069b29 100644 --- a/cmd/pad/cmd_server.go +++ b/cmd/pad/cmd_server.go @@ -23,6 +23,7 @@ import ( "github.com/PerpetualSoftware/pad/internal/cli" "github.com/PerpetualSoftware/pad/internal/cmdhelp" "github.com/PerpetualSoftware/pad/internal/config" + "github.com/PerpetualSoftware/pad/internal/decision" "github.com/PerpetualSoftware/pad/internal/billing" "github.com/PerpetualSoftware/pad/internal/collab" @@ -886,6 +887,20 @@ func serveCmd() *cobra.Command { } srv.StartReminderTick() + // Typed decisions (PLAN-3114 / TASK-3117). With no provider + // configured the runner is nil: no write enqueues a job and no + // tick starts. A provider that is NAMED but cannot be built (an + // unknown name, a missing key) is logged at ERROR and the server + // runs without decisions — an operator asked for them and is not + // getting them, which must be visible, but it is not a reason to + // refuse to serve everything else. + decisionProvider, derr := decision.New(resolveDecisionConfig(cfg)) + if derr != nil { + slog.Error("decision provider misconfigured; decisions are OFF", "error", derr) + } + srv.SetDecisionRunner(decision.NewRunner(s, decisionProvider, decision.NewRegistry())) + srv.StartDecisionTick() + // Workspace hard-purge sweeper (TASK-1966). Periodic sweep // that hard-deletes workspaces soft-deleted more than 30 days // ago — cascading every child row and reclaiming attachment diff --git a/cmd/pad/server_info.go b/cmd/pad/server_info.go index 931389279..26b01caeb 100644 --- a/cmd/pad/server_info.go +++ b/cmd/pad/server_info.go @@ -124,20 +124,23 @@ does not inspect remote server internals.`, } } -// describeDecisionProvider resolves the decision provider for the report. -// -// The resolution order is the same one the server uses — config file first, -// environment last, because the environment overrides the instance-admin -// setting (ruled day 73 on PLAN-3114) — so this line cannot disagree with -// what the server would build from the same inputs. It goes through -// decision.Describe, which never returns the API key. -func describeDecisionProvider(cfg *config.Config) serverInfoDecision { - resolved := decision.Resolve(decision.Config{ +// resolveDecisionConfig is the ONE resolution of the decision provider's +// settings, shared by the server's boot path and `pad server info`: config +// file first, environment last, because the environment overrides the +// instance-admin setting (ruled day 73 on PLAN-3114). Sharing it is what makes +// the report unable to disagree with what the server builds. +func resolveDecisionConfig(cfg *config.Config) decision.Config { + return decision.Resolve(decision.Config{ Provider: cfg.DecisionProvider, APIKey: cfg.TypesafeAPIKey, Model: cfg.DecisionModel, }, decision.EnvConfig()) - name, model := decision.Describe(resolved) +} + +// describeDecisionProvider resolves the decision provider for the report. It +// goes through decision.Describe, which never returns the API key. +func describeDecisionProvider(cfg *config.Config) serverInfoDecision { + name, model := decision.Describe(resolveDecisionConfig(cfg)) return serverInfoDecision{Name: name, Model: model} } diff --git a/internal/cli/item_summary.go b/internal/cli/item_summary.go index 3d5825de2..71043370b 100644 --- a/internal/cli/item_summary.go +++ b/internal/cli/item_summary.go @@ -85,6 +85,9 @@ type ItemAgentView struct { Convention *models.ItemConventionMetadata `json:"convention,omitempty"` ImplementationNotes []models.ItemImplementationNote `json:"implementation_notes,omitempty"` DecisionLog []models.ItemDecisionLogEntry `json:"decision_log,omitempty"` + // Decisions are the typed-decision answers (TASK-3117) — agent-facing + // by purpose, so the agent projection must not be the door that drops them. + Decisions []models.ItemDecision `json:"decisions,omitempty"` } // contentPreviewLimit caps the content_preview at a small, agent-friendly @@ -182,6 +185,7 @@ func ToItemAgentView(item models.Item) ItemAgentView { Convention: item.Convention, ImplementationNotes: item.ImplementationNotes, DecisionLog: item.DecisionLog, + Decisions: item.Decisions, } } diff --git a/internal/cli/item_summary_test.go b/internal/cli/item_summary_test.go index 4373eb06a..07d39435b 100644 --- a/internal/cli/item_summary_test.go +++ b/internal/cli/item_summary_test.go @@ -176,3 +176,34 @@ func TestToItemSummaries_EmptyIsNonNil(t *testing.T) { t.Errorf("empty summaries should encode as [], got %s", b) } } + +// The agent projection is the door agents read an item through, and decisions +// exist for agents — so it must carry them (TASK-3117), and omit the member +// entirely when there are none. +func TestToItemAgentView_CarriesDecisions(t *testing.T) { + conf := 0.9 + item := models.Item{ID: "i1", Title: "t", Decisions: []models.ItemDecision{{ + QuestionSet: "triage", QuestionKey: "urgent", Kind: "choice", + Answer: json.RawMessage(`{"type":"choice","choice":"yes"}`), Confidence: &conf, Current: true, + }}} + b, err := json.Marshal(ToItemAgentView(item)) + if err != nil { + t.Fatal(err) + } + var got struct { + Decisions []models.ItemDecision `json:"decisions"` + } + if err := json.Unmarshal(b, &got); err != nil { + t.Fatal(err) + } + if len(got.Decisions) != 1 || got.Decisions[0].QuestionKey != "urgent" || !got.Decisions[0].Current { + t.Fatalf("agent view decisions = %+v (json %s)", got.Decisions, b) + } + + b, _ = json.Marshal(ToItemAgentView(models.Item{ID: "i2", Title: "t"})) + var m map[string]json.RawMessage + _ = json.Unmarshal(b, &m) + if _, has := m["decisions"]; has { + t.Fatalf("agent view carries a decisions member for an item with none: %s", b) + } +} diff --git a/internal/decision/runner.go b/internal/decision/runner.go new file mode 100644 index 000000000..ea6372c5b --- /dev/null +++ b/internal/decision/runner.go @@ -0,0 +1,600 @@ +package decision + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "sort" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// The runner: storage and evaluation of typed decisions about items +// (PLAN-3114 unit 2, TASK-3117). +// +// A write marks an evaluation owed (store.enqueueDecisionJobsTx, in the +// write's own transaction); the server's tick claims owed jobs and calls +// [Runner.Evaluate] for each. Nothing on a write path waits on the network. + +// QuestionSet is a named group of questions asked about an item in ONE +// provider call. +type QuestionSet struct { + // Name is the registry key and the stored question_set. Server-chosen, + // never caller text. + Name string + + // Collections restricts the set to items in these collection slugs. + // Empty means every collection. + Collections []string + + // Questions maps a stable question key to the question. A key is stored + // with every answer, so renaming one orphans its history. + Questions map[string]Question +} + +func (qs QuestionSet) appliesTo(collectionSlug string) bool { + if len(qs.Collections) == 0 { + return true + } + for _, c := range qs.Collections { + if c == collectionSlug { + return true + } + } + return false +} + +// Registry holds the question sets later units register. This unit ships it +// empty in production; tests register their own. +type Registry struct { + mu sync.RWMutex + sets map[string]QuestionSet +} + +// NewRegistry returns an empty registry. +func NewRegistry() *Registry { return &Registry{sets: map[string]QuestionSet{}} } + +// Register adds a set, refusing an invalid one or a duplicate name: a second +// registration under one name would silently change what stored answers mean. +func (r *Registry) Register(qs QuestionSet) error { + if qs.Name == "" { + return errors.New("decision: question set has no name") + } + if len(qs.Questions) == 0 { + return fmt.Errorf("decision: question set %q has no questions", qs.Name) + } + for key, q := range qs.Questions { + if key == "" { + return fmt.Errorf("decision: question set %q has an empty question key", qs.Name) + } + if err := q.Validate(); err != nil { + return fmt.Errorf("decision: question set %q, question %q: %w", qs.Name, key, err) + } + } + r.mu.Lock() + defer r.mu.Unlock() + if _, dup := r.sets[qs.Name]; dup { + return fmt.Errorf("decision: question set %q is already registered", qs.Name) + } + r.sets[qs.Name] = qs + return nil +} + +// Get returns a registered set. +func (r *Registry) Get(name string) (QuestionSet, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + qs, ok := r.sets[name] + return qs, ok +} + +// SetsFor names the sets that apply to a collection, sorted. It is the +// store's DecisionSetResolver, so it must stay pure (no database access): the +// store calls it inside the write transaction. +func (r *Registry) SetsFor(collectionSlug string) []string { + r.mu.RLock() + defer r.mu.RUnlock() + var out []string + for name, qs := range r.sets { + if qs.appliesTo(collectionSlug) { + out = append(out, name) + } + } + sort.Strings(out) + return out +} + +// The state bounds. They live beside the builder because they are PART OF THE +// HASH: changing one changes every item's state bytes, and so marks every +// stored answer not-current until it is re-evaluated. That is correct — the +// model would be reading different state — but it is a fleet-wide re-ask, so +// change them deliberately. +const ( + // RecentTrailWindow is how many of an item's newest comments the state + // carries. + RecentTrailWindow = 10 + + // maxStateBodyRunes bounds the item body. The provider truncates only a + // STRING state; a structured one over its budget is refused outright + // (typesafe.go fitState), so the builder bounds the body itself — and + // bounding it HERE, before hashing, is what keeps the hashed bytes + // identical to the bytes sent. + maxStateBodyRunes = 16000 + + // maxStateCommentRunes bounds each trail comment, for the same reason. + maxStateCommentRunes = 2000 +) + +// ItemState is the state a question set is asked about. Its JSON encoding is +// what the provider receives and what state_hash is computed over — one +// serialisation, see [BuildItemState]. +type ItemState struct { + Title string `json:"title"` + Collection string `json:"collection"` + Fields map[string]any `json:"fields"` + Body string `json:"body"` + Trail []ItemStateRemark `json:"recent_trail"` +} + +// ItemStateRemark is one trail comment. The timestamp is included because +// "the last comment was three weeks ago" is a fact about the item; the +// comment's id is not, since it carries no meaning a model could read. +type ItemStateRemark struct { + Author string `json:"author"` + Body string `json:"body"` + CreatedAt string `json:"created_at"` +} + +// BuiltState is a built, serialised state. +type BuiltState struct { + // Bytes is the exact JSON sent to the provider. + Bytes json.RawMessage + // Hash is hex sha256 of Bytes. + Hash string + // Truncated reports that the builder shortened the body or a comment. + Truncated bool +} + +// BuildItemState builds an item's state from its row and trail. +// +// ONE SERIALISATION PATH: the bytes returned are both hashed and sent. The +// runner passes them to the provider as json.RawMessage, which the provider's +// own json.Marshal emits verbatim (RawMessage compaction is the identity on +// json.Marshal output — pinned by a test that captures the request body). +// Hashing a separately-built value would be a second path that could drift. +// +// What is DELIBERATELY absent: seq, updated_at, sort orders, ids. None is +// something a question can read, and each changes on writes that change +// nothing else — a role-board reorder, a wiki-link re-resolution. Including +// them would make those writes cost a provider call (TASK-3117 ruling 2). +func BuildItemState(item *models.Item, comments []models.Comment) (BuiltState, error) { + fields := map[string]any{} + if item.Fields != "" { + dec := json.NewDecoder(strings.NewReader(item.Fields)) + dec.UseNumber() // a stored 3.0 must not re-encode as 3 and move the hash + if err := dec.Decode(&fields); err != nil { + return BuiltState{}, fmt.Errorf("decision state: item %s fields: %w", item.ID, err) + } + } + body, truncated := clipRunes(item.Content, maxStateBodyRunes) + st := ItemState{ + Title: item.Title, + Collection: item.CollectionSlug, + Fields: fields, + Body: body, + Trail: []ItemStateRemark{}, + } + if len(comments) > RecentTrailWindow { + comments = comments[len(comments)-RecentTrailWindow:] + } + for _, c := range comments { + cb, ct := clipRunes(c.Body, maxStateCommentRunes) + truncated = truncated || ct + st.Trail = append(st.Trail, ItemStateRemark{ + Author: c.Author, + Body: cb, + CreatedAt: c.CreatedAt.UTC().Format(time.RFC3339), + }) + } + b, err := json.Marshal(st) + if err != nil { + return BuiltState{}, fmt.Errorf("decision state: marshal item %s: %w", item.ID, err) + } + sum := sha256.Sum256(b) + return BuiltState{Bytes: b, Hash: hex.EncodeToString(sum[:]), Truncated: truncated}, nil +} + +func clipRunes(s string, n int) (string, bool) { + if utf8.RuneCountInString(s) <= n { + return s, false + } + i := 0 + for pos := range s { + if i == n { + return s[:pos], true + } + i++ + } + return s, false +} + +// QuestionFingerprint is hex sha256 of the model and one question's full +// definition. An answer is current only while its fingerprint still matches: +// rewording the instructions, changing a criterion, or re-pinning the model +// changes what was asked, and the old answer is not an answer to the new +// question (codex round 6). +// +// Marshalled from a fixed struct, so field order is the struct's and map keys +// (Choice options) are sorted by encoding/json: one question has one print. +func QuestionFingerprint(model string, q Question) string { + b, _ := json.Marshal(struct { + Model string `json:"model"` + Kind Kind `json:"kind"` + Instructions string `json:"instructions"` + Options map[string]string `json:"options,omitempty"` + Levels []string `json:"levels,omitempty"` + TrueDesc string `json:"true_desc,omitempty"` + FalseDesc string `json:"false_desc,omitempty"` + }{model, q.Kind, q.Instructions, q.Options, q.Levels, q.TrueDesc, q.FalseDesc}) + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +// Runner evaluates owed decisions. +type Runner struct { + store *store.Store + provider Provider + registry *Registry + + // beforeAsk is a TEST-ONLY seam, nil in production: it runs immediately + // before the final liveness check that precedes the provider call, so a + // test can land a delete in exactly the window that check exists for. + beforeAsk func() +} + +// NewRunner returns a runner, or nil when provider is nil — the configured-off +// case. Every method is nil-safe, so the server holds a *Runner and never +// branches on the provider itself. +func NewRunner(s *store.Store, p Provider, reg *Registry) *Runner { + if p == nil || s == nil { + return nil + } + if reg == nil { + reg = NewRegistry() + } + return &Runner{store: s, provider: p, registry: reg} +} + +// Registry returns the runner's registry (nil for a nil runner). +func (r *Runner) Registry() *Registry { + if r == nil { + return nil + } + return r.registry +} + +// Install wires the runner's registry into the store's write doors. With a +// nil runner it removes any resolver, so the doors enqueue nothing. +func (r *Runner) Install(s *store.Store) { + if r == nil { + s.SetDecisionSetResolver(nil) + return + } + s.SetDecisionSetResolver(r.registry.SetsFor) +} + +// ErrItemGone reports that the item no longer exists or is soft-deleted: the +// job is dropped rather than retried. +var ErrItemGone = errors.New("decision: item is gone") + +// ErrWorkspaceDeleted reports that the item's workspace is soft-deleted. The +// job is HELD, not dropped: a restore must find the work still owed. +var ErrWorkspaceDeleted = errors.New("decision: item's workspace is soft-deleted") + +// ErrUnknownSet reports a job for a set no longer registered: dropped. +var ErrUnknownSet = errors.New("decision: question set is not registered") + +// ErrSetNotApplicable reports a job for a set that does not cover the item's +// CURRENT collection — the item moved after the job was owed: dropped. +var ErrSetNotApplicable = errors.New("decision: question set does not apply to the item's collection") + +// State builds the item's current state. +func (r *Runner) State(itemID string) (*models.Item, BuiltState, error) { + item, err := r.store.GetItem(itemID) + if err != nil { + return nil, BuiltState{}, fmt.Errorf("decision: read item %s: %w", itemID, err) + } + if item == nil || item.DeletedAt != nil { + return nil, BuiltState{}, ErrItemGone + } + comments, err := r.store.RecentComments(itemID, RecentTrailWindow) + if err != nil { + return nil, BuiltState{}, err + } + st, err := BuildItemState(item, comments) + return item, st, err +} + +// Evaluate asks the named set about the item, unless answers for its current +// state are already stored. It reports whether the provider was called. +func (r *Runner) Evaluate(ctx context.Context, itemID, setName string) (bool, error) { + if r == nil { + return false, nil + } + qs, ok := r.registry.Get(setName) + if !ok { + return false, ErrUnknownSet + } + item, st, err := r.State(itemID) + if err != nil { + return false, err + } + // Applicability is re-checked HERE, against the collection the item is in + // now, not trusted from enqueue time: a job owed before a move still names + // the old collection's set (codex round 2). Checked before the idempotency + // read so an inapplicable set never costs a call or a query. + if !qs.appliesTo(item.CollectionSlug) { + return false, ErrSetNotApplicable + } + keys := make([]string, 0, len(qs.Questions)) + qhash := make(map[string]string, len(qs.Questions)) + for k, q := range qs.Questions { + keys = append(keys, k) + qhash[k] = QuestionFingerprint(r.provider.Model(), q) + } + sort.Strings(keys) + have, err := r.store.HasItemDecisionsAtState(item.ID, qs.Name, st.Hash, qhash) + if err != nil { + return false, err + } + if have { + return false, nil + } + + if r.beforeAsk != nil { + r.beforeAsk() + } + // Liveness again, as the LAST statement before the send (codex round 4): + // the check in State ran before the idempotency read, and a delete landing + // in between must still stop the call. RESIDUAL, stated rather than + // implied: no fence can hold across a network request, so a delete that + // commits after this read and before the provider receives the bytes is + // not stopped — the window is one statement plus the request, per job. + // + // This is the ONE liveness guard on the send path. A second, earlier copy + // in State was redundant with it (mutant M26 survived its removal), so it + // is gone; State serves the read path too, where nothing is sent. + itemLive, wsLive, err := r.store.ItemLiveness(item.ID) + if err != nil { + return false, err + } + if !itemLive { + return false, ErrItemGone + } + if !wsLive { + return false, ErrWorkspaceDeleted + } + answers, usage, err := r.provider.Ask(ctx, st.Bytes, qs.Questions) + if err != nil { + return true, err + } + rows := make([]models.ItemDecision, 0, len(answers)) + for _, key := range keys { + a, ok := answers[key] + if !ok { + // The provider contract is one answer per question; a missing + // one is a provider defect, and storing a partial set would make + // the idempotency check re-ask forever. + return true, fmt.Errorf("decision: provider returned no answer for %q", key) + } + raw, err := json.Marshal(storedAnswerOf(a)) + if err != nil { + return true, err + } + rows = append(rows, models.ItemDecision{ + ItemID: item.ID, + QuestionSet: qs.Name, + QuestionKey: key, + Kind: string(a.Kind), + Answer: raw, + Confidence: a.Confidence, + Provider: r.provider.Name(), + Model: r.provider.Model(), + StateHash: st.Hash, + QuestionHash: qhash[key], + ItemSeq: item.Seq, + StateTruncated: st.Truncated || usage.StateTruncated, + }) + } + return true, r.store.InsertItemDecisions(item.WorkspaceID, rows) +} + +// storedAnswer is the persisted shape of an [Answer]: the provider's wire +// vocabulary, with only the members the kind carries. +type storedAnswer struct { + Type Kind `json:"type"` + Choice *string `json:"choice,omitempty"` + Score *float64 `json:"score,omitempty"` + Noul *float64 `json:"noul,omitempty"` + Legend map[string]string `json:"legend,omitempty"` + Probabilities map[string]float64 `json:"probabilities,omitempty"` + Confidence *float64 `json:"confidence,omitempty"` +} + +func storedAnswerOf(a Answer) storedAnswer { + out := storedAnswer{Type: a.Kind, Legend: a.Legend, Probabilities: a.Probabilities, Confidence: a.Confidence} + switch a.Kind { + case KindChoice: + c := a.Choice + out.Choice = &c + case KindScore: + sc := a.Score + out.Score = &sc + case KindNoul: + n := a.Noul + out.Noul = &n + } + return out +} + +// Decisions returns the item's latest answer per question, each marked +// Current when it was computed from the item's present state. A nil runner +// returns an empty list: no provider, no decisions, no error. +func (r *Runner) Decisions(itemID string) ([]models.ItemDecision, error) { + if r == nil { + return []models.ItemDecision{}, nil + } + rows, err := r.store.LatestItemDecisions(itemID) + if err != nil { + return nil, err + } + if len(rows) == 0 { + return []models.ItemDecision{}, nil + } + _, st, err := r.State(itemID) + if err != nil { + if errors.Is(err, ErrItemGone) { + return []models.ItemDecision{}, nil + } + return nil, err + } + model := r.provider.Model() + for i := range rows { + // Current needs all three to still hold: the item state, and the + // question as registered now under the model pinned now. A set or key + // no longer registered has no "now" to match, so it is not current. + qhashNow := "" + if qs, ok := r.registry.Get(rows[i].QuestionSet); ok { + if q, ok := qs.Questions[rows[i].QuestionKey]; ok { + qhashNow = QuestionFingerprint(model, q) + } + } + rows[i].Current = rows[i].StateHash == st.Hash && qhashNow != "" && rows[i].QuestionHash == qhashNow + } + return rows, nil +} + +// Job-handling bounds. +const ( + // maxDecisionAttempts: a job PERMANENTLY refused this many times in a row + // is dropped. The next write to the item re-enqueues it, so dropping loses + // nothing a user did — it stops a refused state (an over-budget item, a + // malformed request) from costing a call every backoff forever. A + // TRANSIENT failure is never dropped; see isPermanentFailure. + maxDecisionAttempts = 5 + // decisionRetryBase is the first backoff; it doubles per attempt... + decisionRetryBase = 30 * time.Second + // ...up to decisionRetryCap, which is what a provider outage costs: one + // call per owed job per hour until the provider is back. + decisionRetryCap = time.Hour +) + +// isPermanentFailure reports whether a failed evaluation says something about +// the REQUEST rather than about the provider's availability (codex round 6). +// +// Permanent: the provider answered 4xx other than 429 — the request itself is +// the problem, and asking again unchanged gets the same answer. Everything else +// is transient: 5xx, 429/529 after the provider's own retries, transport +// errors. A transient failure must never drop the job, or an outage would +// silently discard every evaluation owed while it lasted. +func isPermanentFailure(err error) bool { + var apiErr *APIError + if errors.As(err, &apiErr) { + return apiErr.StatusCode >= 400 && apiErr.StatusCode < 500 && apiErr.StatusCode != http.StatusTooManyRequests + } + return false +} + +func decisionBackoff(attempts int) time.Duration { + d := decisionRetryBase + for i := 0; i < attempts && d < decisionRetryCap; i++ { + d *= 2 + } + if d > decisionRetryCap { + d = decisionRetryCap + } + return d +} + +// RunOnce evaluates up to limit owed jobs, CLAIMING ONE AT A TIME, and +// returns how many it claimed. +// +// One at a time so that each job's lease starts when its own evaluation does +// (codex round 4). Claiming the whole pass up front made the last job wait +// behind every earlier provider call: at the provider's worst case (4 attempts +// x 60s plus up to 30s between them, ~5.5 min per job) twenty jobs is far past +// any lease, and an expired claim is taken by another instance — a duplicate +// evaluation. A single job's worst case fits inside the lease; see +// defaultDecisionClaimLease in the server. +func (r *Runner) RunOnce(ctx context.Context, runnerID string, limit int, lease time.Duration) (int, error) { + if r == nil { + return 0, nil + } + n := 0 + for n < limit { + if ctx.Err() != nil { + break // stopping: claim nothing further + } + jobs, err := r.store.ClaimDecisionJobs(runnerID, 1, lease) + if err != nil { + return n, err + } + if len(jobs) == 0 { + break + } + n++ + r.runJob(ctx, jobs[0]) + } + return n, nil +} + +func (r *Runner) runJob(ctx context.Context, j store.DecisionJob) { + _, err := r.Evaluate(ctx, j.ItemID, j.QuestionSet) + var serr error + switch { + case err != nil && ctx.Err() != nil: + // Cancelled by shutdown, not a verdict on the job: no attempt counted. + r.release(j) + return + case errors.Is(err, ErrWorkspaceDeleted): + // Held for a restore, not a failure: released with no attempt, and + // the claim scan will not offer it again while the workspace stays + // deleted. + r.release(j) + return + case err == nil: + serr = r.store.CompleteDecisionJob(j) + case errors.Is(err, ErrItemGone), errors.Is(err, ErrUnknownSet), errors.Is(err, ErrSetNotApplicable): + serr = r.store.DropDecisionJob(j) + case isPermanentFailure(err) && j.Attempts+1 >= maxDecisionAttempts: + slog.Warn("decision job dropped after repeated refusals", + "item_id", j.ItemID, "question_set", j.QuestionSet, "attempts", j.Attempts+1, "error", err) + serr = r.store.DropDecisionJob(j) + default: + slog.Info("decision job failed, will retry", + "item_id", j.ItemID, "question_set", j.QuestionSet, "attempt", j.Attempts+1, + "permanent", isPermanentFailure(err), "error", err) + serr = r.store.FailDecisionJob(j, err, decisionBackoff(j.Attempts)) + } + if serr != nil { + // The claim's lease still bounds this: an unrecorded outcome leaves + // the job claimed until the lease lapses, and then it is re-run. + slog.Error("decision job outcome not recorded", "item_id", j.ItemID, "question_set", j.QuestionSet, "error", serr) + } +} + +func (r *Runner) release(j store.DecisionJob) { + if err := r.store.ReleaseDecisionJob(j); err != nil { + slog.Error("decision job not released", "item_id", j.ItemID, "question_set", j.QuestionSet, "error", err) + } +} diff --git a/internal/decision/runner_test.go b/internal/decision/runner_test.go new file mode 100644 index 000000000..7022dc303 --- /dev/null +++ b/internal/decision/runner_test.go @@ -0,0 +1,648 @@ +package decision + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" + "github.com/PerpetualSoftware/pad/internal/store/storetest" +) + +// Runner tests (TASK-3117). The provider is the REAL typesafe provider pointed +// at a fake server, so what these tests see is what went over the wire. + +const triageSet = "triage" + +func triageQuestions() map[string]Question { + return map[string]Question{ + "urgent": Noul("Is this item urgent?", "", ""), + "kind": Choice("What kind of item is this?", map[string]string{"bug": "a defect", "chore": "maintenance"}), + } +} + +// noulChoiceHandler answers every question it is sent. +func noulChoiceHandler(f *fake, req wireRequest, raw []byte, w http.ResponseWriter) { + answers := map[string]any{} + for key, q := range req.Questions { + switch q.Type { + case string(KindNoul): + answers[key] = map[string]any{"type": "noul", "noul": 0.8} + case string(KindChoice): + answers[key] = map[string]any{"type": "choice", "choice": "bug", "confidence": 0.9, + "probabilities": map[string]float64{"bug": 0.9, "chore": 0.1}} + } + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "model": "jev-1.13.0", "answers": answers, + "usage": map[string]int{"input_tokens": 10, "output_tokens": 5}, + }) +} + +type runnerFixture struct { + s *store.Store + r *Runner + f *fake + ws *models.Workspace + col *models.Collection + item *models.Item + called func() int +} + +func newRunnerFixture(t *testing.T) *runnerFixture { + t.Helper() + s := storetest.NewSQLite(t) + ws, err := s.CreateWorkspace(models.WorkspaceCreate{Name: "Runner"}) + if err != nil { + t.Fatal(err) + } + col, err := s.CreateCollection(ws.ID, models.CollectionCreate{Name: "Tasks", Schema: `{"fields":[{"key":"status","type":"text"}]}`}) + if err != nil { + t.Fatal(err) + } + p, f := newFake(t, noulChoiceHandler) + reg := NewRegistry() + if err := reg.Register(QuestionSet{Name: triageSet, Questions: triageQuestions()}); err != nil { + t.Fatal(err) + } + r := NewRunner(s, p, reg) + r.Install(s) + item, err := s.CreateItem(ws.ID, col.ID, models.ItemCreate{ + Title: "Login fails & retries", Content: "Steps: open \u2028 the page", Fields: `{"status":"open"}`, + }) + if err != nil { + t.Fatal(err) + } + return &runnerFixture{s: s, r: r, f: f, ws: ws, col: col, item: item, called: func() int { return len(f.requests) }} +} + +func (fx *runnerFixture) evaluate(t *testing.T) bool { + t.Helper() + called, err := fx.r.Evaluate(context.Background(), fx.item.ID, triageSet) + if err != nil { + t.Fatalf("evaluate: %v", err) + } + return called +} + +// ONE SERIALISATION PATH: the state bytes on the wire are byte-identical to +// the bytes hashed. The fixture's title and body carry <, >, & and U+2028, +// which json.Marshal escapes — a second marshal with different escaping would +// be caught here. +func TestRunner_HashedBytesAreTheBytesSent(t *testing.T) { + fx := newRunnerFixture(t) + if !fx.evaluate(t) { + t.Fatal("first evaluation made no provider call") + } + var envelope map[string]json.RawMessage + if err := json.Unmarshal([]byte(fx.f.rawBodies[0]), &envelope); err != nil { + t.Fatal(err) + } + _, st, err := fx.r.State(fx.item.ID) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(envelope["state"], st.Bytes) { + t.Fatalf("state on the wire differs from the hashed bytes:\nwire: %s\nhashed: %s", envelope["state"], st.Bytes) + } + if !bytes.Contains(st.Bytes, []byte(`\u003cb\u003e`)) || !bytes.Contains(st.Bytes, []byte(`\u2028`)) { + t.Fatalf("precondition: the fixture must exercise escaping, got %s", st.Bytes) + } +} + +// Idempotency: an unchanged state makes no second call. +func TestRunner_UnchangedStateShortCircuits(t *testing.T) { + fx := newRunnerFixture(t) + fx.evaluate(t) + if fx.evaluate(t) { + t.Fatal("re-evaluating an unchanged state called the provider") + } + if n := fx.called(); n != 1 { + t.Fatalf("provider called %d times; want 1", n) + } +} + +// A comment changes the state (the trail is hashed), so it re-evaluates — +// the case items.seq alone would have missed (ruling 2). +func TestRunner_CommentReevaluates(t *testing.T) { + fx := newRunnerFixture(t) + fx.evaluate(t) + before, err := fx.s.GetItem(fx.item.ID) + if err != nil { + t.Fatal(err) + } + if _, err := fx.s.CreateComment(fx.ws.ID, fx.item.ID, "", models.CommentCreate{Author: "kestrel", Body: "reproduced on staging"}); err != nil { + t.Fatal(err) + } + after, err := fx.s.GetItem(fx.item.ID) + if err != nil { + t.Fatal(err) + } + // The counterfactual: seq did NOT move, so a seq-keyed runner would have + // short-circuited here. + if after.Seq != before.Seq { + t.Fatalf("precondition: a comment moved items.seq (%d -> %d); this test no longer discriminates", before.Seq, after.Seq) + } + if !fx.evaluate(t) { + t.Fatal("a new comment did not re-evaluate") + } +} + +// Writes that change nothing the state carries move seq but must not cost a +// call: a role-board reorder, a parent-link add (the only link kinds that +// bump seq are parent and implements), an assignment-only update. +func TestRunner_StateInvisibleWritesDoNotReevaluate(t *testing.T) { + fx := newRunnerFixture(t) + other, err := fx.s.CreateItem(fx.ws.ID, fx.col.ID, models.ItemCreate{Title: "Parent", Fields: `{"status":"open"}`}) + if err != nil { + t.Fatal(err) + } + role, err := fx.s.CreateAgentRole(fx.ws.ID, models.AgentRoleCreate{Name: "Implementer"}) + if err != nil { + t.Fatal(err) + } + fx.evaluate(t) + + writes := []struct { + name string + do func() error + }{ + {"role-board reorder", func() error { + return fx.s.UpdateRoleSortOrder(fx.ws.ID, []store.RoleSortUpdate{{ItemID: fx.item.ID, RoleSortOrder: 7}}) + }}, + {"parent link add", func() error { + _, err := fx.s.CreateItemLink(fx.ws.ID, models.ItemLinkCreate{TargetID: other.ID, LinkType: "parent"}, fx.item.ID) + return err + }}, + {"role assignment", func() error { + _, err := fx.s.UpdateItem(fx.item.ID, models.ItemUpdate{AgentRoleID: &role.ID}) + return err + }}, + } + for _, w := range writes { + before, _ := fx.s.GetItem(fx.item.ID) + if err := w.do(); err != nil { + t.Fatalf("%s: %v", w.name, err) + } + after, _ := fx.s.GetItem(fx.item.ID) + if after.Seq == before.Seq { + t.Fatalf("precondition: %s did not move items.seq; this leg no longer discriminates", w.name) + } + if fx.evaluate(t) { + t.Fatalf("%s re-evaluated an unchanged state", w.name) + } + } + if n := fx.called(); n != 1 { + t.Fatalf("provider called %d times across three state-invisible writes; want 1", n) + } +} + +// The read marks a row current only while its state is the item's state. +func TestRunner_DecisionsReadMarksCurrent(t *testing.T) { + fx := newRunnerFixture(t) + fx.evaluate(t) + got, err := fx.r.Decisions(fx.item.ID) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 { + t.Fatalf("got %d decisions; want 2", len(got)) + } + for _, d := range got { + if !d.Current { + t.Fatalf("%s: freshly evaluated answer is not current", d.QuestionKey) + } + } + if _, err := fx.s.CreateComment(fx.ws.ID, fx.item.ID, "", models.CommentCreate{Author: "a", Body: "new info"}); err != nil { + t.Fatal(err) + } + got, err = fx.r.Decisions(fx.item.ID) + if err != nil { + t.Fatal(err) + } + for _, d := range got { + if d.Current { + t.Fatalf("%s: answer computed before a comment still reads as current", d.QuestionKey) + } + } +} + +// Nil provider: no runner, no resolver, no job rows, and an empty read. +func TestRunner_NilProviderIsANoOp(t *testing.T) { + s := storetest.NewSQLite(t) + r := NewRunner(s, nil, NewRegistry()) + if r != nil { + t.Fatal("NewRunner with a nil provider returned a runner") + } + r.Install(s) + ws, _ := s.CreateWorkspace(models.WorkspaceCreate{Name: "Off"}) + col, _ := s.CreateCollection(ws.ID, models.CollectionCreate{Name: "Tasks", Schema: `{"fields":[]}`}) + item, err := s.CreateItem(ws.ID, col.ID, models.ItemCreate{Title: "x"}) + if err != nil { + t.Fatal(err) + } + if j, err := s.GetDecisionJob(item.ID, triageSet); err != nil || j != nil { + t.Fatalf("nil provider left a job row: %+v, %v", j, err) + } + got, err := r.Decisions(item.ID) + if err != nil || got == nil || len(got) != 0 { + t.Fatalf("nil runner read = %v, %v; want an empty non-nil list", got, err) + } + if n, err := r.RunOnce(context.Background(), "x", 10, time.Minute); n != 0 || err != nil { + t.Fatalf("nil runner RunOnce = %d, %v", n, err) + } +} + +// End to end through the queue: a write enqueues, one tick evaluates and +// retires the job, and a second tick finds nothing. +func TestRunner_RunOnceDrainsTheQueue(t *testing.T) { + fx := newRunnerFixture(t) + n, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute) + if err != nil || n != 1 { + t.Fatalf("RunOnce claimed %d (err %v); want 1", n, err) + } + if fx.called() != 1 { + t.Fatalf("provider called %d times; want 1", fx.called()) + } + if j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet); j != nil { + t.Fatalf("job still owed after a successful evaluation: %+v", j) + } + if n, _ := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); n != 0 { + t.Fatalf("second tick claimed %d; want 0", n) + } +} + +// A failing provider leaves the job owed with the failure recorded, and a +// deleted item's job is dropped rather than retried. +func TestRunner_FailureAndGoneItem(t *testing.T) { + fx := newRunnerFixture(t) + fx.f.srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"detail":"bad request"}`, http.StatusBadRequest) + }) + if _, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil { + t.Fatal(err) + } + j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet) + if j == nil || j.Attempts != 1 || j.LastError == "" { + t.Fatalf("after a provider failure the job is %+v; want attempts=1 with an error", j) + } + + if err := fx.s.DeleteItem(fx.item.ID); err != nil { + t.Fatal(err) + } + if _, err := fx.r.Evaluate(context.Background(), fx.item.ID, triageSet); !errors.Is(err, ErrItemGone) { + t.Fatalf("evaluating a deleted item: err = %v; want ErrItemGone", err) + } +} + +func TestRegistry_RefusesDuplicatesAndInvalidSets(t *testing.T) { + reg := NewRegistry() + if err := reg.Register(QuestionSet{Name: "a", Questions: triageQuestions()}); err != nil { + t.Fatal(err) + } + for i, qs := range []QuestionSet{ + {Name: "a", Questions: triageQuestions()}, + {Name: "", Questions: triageQuestions()}, + {Name: "b"}, + {Name: "c", Questions: map[string]Question{"": Noul("x", "", "")}}, + {Name: "d", Questions: map[string]Question{"k": Choice("x", map[string]string{"only": "one"})}}, + } { + if err := reg.Register(qs); err == nil { + t.Errorf("case %d: Register(%+v) accepted an invalid set", i, qs) + } + } + reg2 := NewRegistry() + _ = reg2.Register(QuestionSet{Name: "scoped", Collections: []string{"bugs"}, Questions: triageQuestions()}) + _ = reg2.Register(QuestionSet{Name: "all", Questions: triageQuestions()}) + if got := fmt.Sprint(reg2.SetsFor("bugs")); got != "[all scoped]" { + t.Fatalf("SetsFor(bugs) = %s", got) + } + if got := fmt.Sprint(reg2.SetsFor("tasks")); got != "[all]" { + t.Fatalf("SetsFor(tasks) = %s", got) + } +} + +// A pass that starts already cancelled claims nothing, calls nothing, and +// leaves the job owed with no attempt counted. +func TestRunner_CancelledPassReleasesWithoutCountingAFailure(t *testing.T) { + fx := newRunnerFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + n, err := fx.r.RunOnce(ctx, "stopping", 10, time.Hour) + if err != nil || n != 0 { + t.Fatalf("an already-cancelled pass claimed %d (err %v); want 0", n, err) + } + if fx.called() != 0 { + t.Fatalf("a cancelled pass called the provider %d times", fx.called()) + } + j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet) + if j == nil || j.Attempts != 0 || j.ClaimedBy != "" { + t.Fatalf("after a cancelled pass the job is %+v; want owed, unclaimed, attempts=0", j) + } +} + +// Cancelled MID-CALL: the provider request is in flight when shutdown +// cancels the context. runJob must release the claim without counting an +// attempt — distinct from the test above, where cancellation lands before +// any job starts and runJob is never reached. +func TestRunner_CancelledMidCallReleasesWithoutCountingAFailure(t *testing.T) { + fx := newRunnerFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + fx.f.srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cancel() + // Bounded: the server notices a client disconnect only after the + // body is read, so an unbounded wait here holds Close() forever. + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + case <-time.After(2 * time.Second): + } + }) + n, err := fx.r.RunOnce(ctx, "stopping", 10, time.Hour) + if err != nil || n != 1 { + t.Fatalf("RunOnce claimed %d (err %v); want 1", n, err) + } + j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet) + if j == nil || j.Attempts != 0 || j.ClaimedBy != "" || j.LastError != "" { + t.Fatalf("after a mid-call cancel the job is %+v; want owed, unclaimed, attempts=0, no error", j) + } +} + +// A soft-deleted WORKSPACE or COLLECTION stops evaluation even though the item +// row itself is live: nothing its owner deleted goes to the provider. A +// workspace's jobs are held for its restore; a collection's are dropped. +func TestRunner_DeletedWorkspaceOrCollectionIsNeverSent(t *testing.T) { + for _, tc := range []struct { + name string + del func(fx *runnerFixture) error + }{ + {"workspace", func(fx *runnerFixture) error { return fx.s.DeleteWorkspace(fx.ws.Slug) }}, + {"collection", func(fx *runnerFixture) error { return fx.s.DeleteCollection(fx.col.ID, "") }}, + } { + t.Run(tc.name, func(t *testing.T) { + fx := newRunnerFixture(t) + if err := tc.del(fx); err != nil { + t.Fatalf("soft-delete %s: %v", tc.name, err) + } + if it, _ := fx.s.GetItem(fx.item.ID); it == nil { + t.Fatalf("precondition: GetItem hides the item after a %s delete; this leg no longer discriminates", tc.name) + } + if _, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil { + t.Fatal(err) + } + if fx.called() != 0 { + t.Fatalf("an item in a soft-deleted %s was sent to the provider", tc.name) + } + j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet) + switch tc.name { + case "workspace": + // HELD, not dropped: a workspace restore must find the work + // still owed (codex round 5). + if j == nil || j.ClaimedBy != "" || j.Attempts != 0 { + t.Fatalf("job for an item in a soft-deleted workspace is %+v; want held (owed, unclaimed, attempts=0)", j) + } + case "collection": + // A collection has no restore path, so its jobs are dropped. + if j != nil { + t.Fatalf("job for an item in a soft-deleted collection was kept: %+v", j) + } + } + }) + } +} + +// Codex round 2 P2: a job owed for a set scoped to the item's OLD collection +// must not reach the provider after the item moves out of that collection. +func TestRunner_SetNoLongerApplicableAfterMoveIsNotAsked(t *testing.T) { + s := storetest.NewSQLite(t) + ws, _ := s.CreateWorkspace(models.WorkspaceCreate{Name: "Scope"}) + bugs, err := s.CreateCollection(ws.ID, models.CollectionCreate{Name: "Bugs", Schema: `{"fields":[]}`}) + if err != nil { + t.Fatal(err) + } + ideas, err := s.CreateCollection(ws.ID, models.CollectionCreate{Name: "Ideas", Schema: `{"fields":[]}`}) + if err != nil { + t.Fatal(err) + } + p, f := newFake(t, noulChoiceHandler) + reg := NewRegistry() + if err := reg.Register(QuestionSet{Name: "bug-triage", Collections: []string{bugs.Slug}, Questions: triageQuestions()}); err != nil { + t.Fatal(err) + } + r := NewRunner(s, p, reg) + r.Install(s) + + item, err := s.CreateItem(ws.ID, bugs.ID, models.ItemCreate{Title: "Crash on save"}) + if err != nil { + t.Fatal(err) + } + if j, _ := s.GetDecisionJob(item.ID, "bug-triage"); j == nil { + t.Fatal("precondition: no job owed for the scoped set") + } + if _, err := s.MoveItem(item.ID, ideas.ID, item.Fields); err != nil { + t.Fatal(err) + } + if _, err := r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil { + t.Fatal(err) + } + if len(f.requests) != 0 { + t.Fatalf("asked a set scoped to %q about an item now in %q", bugs.Slug, ideas.Slug) + } + if j, _ := s.GetDecisionJob(item.ID, "bug-triage"); j != nil { + t.Fatalf("an inapplicable job was kept: %+v", j) + } +} + +// Codex round 4: jobs are claimed ONE AT A TIME, so a job's lease starts when +// its own evaluation starts. Claiming a whole pass up front let the last job +// wait out its lease behind the others' provider calls and be re-claimed by +// another instance — a duplicate evaluation. +func TestRunner_ClaimsOneJobAtATime(t *testing.T) { + fx := newRunnerFixture(t) + for i := 0; i < 2; i++ { + if _, err := fx.s.CreateItem(fx.ws.ID, fx.col.ID, models.ItemCreate{Title: fmt.Sprintf("more %d", i)}); err != nil { + t.Fatal(err) + } + } + items, err := fx.s.ListItems(fx.ws.ID, models.ItemListParams{}) + if err != nil || len(items) != 3 { + t.Fatalf("precondition: %d items (%v)", len(items), err) + } + maxClaimed := 0 + fx.f.srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + claimed := 0 + for _, it := range items { + if j, _ := fx.s.GetDecisionJob(it.ID, triageSet); j != nil && j.ClaimedBy != "" { + claimed++ + } + } + if claimed > maxClaimed { + maxClaimed = claimed + } + var req wireRequest + _ = json.NewDecoder(r.Body).Decode(&req) + noulChoiceHandler(fx.f, req, nil, w) + }) + n, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Hour) + if err != nil || n != 3 { + t.Fatalf("RunOnce processed %d (err %v); want 3", n, err) + } + if maxClaimed != 1 { + t.Fatalf("up to %d jobs were claimed while a provider call was in flight; want 1", maxClaimed) + } +} + +// Codex round 4: liveness is re-checked immediately before the provider call, +// so a delete landing between building the state and asking stops the send. +func TestRunner_DeleteJustBeforeAskIsNotSent(t *testing.T) { + fx := newRunnerFixture(t) + fx.r.beforeAsk = func() { + if err := fx.s.DeleteWorkspace(fx.ws.Slug); err != nil { + t.Fatal(err) + } + } + if _, err := fx.r.Evaluate(context.Background(), fx.item.ID, triageSet); !errors.Is(err, ErrWorkspaceDeleted) { + t.Fatalf("err = %v; want ErrWorkspaceDeleted", err) + } + if fx.called() != 0 { + t.Fatal("content of a workspace deleted before the call was sent to the provider") + } +} + +// Codex round 5: work owed in a workspace that is soft-deleted and then +// RESTORED is evaluated after the restore — nothing enqueues on restore (it +// is a fan-out), so the job must have survived the deletion. +func TestRunner_WorkspaceRestoreResumesHeldWork(t *testing.T) { + fx := newRunnerFixture(t) + if err := fx.s.DeleteWorkspace(fx.ws.Slug); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { // more than one pass: a held job must not be claimed, nor spin + if n, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil || n != 0 { + t.Fatalf("pass %d over a deleted workspace claimed %d (err %v); want 0", i, n, err) + } + } + if err := fx.s.RestoreWorkspace(fx.ws.Slug); err != nil { + t.Fatal(err) + } + if n, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil || n != 1 { + t.Fatalf("after restore RunOnce claimed %d (err %v); want 1", n, err) + } + if fx.called() != 1 { + t.Fatalf("provider called %d times after restore; want 1", fx.called()) + } +} + +// Codex round 6: an answer is current only for the QUESTION and MODEL that +// produced it. Rewording a question, or changing the pinned model, at an +// unchanged item state must re-ask — and the old answer must stop reading as +// current. +func TestRunner_QuestionOrModelChangeReasks(t *testing.T) { + fx := newRunnerFixture(t) + fx.evaluate(t) + + reworded := NewRegistry() + qs := triageQuestions() + qs["urgent"] = Noul("Does this block a release?", "", "") + if err := reworded.Register(QuestionSet{Name: triageSet, Questions: qs}); err != nil { + t.Fatal(err) + } + p2 := newTypesafe("test-key", "jev-1.13.0") + p2.setEndpoint(fx.f.srv.URL) + r2 := NewRunner(fx.s, p2, reworded) + got, err := r2.Decisions(fx.item.ID) + if err != nil { + t.Fatal(err) + } + for _, d := range got { + want := d.QuestionKey != "urgent" + if d.Current != want { + t.Fatalf("after rewording `urgent`, %s current=%v; want %v", d.QuestionKey, d.Current, want) + } + } + if called, err := r2.Evaluate(context.Background(), fx.item.ID, triageSet); err != nil || !called { + t.Fatalf("a reworded question at an unchanged state was not re-asked (called=%v err=%v)", called, err) + } + + p3 := newTypesafe("test-key", "jev-9.9.9") + p3.setEndpoint(fx.f.srv.URL) + r3 := NewRunner(fx.s, p3, reworded) + if called, err := r3.Evaluate(context.Background(), fx.item.ID, triageSet); err != nil || !called { + t.Fatalf("a model change at an unchanged state was not re-asked (called=%v err=%v)", called, err) + } +} + +// Codex round 6: a provider OUTAGE must not lose owed work. Transient failures +// (5xx, 429, transport) back off indefinitely; only a permanent refusal (a 4xx +// naming the request itself) is dropped after repeated attempts. +func TestRunner_TransientFailuresAreNeverDropped(t *testing.T) { + fx := newRunnerFixture(t) + fx.f.srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"detail":"upstream down"}`, http.StatusServiceUnavailable) + }) + for i := 0; i < maxDecisionAttempts+2; i++ { + // Expire any backoff so each pass claims the job again. + expireDecisionBackoff(t, fx) + if _, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil { + t.Fatal(err) + } + } + if j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet); j == nil { + t.Fatalf("a job failing only transiently was dropped after %d attempts", maxDecisionAttempts+2) + } +} + +func TestRunner_PermanentRefusalIsDroppedAfterRepeatedAttempts(t *testing.T) { + fx := newRunnerFixture(t) + fx.f.srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"detail":"bad request"}`, http.StatusBadRequest) + }) + for i := 0; i < maxDecisionAttempts; i++ { + expireDecisionBackoff(t, fx) + if _, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil { + t.Fatal(err) + } + } + if j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet); j != nil { + t.Fatalf("a permanently refused job was kept after %d attempts: %+v", maxDecisionAttempts, j) + } +} + +// expireDecisionBackoff makes a held-for-backoff job claimable now, through +// the same release a superseding write would cause. +func expireDecisionBackoff(t *testing.T, fx *runnerFixture) { + t.Helper() + j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet) + if j == nil || j.ClaimedBy == "" { + return + } + if err := fx.s.ReleaseDecisionJob(store.DecisionJob{ItemID: fx.item.ID, QuestionSet: triageSet, ClaimedBy: j.ClaimedBy}); err != nil { + t.Fatal(err) + } +} + +// M35: a workspace deleted between the claim and the send, through RunOnce — +// the job is HELD (owed, unclaimed, no attempt), not dropped. +func TestRunner_WorkspaceDeletedMidJobIsHeld(t *testing.T) { + fx := newRunnerFixture(t) + fx.r.beforeAsk = func() { + if err := fx.s.DeleteWorkspace(fx.ws.Slug); err != nil { + t.Fatal(err) + } + } + if _, err := fx.r.RunOnce(context.Background(), "tick", 10, time.Minute); err != nil { + t.Fatal(err) + } + j, _ := fx.s.GetDecisionJob(fx.item.ID, triageSet) + if j == nil || j.ClaimedBy != "" || j.Attempts != 0 { + t.Fatalf("job for a workspace deleted mid-job is %+v; want held (owed, unclaimed, attempts=0)", j) + } +} diff --git a/internal/models/decision.go b/internal/models/decision.go new file mode 100644 index 000000000..1b58022a7 --- /dev/null +++ b/internal/models/decision.go @@ -0,0 +1,34 @@ +package models + +import "encoding/json" + +// ItemDecision is one stored answer to one typed question about an item +// (PLAN-3114 unit 2, TASK-3117). +// +// Answer is the provider's answer as the decision package marshalled it; its +// shape depends on Kind (choice / score / noul) and is documented on +// decision.Answer. Confidence is nil where the primitive returns none — a Noul +// answer carries no confidence, and a zero would read as "certainly not". +// +// Current is computed at READ time, never stored: it is true when StateHash +// equals the hash of the item's state as it stands now AND QuestionHash equals +// the fingerprint of the question as registered now, under the model the +// provider is pinned to now. An answer computed before a later edit, comment, +// question rewording or model change is kept for the audit but is not current. +type ItemDecision struct { + ID string `json:"id"` + ItemID string `json:"item_id"` + QuestionSet string `json:"question_set"` + QuestionKey string `json:"question_key"` + Kind string `json:"kind"` + Answer json.RawMessage `json:"answer"` + Confidence *float64 `json:"confidence"` + Provider string `json:"provider"` + Model string `json:"model"` + StateHash string `json:"state_hash"` + QuestionHash string `json:"question_hash"` + ItemSeq int64 `json:"item_seq"` + StateTruncated bool `json:"state_truncated,omitempty"` + EvaluatedAt string `json:"evaluated_at"` + Current bool `json:"current"` +} diff --git a/internal/models/item.go b/internal/models/item.go index f7e1e3a3b..849271261 100644 --- a/internal/models/item.go +++ b/internal/models/item.go @@ -421,6 +421,13 @@ type Item struct { // nor the version history. Lease *ItemLease `json:"lease,omitempty"` + // Decisions are the item's latest typed-decision answers, one per + // question, each marked current or not (TASK-3117). Populated ONLY on the + // single-item GET, and absent when there are none — including on every + // instance with no decision provider configured, so the item shape there + // is unchanged. Never persisted through item paths. + Decisions []ItemDecision `json:"decisions,omitempty"` + DerivedClosure *ItemDerivedClosure `json:"derived_closure,omitempty"` CodeContext *ItemCodeContext `json:"code_context,omitempty"` Convention *ItemConventionMetadata `json:"convention,omitempty"` diff --git a/internal/server/decision_tick.go b/internal/server/decision_tick.go new file mode 100644 index 000000000..e9c502c0e --- /dev/null +++ b/internal/server/decision_tick.go @@ -0,0 +1,168 @@ +package server + +import ( + "context" + "log/slog" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/PerpetualSoftware/pad/internal/decision" +) + +// The decision tick: the half of TASK-3117 that does the provider calls. +// +// Writes only mark an evaluation owed (a decision_jobs row, in the write's own +// transaction — the day-73 async-only ruling). This loop claims owed jobs and +// runs them. Same settled sweeper shape as reminder_tick.go: config struct +// with its own mutex and stop channel, tracked by Server.bg so Stop() drains +// it before the DB closes, recoverSweeper on the goroutine, and an injectable +// tick channel for tests. +// +// With no provider configured the runner is nil, the store has no resolver, +// no job row is ever written, and this loop is never started. + +const ( + // defaultDecisionTickInterval bounds how stale a decision is after a + // write: at most one interval plus the provider's latency. Decisions are + // advisory reads, not alarms, so a few seconds is well inside what a + // reader would notice, and an idle tick costs one indexed scan of an + // empty table. + defaultDecisionTickInterval = 5 * time.Second + + // defaultDecisionTickLimit caps jobs per pass. Each job is a provider + // round trip, evaluated sequentially, so this bounds one pass's wall + // time rather than throughput. + defaultDecisionTickLimit = 20 + + // defaultDecisionClaimLease must outlast ONE JOB, not one pass: the + // runner claims jobs one at a time (codex round 4), so a lease starts + // when its own evaluation does. A lease that lapses mid-call hands the + // job to another instance and buys a duplicate provider call. + // + // THE RECEIPT: typesafe.go's retry policy is maxAttempts=4 at a 60s + // requestTimeout, with up to maxRetryAfter=30s between attempts — a + // worst case of 4*60 + 3*30 = 330s per job. 10 minutes is ~1.8x that. + // A provider with a longer retry budget needs this raised with it. + defaultDecisionClaimLease = 10 * time.Minute +) + +type decisionTickConfig struct { + mu sync.Mutex + runner *decision.Runner + interval time.Duration + limit int + lease time.Duration + runnerID string + stop chan struct{} + cancel context.CancelFunc + running bool + tick <-chan time.Time +} + +// SetDecisionRunner attaches the decision runner and wires its registry into +// the store's write doors. A nil runner (no provider configured) removes any +// resolver, so the doors enqueue nothing. +func (s *Server) SetDecisionRunner(r *decision.Runner) { + s.decisionTick.mu.Lock() + s.decisionTick.runner = r + s.decisionTick.mu.Unlock() + if s.store != nil { + r.Install(s.store) + } +} + +func (s *Server) decisionRunner() *decision.Runner { + s.decisionTick.mu.Lock() + defer s.decisionTick.mu.Unlock() + return s.decisionTick.runner +} + +// SetDecisionTickChannel replaces the interval ticker with a caller-driven +// channel. Test affordance only. +func (s *Server) SetDecisionTickChannel(c <-chan time.Time) { + s.decisionTick.mu.Lock() + defer s.decisionTick.mu.Unlock() + s.decisionTick.tick = c +} + +// StartDecisionTick starts the evaluation loop. Idempotent, and a no-op with +// no runner attached. +func (s *Server) StartDecisionTick() { + s.decisionTick.mu.Lock() + if s.decisionTick.running || s.decisionTick.runner == nil { + s.decisionTick.mu.Unlock() + return + } + if s.decisionTick.interval == 0 { + s.decisionTick.interval = defaultDecisionTickInterval + } + if s.decisionTick.limit == 0 { + s.decisionTick.limit = defaultDecisionTickLimit + } + if s.decisionTick.lease == 0 { + s.decisionTick.lease = defaultDecisionClaimLease + } + if s.decisionTick.runnerID == "" { + s.decisionTick.runnerID = "decision-" + uuid.NewString() + } + ctx, cancel := context.WithCancel(context.Background()) + s.decisionTick.stop = make(chan struct{}) + s.decisionTick.cancel = cancel + s.decisionTick.running = true + interval := s.decisionTick.interval + stop := s.decisionTick.stop + tick := s.decisionTick.tick + s.decisionTick.mu.Unlock() + + slog.Info("decision tick started", "interval", interval.String()) + + s.bg.Add(1) + go func() { + defer s.bg.Done() + defer s.recoverSweeper("decision-tick") + var c <-chan time.Time + if tick != nil { + c = tick + } else { + t := time.NewTicker(interval) + defer t.Stop() + c = t.C + } + for { + select { + case <-stop: + return + case <-c: + s.runDecisionTick(ctx) + } + } + }() +} + +// stopDecisionTick signals the loop to exit and cancels an in-flight provider +// call, so Stop() does not wait out a network round trip before closing the +// database. The runner releases a cancelled job's claim without counting an +// attempt, so it is claimable again at once. +func (s *Server) stopDecisionTick() { + s.decisionTick.mu.Lock() + defer s.decisionTick.mu.Unlock() + if !s.decisionTick.running { + return + } + close(s.decisionTick.stop) + s.decisionTick.cancel() + s.decisionTick.running = false +} + +func (s *Server) runDecisionTick(ctx context.Context) { + s.decisionTick.mu.Lock() + r, id, limit, lease := s.decisionTick.runner, s.decisionTick.runnerID, s.decisionTick.limit, s.decisionTick.lease + s.decisionTick.mu.Unlock() + + n, err := r.RunOnce(ctx, id, limit, lease) + if err != nil { + slog.Error("decision tick: claim failed", "error", err, "claimed", n) + } +} diff --git a/internal/server/handlers_decisions.go b/internal/server/handlers_decisions.go new file mode 100644 index 000000000..c8665306e --- /dev/null +++ b/internal/server/handlers_decisions.go @@ -0,0 +1,42 @@ +package server + +import ( + "net/http" + + "github.com/go-chi/chi/v5" +) + +// handleListItemDecisions returns an item's latest typed-decision answer per +// question (TASK-3117). An answer computed from an earlier state of the item +// is still returned — it is the newest one there is — with current=false. +// +// With no provider configured the list is empty: no rows are ever written, +// and rows written under an earlier configuration are not served, since +// nothing would keep them current. +func (s *Server) handleListItemDecisions(w http.ResponseWriter, r *http.Request) { + workspaceID, ok := s.getWorkspaceID(w, r) + if !ok { + return + } + item, err := s.store.ResolveItem(workspaceID, chi.URLParam(r, "itemSlug")) + if err != nil { + writeInternalError(w, err) + return + } + if item == nil { + writeError(w, http.StatusNotFound, "not_found", "Item not found") + return + } + if !s.requireItemVisible(w, r, workspaceID, item) { + return + } + decisions, err := s.decisionRunner().Decisions(item.ID) + if err != nil { + writeInternalError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "ref": item.Ref, + "decisions": decisions, + }) +} diff --git a/internal/server/handlers_decisions_test.go b/internal/server/handlers_decisions_test.go new file mode 100644 index 000000000..dccd783e7 --- /dev/null +++ b/internal/server/handlers_decisions_test.go @@ -0,0 +1,173 @@ +package server + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/decision" + "github.com/PerpetualSoftware/pad/internal/models" +) + +// stubProvider answers every question with a fixed Noul and counts calls. +type stubProvider struct { + mu sync.Mutex + calls int +} + +func (p *stubProvider) Name() string { return "stub" } +func (p *stubProvider) Model() string { return "stub-1" } +func (p *stubProvider) Ask(_ context.Context, _ any, qs map[string]decision.Question) (map[string]decision.Answer, decision.Usage, error) { + p.mu.Lock() + p.calls++ + p.mu.Unlock() + out := map[string]decision.Answer{} + for k := range qs { + out[k] = decision.Answer{Kind: decision.KindNoul, Noul: 0.75} + } + return out, decision.Usage{}, nil +} + +func (p *stubProvider) count() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.calls +} + +func attachStubRunner(t *testing.T, srv *Server) *stubProvider { + t.Helper() + reg := decision.NewRegistry() + if err := reg.Register(decision.QuestionSet{ + Name: "triage", + Questions: map[string]decision.Question{"urgent": decision.Noul("Is this urgent?", "", "")}, + }); err != nil { + t.Fatal(err) + } + p := &stubProvider{} + srv.SetDecisionRunner(decision.NewRunner(srv.store, p, reg)) + return p +} + +type decisionsBody struct { + Ref string `json:"ref"` + Decisions []models.ItemDecision `json:"decisions"` +} + +func getDecisions(t *testing.T, srv *Server, wsSlug, itemSlug string) decisionsBody { + t.Helper() + rr := doRequest(srv, "GET", "/api/v1/workspaces/"+wsSlug+"/items/"+itemSlug+"/decisions", nil) + if rr.Code != http.StatusOK { + t.Fatalf("GET decisions: %d %s", rr.Code, rr.Body.String()) + } + var b decisionsBody + parseJSON(t, rr, &b) + if b.Decisions == nil { + t.Fatalf("decisions is null, want a list: %s", rr.Body.String()) + } + return b +} + +func itemJSONKeys(t *testing.T, srv *Server, wsSlug, itemSlug string) map[string]json.RawMessage { + t.Helper() + rr := doRequest(srv, "GET", "/api/v1/workspaces/"+wsSlug+"/items/"+itemSlug, nil) + if rr.Code != http.StatusOK { + t.Fatalf("GET item: %d %s", rr.Code, rr.Body.String()) + } + var m map[string]json.RawMessage + if err := json.Unmarshal(rr.Body.Bytes(), &m); err != nil { + t.Fatal(err) + } + return m +} + +// No provider: the endpoint answers an empty list, the item has no +// `decisions` member, and no write left a job row. +func TestDecisions_NoProviderIsEmpty(t *testing.T) { + t.Parallel() + srv := testServer(t) + slug := createWSWithCollections(t, srv) + item := createItem(t, srv, slug, "tasks", map[string]interface{}{"title": "Plain", "fields": `{"status":"open"}`}) + + if b := getDecisions(t, srv, slug, item.Slug); len(b.Decisions) != 0 { + t.Fatalf("no provider, got %d decisions", len(b.Decisions)) + } + if _, has := itemJSONKeys(t, srv, slug, item.Slug)["decisions"]; has { + t.Fatal("item GET carries a decisions member with no provider configured") + } + if j, err := srv.store.GetDecisionJob(item.ID, "triage"); err != nil || j != nil { + t.Fatalf("no provider, but a job row exists: %+v %v", j, err) + } +} + +// Through the HTTP doors end to end, with the tick as the evaluator — the +// binding under test is the wiring (write door -> queue -> tick -> read), +// not the runner called directly. +func TestDecisions_TickEvaluatesAndReadReportsCurrency(t *testing.T) { + t.Parallel() + srv := testServer(t) + p := attachStubRunner(t, srv) + tick := make(chan time.Time) + srv.SetDecisionTickChannel(tick) + srv.StartDecisionTick() + t.Cleanup(srv.stopDecisionTick) + + slug := createWSWithCollections(t, srv) + item := createItem(t, srv, slug, "tasks", map[string]interface{}{"title": "Wired", "fields": `{"status":"open"}`}) + if j, _ := srv.store.GetDecisionJob(item.ID, "triage"); j == nil { + t.Fatal("create through the API did not enqueue a job") + } + + runTickUntil := func(cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatal("tick did not complete the evaluation within 5s") + } + select { + case tick <- time.Now(): + case <-time.After(100 * time.Millisecond): + } + } + } + runTickUntil(func() bool { + j, _ := srv.store.GetDecisionJob(item.ID, "triage") + return j == nil && p.count() == 1 + }) + + b := getDecisions(t, srv, slug, item.Slug) + if len(b.Decisions) != 1 || !b.Decisions[0].Current || b.Decisions[0].QuestionKey != "urgent" { + t.Fatalf("after evaluation: %+v", b.Decisions) + } + if _, has := itemJSONKeys(t, srv, slug, item.Slug)["decisions"]; !has { + t.Fatal("item GET omits decisions although one is stored") + } + + // A comment through the API: the stored answer is no longer current, and + // the next tick re-evaluates. + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/comments", map[string]string{"body": "new facts"}) + if rr.Code != http.StatusCreated { + t.Fatalf("comment: %d %s", rr.Code, rr.Body.String()) + } + if b := getDecisions(t, srv, slug, item.Slug); b.Decisions[0].Current { + t.Fatal("answer computed before the comment still reads as current") + } + runTickUntil(func() bool { return p.count() == 2 }) + runTickUntil(func() bool { + b := getDecisions(t, srv, slug, item.Slug) + return len(b.Decisions) == 1 && b.Decisions[0].Current + }) +} + +func TestDecisions_UnknownItemIs404(t *testing.T) { + t.Parallel() + srv := testServer(t) + slug := createWSWithCollections(t, srv) + rr := doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items/NOPE-999/decisions", nil) + if rr.Code != http.StatusNotFound { + t.Fatalf("unknown item: %d %s", rr.Code, rr.Body.String()) + } +} diff --git a/internal/server/handlers_items.go b/internal/server/handlers_items.go index 583affa2c..6cfd9d0c2 100644 --- a/internal/server/handlers_items.go +++ b/internal/server/handlers_items.go @@ -1066,6 +1066,16 @@ func (s *Server) handleGetItem(w http.ResponseWriter, r *http.Request) { } item.Lease = lease + // Typed decisions (TASK-3117), same placement reasoning as the lease. A + // failure here is logged and the member omitted rather than failing the + // item read: decisions are advisory, and an item that cannot be fetched + // because a secondary read failed is a worse answer than one without them. + if decisions, derr := s.decisionRunner().Decisions(item.ID); derr != nil { + slog.Warn("item read: decisions unavailable", "item_id", item.ID, "error", derr) + } else if len(decisions) > 0 { + item.Decisions = decisions + } + writeJSON(w, http.StatusOK, item) } diff --git a/internal/server/server.go b/internal/server/server.go index 9b82172a3..8f061c369 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -254,6 +254,11 @@ type Server struct { // the loop via stopReminderTick. reminderTick reminderTickConfig + // decisionTick holds the typed-decision runner and its evaluation loop + // (TASK-3117). Same shape as reminderTick; started via StartDecisionTick + // and stopped by Stop() via stopDecisionTick. + decisionTick decisionTickConfig + // inFlightUploadHashes tracks content_hash values for uploads // that have called AttachmentStore.Put but not yet inserted the // attachments row. Without this, the orphan GC could delete a @@ -501,6 +506,7 @@ func (s *Server) Stop() { // Item reminder tick (IDEA-2641). Same lifecycle pattern; an in-flight // pass is tracked on s.bg and awaited below. s.stopReminderTick() + s.stopDecisionTick() // MCP audit writer / sweeper run on s.bg too. Signal first so // the workers see the close BEFORE Wait() blocks; without the // signal Wait would hang forever on the writer's blocking @@ -1892,6 +1898,10 @@ func (s *Server) setupRouter() { // about the item it names. r.Get("/reminders", s.handleListItemReminders) r.Post("/reminders", s.handleCreateItemReminder) + // Typed decisions (TASK-3117): the item's latest + // answer per question, each marked current or not. + // Read-only; evaluation happens off the write path. + r.Get("/decisions", s.handleListItemDecisions) }) // Links (v2) diff --git a/internal/store/attachments.go b/internal/store/attachments.go index ac38c6769..8b4dd1cff 100644 --- a/internal/store/attachments.go +++ b/internal/store/attachments.go @@ -1645,6 +1645,11 @@ func (s *Store) RemapAttachmentReferencesInWorkspace(workspaceID string, oldToNe // list, which is the kind of claim that goes stale unread. The workspace // seq lock this transaction already took above is the one the subquery // needs, so this costs an indexed MAX(seq) per row and no new locking. + // NOT a decision enqueue door (TASK-3117 ruling 3). This is a system + // fan-out rewrite: one import remap touches every matching item, so an enqueue + // per row would be an unbounded provider bill for one action. The rewrite + // DOES change hashed decision state, so each touched item's stored + // answers read current=false until its next direct write re-evaluates. if _, err := tx.Exec(s.q(`UPDATE items SET content = ?, fields = ?, seq = `+nextWorkspaceSeqSubquery+` WHERE id = ?`), u.content, u.fields, workspaceID, u.id); err != nil { return fmt.Errorf("update item %s: %w", u.id, err) diff --git a/internal/store/collections.go b/internal/store/collections.go index 5dc323216..138d416fd 100644 --- a/internal/store/collections.go +++ b/internal/store/collections.go @@ -936,6 +936,11 @@ func (s *Store) applyFieldMigrationsTx(tx *sql.Tx, collectionID, workspaceID str // a separate statement, so MAX(seq) advances between // them inside the transaction — every row ends up with // a unique sequential seq. + // NOT a decision enqueue door (TASK-3117 ruling 3). This is a system + // fan-out rewrite: one field migration touches every matching item, so an enqueue + // per row would be an unbounded provider bill for one action. The rewrite + // DOES change hashed decision state, so each touched item's stored + // answers read current=false until its next direct write re-evaluates. updateSQL := s.q(fmt.Sprintf(` UPDATE items SET fields = %s, diff --git a/internal/store/comments.go b/internal/store/comments.go index 33733e594..91c7c2066 100644 --- a/internal/store/comments.go +++ b/internal/store/comments.go @@ -128,6 +128,11 @@ func (s *Store) createCommentTx(tx *sql.Tx, workspaceID, itemID, userID string, if err := s.emitCommentEventTx(tx, kernelevents.CommentCreated, created); err != nil { return "", err } + // The recent trail is part of the state a decision reads, so a new + // comment makes an evaluation owed (TASK-3117). + if err := s.enqueueDecisionJobsForItemTx(tx, itemID); err != nil { + return "", err + } return id, nil } @@ -183,6 +188,14 @@ func (s *Store) UpdateComment(id, body string) (*models.Comment, error) { if err := s.emitCommentEventTx(tx, kernelevents.CommentUpdated, updated); err != nil { return nil, err } + // An edited comment can be inside the recent trail a decision reads, + // so an edit that changed the body is a door (TASK-3117 ruling 3: + // user-initiated single-item writes that change hashed state). If + // the comment is older than the trail window the runner finds the + // state unchanged and makes no call. + if err := s.enqueueDecisionJobsForItemTx(tx, updated.ItemID); err != nil { + return nil, err + } } if err := tx.Commit(); err != nil { @@ -398,6 +411,11 @@ func (s *Store) DeleteComment(id string) error { if err := s.emitRefOnlyDeletionTx(tx, kernelevents.CommentDeleted, workspaceID, id, itemID, parentID.String); err != nil { return err } + // Deleting a comment changes the recent trail (TASK-3117 ruling 3); see + // UpdateComment for why an out-of-window comment costs nothing. + if err := s.enqueueDecisionJobsForItemTx(tx, itemID); err != nil { + return err + } if err := tx.Commit(); err != nil { return fmt.Errorf("delete comment: %w", err) } diff --git a/internal/store/decisions.go b/internal/store/decisions.go new file mode 100644 index 000000000..dfdf8bbcb --- /dev/null +++ b/internal/store/decisions.go @@ -0,0 +1,506 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + "strings" + "sync/atomic" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +// Typed-decision storage and its job queue (PLAN-3114 unit 2, TASK-3117). +// +// See migration 090 for why the queue is a private table rather than the +// event outbox, and why answers are keyed on a state hash rather than +// items.seq. +// +// THE WRITE PATH NEVER WAITS ON THE NETWORK (the day-73 async-only ruling). +// A mutation that makes an evaluation owed inserts a decision_jobs row in its +// own transaction and returns; the runner's tick does the provider call later. + +// DecisionSetResolver names the question sets that apply to an item in the +// collection with this slug. The server installs one when a provider is +// configured; with no resolver installed, or one that returns nothing, the +// write doors enqueue nothing — which is the nil-provider case: no job row, no +// error. +// +// It is called INSIDE the write transaction, so it must be pure: a resolver +// that reads the database through the pool would wait on a connection the +// transaction may be holding (the MaxOpenConns(1) deadlock class). That is +// why it takes the slug — read in-tx by the store — rather than an id it +// would have to look up. +type DecisionSetResolver func(collectionSlug string) []string + +// SetDecisionSetResolver installs the resolver, or removes it with nil. +// +// Atomic, unlike the test seams on Store, because this is production wiring: +// every write door reads it, and nothing guarantees it is installed before +// the first request. +func (s *Store) SetDecisionSetResolver(r DecisionSetResolver) { + if r == nil { + s.decisionSets.Store(nil) + return + } + s.decisionSets.Store(&r) +} + +func (s *Store) decisionSetResolver() DecisionSetResolver { + p := s.decisionSets.Load() + if p == nil { + return nil + } + return *p +} + +// decisionSetsPtr is the Store field's type, named so store.go's struct can +// declare it without importing sync/atomic for one line. +type decisionSetsPtr = atomic.Pointer[DecisionSetResolver] + +// decisionTimeFormat is FIXED-WIDTH, unlike RFC3339Nano, which drops trailing +// zeros. These columns are compared and ordered as TEXT, and a variable-width +// fractional part sorts `…05.5Z` after `…05.123Z` but `…05Z` after both — +// the same hazard BUG-3037 measured on updated_at. Nanosecond width because +// two evaluations of one item inside a second are ordinary (an edit, then a +// comment), and "latest" must still mean latest. +const decisionTimeFormat = "2006-01-02T15:04:05.000000000Z" + +func decisionTime(t time.Time) string { return t.UTC().Format(decisionTimeFormat) } + +// enqueueDecisionJobsTx makes an evaluation owed for every question set that +// applies to the item, on the caller's transaction. +// +// COALESCING: one row per (item, question_set). A conflicting enqueue bumps +// generation instead of adding a row, and clears the failure history, because +// the new generation is new work. enqueued_at is NOT moved: an item edited +// continuously would otherwise sort to the back of the claim scan forever. +func (s *Store) enqueueDecisionJobsTx(tx *sql.Tx, workspaceID, itemID, collectionID string) error { + resolve := s.decisionSetResolver() + if resolve == nil { + return nil + } + slug, err := s.getCollectionSlugTx(tx, collectionID) + if err != nil { + return fmt.Errorf("enqueue decision job: read collection %s: %w", collectionID, err) + } + sets := resolve(slug) + if len(sets) == 0 { + return nil + } + ts := decisionTime(time.Now()) + for _, set := range sets { + if _, err := tx.Exec(s.q(` + INSERT INTO decision_jobs (item_id, workspace_id, question_set, generation, enqueued_at, attempts) + VALUES (?, ?, ?, 1, ?, 0) + ON CONFLICT (item_id, question_set) DO UPDATE SET + generation = decision_jobs.generation + 1, + attempts = 0, + last_error = NULL`), + itemID, workspaceID, set, ts, + ); err != nil { + return fmt.Errorf("enqueue decision job %s/%s: %w", itemID, set, err) + } + } + return nil +} + +// enqueueDecisionJobsForItemTx is the comment door's form: a comment knows its +// item but not the item's collection. +func (s *Store) enqueueDecisionJobsForItemTx(tx *sql.Tx, itemID string) error { + if s.decisionSetResolver() == nil { + return nil + } + var workspaceID, collectionID string + if err := tx.QueryRow(s.q(`SELECT workspace_id, collection_id FROM items WHERE id = ?`), itemID). + Scan(&workspaceID, &collectionID); err != nil { + return fmt.Errorf("enqueue decision job: read item %s: %w", itemID, err) + } + return s.enqueueDecisionJobsTx(tx, workspaceID, itemID, collectionID) +} + +// DecisionJob is one claimed unit of owed work. +type DecisionJob struct { + ItemID string + WorkspaceID string + QuestionSet string + // Generation is the generation this claim covers. Completing the job + // deletes it only if the row still carries this generation. + Generation int64 + Attempts int + ClaimedBy string +} + +// decisionJobClaimable is the single definition of "this job may be taken", +// used by BOTH the candidate scan and the claim UPDATE's arbiter so the two +// cannot drift (the reminderFireable lesson). Its one parameter is the current +// time: an unclaimed row, or one whose lease has expired. +// +// A job whose WORKSPACE is soft-deleted is not claimable: it is held in place +// so a workspace restore finds the work still owed (codex round 5). Nothing +// enqueues on restore — that would be a fan-out over every item — so keeping +// the row is the only thing that makes a restore resume. A purge hard-deletes +// the items, and the composite FK cascades the held rows away. +const decisionJobClaimable = `(claimed_by IS NULL OR lease_expires_at IS NULL OR lease_expires_at < ?) + AND NOT EXISTS (SELECT 1 FROM workspaces w WHERE w.id = decision_jobs.workspace_id AND w.deleted_at IS NOT NULL)` + +// ClaimDecisionJobs claims up to limit owed jobs for runner, oldest first, +// each leased until now+lease. +// +// The scan and the claim are separate statements; the claim's WHERE re-checks +// claimability AND the scanned generation, so a job another runner took, or +// one re-enqueued in between, is skipped rather than double-claimed. +func (s *Store) ClaimDecisionJobs(runner string, limit int, lease time.Duration) ([]DecisionJob, error) { + if limit <= 0 { + return nil, nil + } + nowT := time.Now() + nowS := decisionTime(nowT) + until := decisionTime(nowT.Add(lease)) + + rows, err := s.db.Query(s.q(` + SELECT item_id, workspace_id, question_set, generation, attempts + FROM decision_jobs + WHERE `+decisionJobClaimable+` + ORDER BY enqueued_at, item_id, question_set + LIMIT ?`), nowS, limit) + if err != nil { + return nil, fmt.Errorf("scan decision jobs: %w", err) + } + var candidates []DecisionJob + for rows.Next() { + var j DecisionJob + if err := rows.Scan(&j.ItemID, &j.WorkspaceID, &j.QuestionSet, &j.Generation, &j.Attempts); err != nil { + rows.Close() + return nil, fmt.Errorf("scan decision job: %w", err) + } + candidates = append(candidates, j) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + + claimed := make([]DecisionJob, 0, len(candidates)) + for _, j := range candidates { + res, err := s.db.Exec(s.q(` + UPDATE decision_jobs SET claimed_by = ?, lease_expires_at = ? + WHERE item_id = ? AND question_set = ? AND generation = ? + AND `+decisionJobClaimable), + runner, until, j.ItemID, j.QuestionSet, j.Generation, nowS) + if err != nil { + return claimed, fmt.Errorf("claim decision job %s/%s: %w", j.ItemID, j.QuestionSet, err) + } + if n, _ := res.RowsAffected(); n == 1 { + j.ClaimedBy = runner + claimed = append(claimed, j) + } + } + return claimed, nil +} + +// CompleteDecisionJob retires a claimed job. +// +// The job is deleted only if its generation is still the one claimed. If a +// write re-enqueued it mid-evaluation, the row stays and the claim is +// released, so the next tick evaluates the newer state — the finished run +// must not delete work it never saw. +func (s *Store) CompleteDecisionJob(j DecisionJob) error { + res, err := s.db.Exec(s.q(` + DELETE FROM decision_jobs + WHERE item_id = ? AND question_set = ? AND generation = ? AND claimed_by = ?`), + j.ItemID, j.QuestionSet, j.Generation, j.ClaimedBy) + if err != nil { + return fmt.Errorf("complete decision job %s/%s: %w", j.ItemID, j.QuestionSet, err) + } + if n, _ := res.RowsAffected(); n == 1 { + return nil + } + return s.releaseDecisionJob(j, "", 0) +} + +// FailDecisionJob records a failed attempt and holds the job until retryAfter +// has passed. The lease doubles as the backoff: the row stays claimed by this +// runner, with an expiry at the retry time, and becomes claimable again when +// it lapses. A job re-enqueued in the meantime is simply released — its new +// generation is new work, and its failure history was cleared by the enqueue. +func (s *Store) FailDecisionJob(j DecisionJob, cause error, retryAfter time.Duration) error { + msg := "" + if cause != nil { + msg = cause.Error() + } + return s.releaseDecisionJob(j, msg, retryAfter) +} + +// DropDecisionJob abandons a claimed job: the item is gone, the set no +// longer applies, or the job has exhausted its attempts. +// +// Generation-matched, exactly like CompleteDecisionJob (codex round 2): the +// verdict is about the generation this runner claimed. A write that landed +// mid-evaluation bumped the row to a newer generation, which is new work the +// verdict says nothing about — so the row is released for it, not deleted. +func (s *Store) DropDecisionJob(j DecisionJob) error { + res, err := s.db.Exec(s.q(` + DELETE FROM decision_jobs + WHERE item_id = ? AND question_set = ? AND generation = ? AND claimed_by = ?`), + j.ItemID, j.QuestionSet, j.Generation, j.ClaimedBy) + if err != nil { + return fmt.Errorf("drop decision job %s/%s: %w", j.ItemID, j.QuestionSet, err) + } + if n, _ := res.RowsAffected(); n == 1 { + return nil + } + return s.releaseSupersededDecisionJob(j) +} + +// ReleaseDecisionJob gives a claimed job back without recording a failure: +// the runner stopped before finishing it (shutdown), which says nothing about +// the job. It is claimable again at once. +func (s *Store) ReleaseDecisionJob(j DecisionJob) error { + return s.releaseDecisionJob(j, "", 0) +} + +func (s *Store) releaseDecisionJob(j DecisionJob, failure string, retryAfter time.Duration) error { + if failure == "" { + // Superseded: release the claim so the newer generation is claimable + // at once. + _, err := s.db.Exec(s.q(` + UPDATE decision_jobs SET claimed_by = NULL, lease_expires_at = NULL + WHERE item_id = ? AND question_set = ? AND claimed_by = ?`), + j.ItemID, j.QuestionSet, j.ClaimedBy) + if err != nil { + return fmt.Errorf("release decision job %s/%s: %w", j.ItemID, j.QuestionSet, err) + } + return nil + } + _, err := s.db.Exec(s.q(` + UPDATE decision_jobs + SET attempts = attempts + 1, last_error = ?, lease_expires_at = ? + WHERE item_id = ? AND question_set = ? AND generation = ? AND claimed_by = ?`), + truncateDecisionError(failure), decisionTime(time.Now().Add(retryAfter)), + j.ItemID, j.QuestionSet, j.Generation, j.ClaimedBy) + if err != nil { + return fmt.Errorf("fail decision job %s/%s: %w", j.ItemID, j.QuestionSet, err) + } + // Zero rows means the job was re-enqueued; the enqueue already reset its + // failure history, so release the claim for the new generation. + return s.releaseSupersededDecisionJob(j) +} + +func (s *Store) releaseSupersededDecisionJob(j DecisionJob) error { + _, err := s.db.Exec(s.q(` + UPDATE decision_jobs SET claimed_by = NULL, lease_expires_at = NULL + WHERE item_id = ? AND question_set = ? AND claimed_by = ? AND generation <> ?`), + j.ItemID, j.QuestionSet, j.ClaimedBy, j.Generation) + if err != nil { + return fmt.Errorf("release decision job %s/%s: %w", j.ItemID, j.QuestionSet, err) + } + return nil +} + +// maxDecisionErrorBytes bounds last_error: a provider error can carry a +// response body, and this column is diagnostics, not a log. +const maxDecisionErrorBytes = 1024 + +func truncateDecisionError(s string) string { + if len(s) <= maxDecisionErrorBytes { + return s + } + return s[:maxDecisionErrorBytes] +} + +// DecisionJobState is a test/diagnostic read of one job row. +type DecisionJobState struct { + Generation int64 + Attempts int + ClaimedBy string + LastError string +} + +// GetDecisionJob returns a job's row, or nil when no evaluation is owed. +func (s *Store) GetDecisionJob(itemID, questionSet string) (*DecisionJobState, error) { + var st DecisionJobState + var claimedBy, lastError sql.NullString + err := s.db.QueryRow(s.q(` + SELECT generation, attempts, claimed_by, last_error FROM decision_jobs + WHERE item_id = ? AND question_set = ?`), itemID, questionSet). + Scan(&st.Generation, &st.Attempts, &claimedBy, &lastError) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("get decision job: %w", err) + } + st.ClaimedBy, st.LastError = claimedBy.String, lastError.String + return &st, nil +} + +// HasItemDecisionsAtState reports whether every asked question already has an +// answer for this item, set and state hash — the runner's idempotency check. +// keys maps each asked question key to its current question hash (model plus +// definition): an answer to an older wording or model does not count. +func (s *Store) HasItemDecisionsAtState(itemID, questionSet, stateHash string, keys map[string]string) (bool, error) { + if len(keys) == 0 { + return true, nil + } + // Counted over the ASKED (key, question_hash) pairs only (codex rounds 3 + // and 6). Counting every stored key was wrong in a way the count cannot + // see: a RENAMED key keeps the total equal — the old name answered, the + // new one not — so the set read complete and the new key was never asked. + // Matching the question hash as well is what makes a reworded question or + // a new model re-ask at an unchanged item state. + pairs := make([]string, 0, len(keys)) + args := []any{itemID, questionSet, stateHash} + for k, qh := range keys { + pairs = append(pairs, "(question_key = ? AND question_hash = ?)") + args = append(args, k, qh) + } + var n int + if err := s.db.QueryRow(s.q(` + SELECT COUNT(DISTINCT question_key) FROM item_decisions + WHERE item_id = ? AND question_set = ? AND state_hash = ? + AND (`+strings.Join(pairs, " OR ")+`)`), args...).Scan(&n); err != nil { + return false, fmt.Errorf("check item decisions: %w", err) + } + return n == len(keys), nil +} + +// InsertItemDecisions stores one evaluation's answers in one transaction. +// A row already present for the same (item, set, key, state) is kept, not +// replaced: two runners that evaluated the same state concurrently stored +// equivalent answers, and the first one is the record. +func (s *Store) InsertItemDecisions(workspaceID string, rows []models.ItemDecision) error { + if len(rows) == 0 { + return nil + } + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck // no-op after commit + for _, r := range rows { + truncated := 0 + if r.StateTruncated { + truncated = 1 + } + evaluated := r.EvaluatedAt + if evaluated == "" { + evaluated = decisionTime(time.Now()) + } + if _, err := tx.Exec(s.q(` + INSERT INTO item_decisions + (id, workspace_id, item_id, question_set, question_key, kind, answer, confidence, + provider, model, state_hash, question_hash, item_seq, state_truncated, evaluated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (item_id, question_set, question_key, state_hash, question_hash) DO NOTHING`), + newID(), workspaceID, r.ItemID, r.QuestionSet, r.QuestionKey, r.Kind, string(r.Answer), r.Confidence, + r.Provider, r.Model, r.StateHash, r.QuestionHash, r.ItemSeq, truncated, evaluated, + ); err != nil { + return fmt.Errorf("insert item decision %s/%s: %w", r.QuestionSet, r.QuestionKey, err) + } + } + return tx.Commit() +} + +// LatestItemDecisions returns the newest answer per (question_set, +// question_key) for an item, ordered by set then key. Current is left false; +// only the caller can compute the item's present state hash. +// +// Older rows stay in the table for the audit; this read hides them. +func (s *Store) LatestItemDecisions(itemID string) ([]models.ItemDecision, error) { + rows, err := s.db.Query(s.q(` + SELECT d.id, d.item_id, d.question_set, d.question_key, d.kind, d.answer, d.confidence, + d.provider, d.model, d.state_hash, d.question_hash, d.item_seq, d.state_truncated, d.evaluated_at + FROM item_decisions d + WHERE d.item_id = ? + AND NOT EXISTS ( + SELECT 1 FROM item_decisions n + WHERE n.item_id = d.item_id AND n.question_set = d.question_set AND n.question_key = d.question_key + AND (n.evaluated_at > d.evaluated_at OR (n.evaluated_at = d.evaluated_at AND n.id > d.id)) + ) + ORDER BY d.question_set, d.question_key`), itemID) + if err != nil { + return nil, fmt.Errorf("list item decisions: %w", err) + } + defer rows.Close() + var out []models.ItemDecision + for rows.Next() { + var d models.ItemDecision + var answer string + var conf sql.NullFloat64 + var truncated int + if err := rows.Scan(&d.ID, &d.ItemID, &d.QuestionSet, &d.QuestionKey, &d.Kind, &answer, &conf, + &d.Provider, &d.Model, &d.StateHash, &d.QuestionHash, &d.ItemSeq, &truncated, &d.EvaluatedAt); err != nil { + return nil, fmt.Errorf("scan item decision: %w", err) + } + d.Answer = []byte(answer) + if conf.Valid { + c := conf.Float64 + d.Confidence = &c + } + d.StateTruncated = truncated != 0 + out = append(out, d) + } + return out, rows.Err() +} + +// RecentComments returns an item's newest n comments, oldest first — the +// "recent trail" a decision's state carries. Ordered by (created_at, id) so a +// burst of comments inside one second still has one order, and therefore one +// state hash. +func (s *Store) RecentComments(itemID string, n int) ([]models.Comment, error) { + if n <= 0 { + return nil, nil + } + rows, err := s.db.Query(s.q(` + SELECT `+commentListCols+` + FROM comments c + `+commentAgentJoin+` + WHERE c.item_id = ? + ORDER BY c.created_at DESC, c.id DESC + LIMIT ?`), itemID, n) + if err != nil { + return nil, fmt.Errorf("recent comments: %w", err) + } + defer rows.Close() + out, err := scanComments(rows) + if err != nil { + return nil, err + } + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, nil +} + +// ItemLiveness reports whether an item may be sent to a decision provider, +// in two parts because the runner treats them differently: +// +// - itemLive: the item and its collection are not soft-deleted. False means +// the job is dropped — neither has a restore path the queue must survive +// (an item restore re-enqueues through its own door). +// - workspaceLive: its workspace is not soft-deleted. False means the job is +// HELD for a possible workspace restore. +// +// GetItem filters only the ITEM's soft delete, so this exists to stop content +// its owner deleted — by deleting the workspace or collection — from reaching +// a third-party provider. Same liveness the reminder tick's reminderFireable +// enforces, for the same reason. +func (s *Store) ItemLiveness(itemID string) (itemLive, workspaceLive bool, err error) { + var itemDeleted, collDeleted, wsDeleted sql.NullString + err = s.db.QueryRow(s.q(` + SELECT i.deleted_at, c.deleted_at, w.deleted_at FROM items i + JOIN collections c ON c.id = i.collection_id + JOIN workspaces w ON w.id = i.workspace_id + WHERE i.id = ?`), itemID).Scan(&itemDeleted, &collDeleted, &wsDeleted) + if errors.Is(err, sql.ErrNoRows) { + return false, false, nil + } + if err != nil { + return false, false, fmt.Errorf("item liveness: %w", err) + } + return !itemDeleted.Valid && !collDeleted.Valid, !wsDeleted.Valid, nil +} diff --git a/internal/store/decisions_test.go b/internal/store/decisions_test.go new file mode 100644 index 000000000..4db92070a --- /dev/null +++ b/internal/store/decisions_test.go @@ -0,0 +1,418 @@ +package store + +import ( + "errors" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +// Typed-decision queue and storage (TASK-3117). Runs on both backends via +// testStore (PAD_TEST_POSTGRES_URL selects Postgres). + +func decisionFixture(t *testing.T) (*Store, *models.Workspace, *models.Collection) { + t.Helper() + s := testStore(t) + ws := createTestWorkspace(t, s, "Decisions") + col := createTestCollection(t, s, ws.ID, "Tasks") + return s, ws, col +} + +func countDecisionJobs(t *testing.T, s *Store) int { + t.Helper() + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM decision_jobs`).Scan(&n); err != nil { + t.Fatalf("count decision_jobs: %v", err) + } + return n +} + +func mustJob(t *testing.T, s *Store, itemID, set string) *DecisionJobState { + t.Helper() + j, err := s.GetDecisionJob(itemID, set) + if err != nil { + t.Fatalf("get job: %v", err) + } + if j == nil { + t.Fatalf("no decision job owed for %s/%s", itemID, set) + } + return j +} + +// The nil-provider case: no resolver installed means every door writes nothing. +// The doors are exercised for real, so a pass is not a door that never ran. +func TestDecisionJobs_NoResolverEnqueuesNothing(t *testing.T) { + s, ws, col := decisionFixture(t) + item := createTestItem(t, s, ws.ID, col.ID, "Quiet", "body") + title := "Quiet, renamed" + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{Title: &title}); err != nil { + t.Fatalf("update: %v", err) + } + if _, err := s.CreateComment(ws.ID, item.ID, "", models.CommentCreate{Author: "a", Body: "hi"}); err != nil { + t.Fatalf("comment: %v", err) + } + if n := countDecisionJobs(t, s); n != 0 { + t.Fatalf("decision_jobs has %d rows with no resolver installed; want 0", n) + } + + // Positive control on the same fixture: installing a resolver makes the + // very next write enqueue, so the zero above is about the resolver. + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{Title: &title}); err != nil { + t.Fatalf("update: %v", err) + } + if n := countDecisionJobs(t, s); n != 1 { + t.Fatalf("after installing a resolver, decision_jobs has %d rows; want 1", n) + } +} + +// Each of the three doors enqueues, and repeats coalesce into ONE row whose +// generation counts the writes. +func TestDecisionJobs_EachDoorEnqueuesAndCoalesces(t *testing.T) { + s, ws, col := decisionFixture(t) + s.SetDecisionSetResolver(func(slug string) []string { + if slug != col.Slug { + t.Errorf("resolver got collection slug %q, want %q", slug, col.Slug) + } + return []string{"triage"} + }) + + item := createTestItem(t, s, ws.ID, col.ID, "Door", "body") + if g := mustJob(t, s, item.ID, "triage").Generation; g != 1 { + t.Fatalf("after create, generation = %d; want 1", g) + } + + title := "Door 2" + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{Title: &title}); err != nil { + t.Fatalf("update: %v", err) + } + if g := mustJob(t, s, item.ID, "triage").Generation; g != 2 { + t.Fatalf("after update, generation = %d; want 2", g) + } + + if _, err := s.CreateComment(ws.ID, item.ID, "", models.CommentCreate{Author: "a", Body: "note"}); err != nil { + t.Fatalf("comment: %v", err) + } + if g := mustJob(t, s, item.ID, "triage").Generation; g != 3 { + t.Fatalf("after comment, generation = %d; want 3", g) + } + if n := countDecisionJobs(t, s); n != 1 { + t.Fatalf("three writes left %d job rows; want 1 (coalesced)", n) + } +} + +func TestDecisionJobs_ResolverScopesByCollection(t *testing.T) { + s, ws, col := decisionFixture(t) + other := createTestCollection(t, s, ws.ID, "Ideas") + s.SetDecisionSetResolver(func(slug string) []string { + if slug == col.Slug { + return []string{"triage"} + } + return nil + }) + createTestItem(t, s, ws.ID, other.ID, "Out of scope", "") + if n := countDecisionJobs(t, s); n != 0 { + t.Fatalf("an item in an unscoped collection enqueued %d jobs; want 0", n) + } + createTestItem(t, s, ws.ID, col.ID, "In scope", "") + if n := countDecisionJobs(t, s); n != 1 { + t.Fatalf("an item in the scoped collection enqueued %d jobs; want 1", n) + } +} + +// A claim is exclusive until its lease lapses, and then any runner may take it. +func TestDecisionJobs_ClaimIsExclusiveUntilLeaseLapses(t *testing.T) { + s, ws, col := decisionFixture(t) + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + item := createTestItem(t, s, ws.ID, col.ID, "Claim", "") + + a, err := s.ClaimDecisionJobs("runner-a", 10, time.Hour) + if err != nil || len(a) != 1 { + t.Fatalf("runner-a claimed %d (err %v); want 1", len(a), err) + } + b, err := s.ClaimDecisionJobs("runner-b", 10, time.Hour) + if err != nil || len(b) != 0 { + t.Fatalf("runner-b claimed %d under a live lease (err %v); want 0", len(b), err) + } + + // A dead runner: its lease is already in the past. + if _, err := s.db.Exec(s.q(`UPDATE decision_jobs SET lease_expires_at = ? WHERE item_id = ?`), + decisionTime(time.Now().Add(-time.Minute)), item.ID); err != nil { + t.Fatal(err) + } + b, err = s.ClaimDecisionJobs("runner-b", 10, time.Hour) + if err != nil || len(b) != 1 { + t.Fatalf("runner-b claimed %d after the lease lapsed (err %v); want 1", len(b), err) + } + // The dead runner's late completion must not delete runner-b's job. + if err := s.CompleteDecisionJob(a[0]); err != nil { + t.Fatal(err) + } + if j := mustJob(t, s, item.ID, "triage"); j.ClaimedBy != "runner-b" { + t.Fatalf("after the stale completion the job is claimed by %q; want runner-b", j.ClaimedBy) + } +} + +// A write that lands mid-evaluation leaves the job owed: completion of the +// older generation releases rather than deletes. +func TestDecisionJobs_ReenqueueDuringEvaluationSurvivesCompletion(t *testing.T) { + s, ws, col := decisionFixture(t) + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + item := createTestItem(t, s, ws.ID, col.ID, "Race", "") + + claimed, err := s.ClaimDecisionJobs("r", 10, time.Hour) + if err != nil || len(claimed) != 1 { + t.Fatalf("claim: %d, %v", len(claimed), err) + } + title := "Race, edited mid-evaluation" + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{Title: &title}); err != nil { + t.Fatal(err) + } + if err := s.CompleteDecisionJob(claimed[0]); err != nil { + t.Fatal(err) + } + j := mustJob(t, s, item.ID, "triage") + if j.Generation != 2 || j.ClaimedBy != "" { + t.Fatalf("after completing gen 1 under a gen-2 enqueue: generation=%d claimed_by=%q; want 2, unclaimed", + j.Generation, j.ClaimedBy) + } + again, err := s.ClaimDecisionJobs("r", 10, time.Hour) + if err != nil || len(again) != 1 || again[0].Generation != 2 { + t.Fatalf("the newer generation is not claimable at once: %+v, %v", again, err) + } + + // Control: with no intervening write, completion deletes. + if err := s.CompleteDecisionJob(again[0]); err != nil { + t.Fatal(err) + } + if n := countDecisionJobs(t, s); n != 0 { + t.Fatalf("completing the current generation left %d rows; want 0", n) + } +} + +func TestDecisionJobs_FailureBacksOff(t *testing.T) { + s, ws, col := decisionFixture(t) + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + item := createTestItem(t, s, ws.ID, col.ID, "Fail", "") + + claimed, _ := s.ClaimDecisionJobs("r", 10, time.Hour) + if err := s.FailDecisionJob(claimed[0], errors.New("provider 503"), time.Hour); err != nil { + t.Fatal(err) + } + j := mustJob(t, s, item.ID, "triage") + if j.Attempts != 1 || j.LastError != "provider 503" { + t.Fatalf("after one failure: attempts=%d last_error=%q", j.Attempts, j.LastError) + } + if again, _ := s.ClaimDecisionJobs("r2", 10, time.Hour); len(again) != 0 { + t.Fatalf("a job inside its backoff was claimed") + } + + // A new write is new work: it resets the failure history. + title := "Fail, edited" + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{Title: &title}); err != nil { + t.Fatal(err) + } + if j := mustJob(t, s, item.ID, "triage"); j.Attempts != 0 || j.LastError != "" { + t.Fatalf("an enqueue did not reset failure history: attempts=%d last_error=%q", j.Attempts, j.LastError) + } +} + +// The read returns one row per question — the newest — and keeps the rest. +func TestItemDecisions_LatestHidesOlderStates(t *testing.T) { + s, ws, col := decisionFixture(t) + item := createTestItem(t, s, ws.ID, col.ID, "Read", "") + + row := func(key, hash, at string) models.ItemDecision { + return models.ItemDecision{ + ItemID: item.ID, QuestionSet: "triage", QuestionKey: key, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.5}`), Provider: "fake", Model: "m", + StateHash: hash, ItemSeq: 1, EvaluatedAt: at, + } + } + t0 := time.Now().UTC() + // Two evaluations inside the SAME second: the fixed-width nanosecond + // format is what orders them. + older := decisionTime(t0) + newer := decisionTime(t0.Add(time.Millisecond)) + if err := s.InsertItemDecisions(ws.ID, []models.ItemDecision{row("urgent", "h1", older), row("stale", "h1", older)}); err != nil { + t.Fatal(err) + } + if err := s.InsertItemDecisions(ws.ID, []models.ItemDecision{row("urgent", "h2", newer)}); err != nil { + t.Fatal(err) + } + // A duplicate of an existing state is kept, not replaced. + if err := s.InsertItemDecisions(ws.ID, []models.ItemDecision{row("urgent", "h2", decisionTime(t0.Add(time.Second)))}); err != nil { + t.Fatal(err) + } + + got, err := s.LatestItemDecisions(item.ID) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 { + t.Fatalf("latest read returned %d rows; want 2 (one per key): %+v", len(got), got) + } + byKey := map[string]models.ItemDecision{} + for _, d := range got { + byKey[d.QuestionKey] = d + } + if byKey["urgent"].StateHash != "h2" || byKey["urgent"].EvaluatedAt != newer { + t.Fatalf("urgent: got hash %q at %q; want h2 at %q", byKey["urgent"].StateHash, byKey["urgent"].EvaluatedAt, newer) + } + if byKey["stale"].StateHash != "h1" { + t.Fatalf("stale: got hash %q; want h1", byKey["stale"].StateHash) + } + var total int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM item_decisions`).Scan(&total); err != nil { + t.Fatal(err) + } + if total != 3 { + t.Fatalf("item_decisions holds %d rows; want 3 (older state kept for the audit, duplicate refused)", total) + } + if ok, _ := s.HasItemDecisionsAtState(item.ID, "triage", "h2", map[string]string{"urgent": "", "stale": ""}); ok { + t.Fatal("HasItemDecisionsAtState reported a complete set at h2, which has only one of two keys") + } + if ok, _ := s.HasItemDecisionsAtState(item.ID, "triage", "h1", map[string]string{"urgent": "", "stale": ""}); !ok { + t.Fatal("HasItemDecisionsAtState missed the complete set at h1") + } +} + +// Move is a door (TASK-3117 ruling 3), keyed on the TARGET collection's sets: +// a move INTO a scoped collection makes an evaluation owed where the item's +// old collection owed none. +func TestDecisionJobs_MoveEnqueuesForTargetCollection(t *testing.T) { + s, ws, col := decisionFixture(t) + other := createTestCollection(t, s, ws.ID, "Ideas") + s.SetDecisionSetResolver(func(slug string) []string { + if slug == col.Slug { + return []string{"triage"} + } + return nil + }) + item := createTestItem(t, s, ws.ID, other.ID, "Mover", "") + if n := countDecisionJobs(t, s); n != 0 { + t.Fatalf("precondition: item in an unscoped collection enqueued %d jobs", n) + } + if _, err := s.MoveItem(item.ID, col.ID, item.Fields); err != nil { + t.Fatalf("move: %v", err) + } + if j := mustJob(t, s, item.ID, "triage"); j.Generation != 1 { + t.Fatalf("after move into the scoped collection, generation = %d; want 1", j.Generation) + } +} + +// Comment edit and delete change the trail, so they are doors too (ruling 3's +// rule applied). An edit that leaves the body unchanged is not. +func TestDecisionJobs_CommentEditAndDeleteEnqueue(t *testing.T) { + s, ws, col := decisionFixture(t) + item := createTestItem(t, s, ws.ID, col.ID, "Trail", "") + c, err := s.CreateComment(ws.ID, item.ID, "", models.CommentCreate{Author: "a", Body: "first"}) + if err != nil { + t.Fatal(err) + } + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + + if _, err := s.UpdateComment(c.ID, "first"); err != nil { + t.Fatal(err) + } + if j, _ := s.GetDecisionJob(item.ID, "triage"); j != nil { + t.Fatalf("a no-op comment edit enqueued: %+v", j) + } + if _, err := s.UpdateComment(c.ID, "first, corrected"); err != nil { + t.Fatal(err) + } + if g := mustJob(t, s, item.ID, "triage").Generation; g != 1 { + t.Fatalf("after a comment edit, generation = %d; want 1", g) + } + if err := s.DeleteComment(c.ID); err != nil { + t.Fatal(err) + } + if g := mustJob(t, s, item.ID, "triage").Generation; g != 2 { + t.Fatalf("after a comment delete, generation = %d; want 2", g) + } +} + +// Codex round 2 P1: dropping a claim (retry limit, gone item) must not take a +// NEWER generation with it — a write that landed mid-evaluation is still owed. +func TestDecisionJobs_DropPreservesANewerGeneration(t *testing.T) { + s, ws, col := decisionFixture(t) + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + item := createTestItem(t, s, ws.ID, col.ID, "Drop", "") + claimed, err := s.ClaimDecisionJobs("r", 10, time.Hour) + if err != nil || len(claimed) != 1 { + t.Fatalf("claim: %d %v", len(claimed), err) + } + title := "Drop, edited mid-evaluation" + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{Title: &title}); err != nil { + t.Fatal(err) + } + if err := s.DropDecisionJob(claimed[0]); err != nil { + t.Fatal(err) + } + j := mustJob(t, s, item.ID, "triage") + if j.Generation != 2 || j.ClaimedBy != "" { + t.Fatalf("after dropping gen 1 under a gen-2 enqueue: %+v; want generation 2, unclaimed", j) + } + + // Control: with no intervening write the drop removes the row. + again, _ := s.ClaimDecisionJobs("r", 10, time.Hour) + if err := s.DropDecisionJob(again[0]); err != nil { + t.Fatal(err) + } + if n := countDecisionJobs(t, s); n != 0 { + t.Fatalf("dropping the current generation left %d rows", n) + } +} + +// Codex round 3: a set whose keys CHANGED at an unchanged state is not +// complete. A rename keeps the count equal (old key stored, new key asked), +// so a count over every stored key would report it complete and the new key +// would never be answered. +func TestItemDecisions_HasStateCountsOnlyTheAskedKeys(t *testing.T) { + s, ws, col := decisionFixture(t) + item := createTestItem(t, s, ws.ID, col.ID, "Keys", "") + row := func(key string) models.ItemDecision { + return models.ItemDecision{ItemID: item.ID, QuestionSet: "triage", QuestionKey: key, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.5}`), Provider: "fake", Model: "m", StateHash: "h", ItemSeq: 1} + } + if err := s.InsertItemDecisions(ws.ID, []models.ItemDecision{row("urgent"), row("old_name")}); err != nil { + t.Fatal(err) + } + if ok, err := s.HasItemDecisionsAtState(item.ID, "triage", "h", map[string]string{"urgent": "", "new_name": ""}); err != nil || ok { + t.Fatalf("renamed key: HasItemDecisionsAtState = %v, %v; want false (new_name has no answer)", ok, err) + } + if ok, _ := s.HasItemDecisionsAtState(item.ID, "triage", "h", map[string]string{"urgent": ""}); !ok { + t.Fatal("control: a set whose every asked key is stored must read complete") + } + // Codex round 6: the same key under a different question hash (a reworded + // question, or a new model) is not an answer. + if ok, _ := s.HasItemDecisionsAtState(item.ID, "triage", "h", map[string]string{"urgent": "reworded"}); ok { + t.Fatal("an answer to an older question definition counted as an answer to the new one") + } +} + +// Codex round 4: restore is a door. A job owed at delete time is dropped as +// gone; without an enqueue on restore the item would come back with stale +// answers and nothing scheduled. +func TestDecisionJobs_RestoreEnqueues(t *testing.T) { + s, ws, col := decisionFixture(t) + s.SetDecisionSetResolver(func(string) []string { return []string{"triage"} }) + item := createTestItem(t, s, ws.ID, col.ID, "Phoenix", "") + claimed, _ := s.ClaimDecisionJobs("r", 10, time.Hour) + if err := s.DropDecisionJob(claimed[0]); err != nil { // as the runner does for a gone item + t.Fatal(err) + } + if err := s.DeleteItem(item.ID); err != nil { + t.Fatal(err) + } + if n := countDecisionJobs(t, s); n != 0 { + t.Fatalf("precondition: %d jobs before restore", n) + } + if _, err := s.RestoreItem(item.ID); err != nil { + t.Fatal(err) + } + if j := mustJob(t, s, item.ID, "triage"); j.Generation != 1 { + t.Fatalf("after restore, generation = %d; want 1", j.Generation) + } +} diff --git a/internal/store/export.go b/internal/store/export.go index 74cd08daf..4941fec3b 100644 --- a/internal/store/export.go +++ b/internal/store/export.go @@ -863,6 +863,14 @@ func (s *Store) ImportWorkspace(data *models.WorkspaceExport, newName string, ow // import observable, the spec pre-names the shape — ONE additive // workspace-level `workspace.imported` event, never per-item fan-out. // Do not "fix" this by hanging an outbox write off this INSERT. + // + // The same reasoning keeps import OFF the typed-decision queue + // (TASK-3117): an enqueue per imported row would be one provider call + // per item of the archive — the fan-out ruling 3 declines for the + // title-rename cascade and field migrations. Unlike those, an imported + // item has NO answers rather than stale ones until its next direct + // write enqueues it. Whether import should enqueue is an open policy + // question on TASK-3117, not something to change silently here. _, err := tx.Exec(s.q(` INSERT INTO items (id, workspace_id, collection_id, title, slug, content, fields, tags, pinned, sort_order, parent_id, created_by, last_modified_by, source, item_number, created_at, updated_at, seq) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?, ?, ?, ?, `+nextWorkspaceSeqSubquery+`)`), diff --git a/internal/store/items.go b/internal/store/items.go index b6f8754a5..e5babc0fa 100644 --- a/internal/store/items.go +++ b/internal/store/items.go @@ -609,6 +609,11 @@ func (s *Store) createItemTxWithID(tx *sql.Tx, id, workspaceID, collectionID str if err := s.emitItemEventTx(tx, kernelevents.ItemCreated, item, nil, ""); err != nil { return nil, err } + // Same transaction, same reason: the owed evaluation commits with the + // item or not at all (TASK-3117). No-op when no provider is configured. + if err := s.enqueueDecisionJobsTx(tx, item.WorkspaceID, item.ID, item.CollectionID); err != nil { + return nil, err + } return item, nil } @@ -3038,6 +3043,12 @@ func (s *Store) updateItemWithParentLinkOnce( if err := s.emitItemUpdateEventsTx(tx, existing, updated, mutSignal.StatusChanged, mutSignal.FromStatus, doneKey, opt.batchID, hierarchyChanged); err != nil { return nil, err } + // Enqueued on EVERY update, including ones that change nothing a + // question can see: whether the state changed is the runner's state-hash + // check, not a guess made here (TASK-3117). No-op with no provider. + if err := s.enqueueDecisionJobsTx(tx, updated.WorkspaceID, updated.ID, updated.CollectionID); err != nil { + return nil, err + } // Routed through the seam so a test can reproduce the ONE commit outcome // this path must survive and cannot otherwise be shown: a transaction that @@ -3261,6 +3272,13 @@ func (s *Store) restoreItemOnce(id string, opt mutationOptions) (*models.Item, e if err := s.emitItemEventTx(tx, kernelevents.ItemRestored, restored, nil, opt.batchID); err != nil { return nil, err } + // Restore is a door (codex round 4): a job owed when the item was + // deleted was dropped as gone, so without this the item returns with + // stale answers and nothing scheduled. If nothing changed while it was + // away, the runner finds its old answers current and makes no call. + if err := s.enqueueDecisionJobsTx(tx, restored.WorkspaceID, restored.ID, restored.CollectionID); err != nil { + return nil, err + } } if err := tx.Commit(); err != nil { @@ -5207,6 +5225,12 @@ func (s *Store) moveItemWithPreCheckOnce( return nil, err } } + // A move changes the collection, which is both hashed decision state + // and what selects the question sets, so it is an enqueue door + // (TASK-3117 ruling 3). Keyed on the TARGET collection's sets. + if err := s.enqueueDecisionJobsTx(tx, moved.WorkspaceID, moved.ID, moved.CollectionID); err != nil { + return nil, err + } } if err := tx.Commit(); err != nil { diff --git a/internal/store/migrations/090_item_decisions.sql b/internal/store/migrations/090_item_decisions.sql new file mode 100644 index 000000000..455024ce9 --- /dev/null +++ b/internal/store/migrations/090_item_decisions.sql @@ -0,0 +1,113 @@ +-- Migration 090: typed-decision storage and its job queue (PLAN-3114 unit 2, +-- TASK-3117). +-- +-- Two tables. item_decisions holds the provider's answers; decision_jobs is +-- the queue that says an item owes an evaluation. Both are keyed on the item +-- through the composite (item_id, workspace_id) foreign key migration 086 +-- introduced, so a denormalized workspace_id cannot disagree with its item. +-- +-- WHY A PRIVATE QUEUE AND NOT THE EVENT OUTBOX (TASK-3117 ruling 1). The +-- outbox's event names are the closed, PUBLIC events/1 contract that webhooks +-- deliver (internal/kernelevents), and the outbox has exactly one consumer: +-- one claimed_by, one dispatched_at, and a drain that acks every row at once +-- when no webhook dispatcher is configured. An internal job riding it would +-- have had to become a public event name, and would have been acked out from +-- under the runner on every instance with webhooks off. decision_jobs borrows +-- the outbox's DISCIPLINE instead: the job row is written in the same +-- transaction as the mutation that makes it owed, and claimed by lease. + +-- WHY A STATE HASH AND NOT THE ITEM'S SEQ (TASK-3117 ruling 2). items.seq is +-- the wrong freshness key in both directions. A comment does not bump it +-- (store/comments.go writes no seq), so a comment-triggered evaluation keyed +-- on seq would find its own previous row and do nothing. And writes the +-- questions cannot see DO bump it — role-board reorders, wiki-link +-- re-resolution, collection-wide rewrites — so each would cost a provider +-- call. state_hash is sha256 of the exact bytes sent to the provider, which is +-- precisely "did anything the model reads change". item_seq is kept for the +-- audit: which version of the row the answer was computed against. +CREATE TABLE IF NOT EXISTS item_decisions ( + id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + item_id TEXT NOT NULL, + + -- The registered question set this answer came from, and the key of the + -- question within it. Both are server-registered names, never caller text. + question_set TEXT NOT NULL, + question_key TEXT NOT NULL, + + -- decision.Kind: choice / score / noul. + kind TEXT NOT NULL, + + -- The answer as JSON, marshalled by the server from decision.Answer. + -- TEXT on both dialects, deliberately not JSONB: nothing in SQL reads + -- inside it, and a jsonb column would be a second parser that refuses + -- what the first one wrote (the 22P05 class migration 084 exists for). + answer TEXT NOT NULL, + + -- NULL where the primitive returns none — a Noul answer carries no + -- confidence, and 0.0 would read as "certainly not". + confidence REAL, + + provider TEXT NOT NULL, + model TEXT NOT NULL, + state_hash TEXT NOT NULL, + + -- sha256 of the provider's model and THIS question's definition (kind, + -- instructions, criteria). state_hash says what the model read; this says + -- what it was asked and by which model. An answer is current only when + -- both still match — rewording a question or changing the pinned model at + -- an unchanged item state must re-ask (TASK-3117 codex round 6). Per + -- question, so editing one question re-asks the set but invalidates only + -- that key's answer. + question_hash TEXT NOT NULL, + item_seq INTEGER NOT NULL, + + -- 1 when the provider truncated the state to fit its budget: the answer + -- was computed from less than the item held. + state_truncated INTEGER NOT NULL DEFAULT 0, + + evaluated_at TEXT NOT NULL, + + FOREIGN KEY (item_id, workspace_id) REFERENCES items(id, workspace_id) ON DELETE CASCADE +); + +-- Idempotency: one answer per question per (state, question definition, +-- model). A second evaluation of an unchanged triple finds this row and makes +-- no call. +CREATE UNIQUE INDEX IF NOT EXISTS idx_item_decisions_state + ON item_decisions(item_id, question_set, question_key, state_hash, question_hash); + +-- The read path: latest row per (set, key) for one item. +CREATE INDEX IF NOT EXISTS idx_item_decisions_latest + ON item_decisions(item_id, question_set, question_key, evaluated_at); + +-- The queue. ONE ROW PER (item, question_set): a burst of writes to one item +-- coalesces into a single owed evaluation rather than one call per write. +-- +-- generation is what makes coalescing safe against a job that is already +-- running. Every enqueue bumps it; the runner records the generation it +-- claimed and deletes the row only if that is still the generation. A write +-- that lands mid-evaluation therefore leaves the row behind, and the next +-- tick evaluates the newer state — rather than the finished run deleting a +-- job it never saw. +-- +-- claimed_by / lease_expires_at: a claim is honored until the lease expires, +-- after which any runner may take the row. A crashed runner strands nothing. +CREATE TABLE IF NOT EXISTS decision_jobs ( + item_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + question_set TEXT NOT NULL, + generation INTEGER NOT NULL DEFAULT 1, + enqueued_at TEXT NOT NULL, + claimed_by TEXT, + lease_expires_at TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + + PRIMARY KEY (item_id, question_set), + FOREIGN KEY (item_id, workspace_id) REFERENCES items(id, workspace_id) ON DELETE CASCADE +); + +-- The claim scan: oldest owed job first. +CREATE INDEX IF NOT EXISTS idx_decision_jobs_enqueued + ON decision_jobs(enqueued_at); diff --git a/internal/store/nul_unprotected_baseline.txt b/internal/store/nul_unprotected_baseline.txt index e0131d2c3..22fb8f376 100644 --- a/internal/store/nul_unprotected_baseline.txt +++ b/internal/store/nul_unprotected_baseline.txt @@ -58,6 +58,11 @@ custom_templates.created_at custom_templates.id custom_templates.updated_at custom_templates.workspace_id +decision_jobs.enqueued_at +decision_jobs.item_id +decision_jobs.lease_expires_at +decision_jobs.question_set +decision_jobs.workspace_id documents.created_at documents.deleted_at documents.id @@ -87,6 +92,17 @@ item_collection_moves.id item_collection_moves.item_id item_collection_moves.to_collection_id item_collection_moves.workspace_id +item_decisions.evaluated_at +item_decisions.id +item_decisions.item_id +item_decisions.kind +item_decisions.model +item_decisions.provider +item_decisions.question_hash +item_decisions.question_key +item_decisions.question_set +item_decisions.state_hash +item_decisions.workspace_id item_grants.created_at item_grants.id item_grants.item_id diff --git a/internal/store/nulcolumns.go b/internal/store/nulcolumns.go index 5f8d17706..6578ecda6 100644 --- a/internal/store/nulcolumns.go +++ b/internal/store/nulcolumns.go @@ -318,6 +318,9 @@ var nulExcluded = map[string]string{ "mcp_audit_log.error_kind": "server enum, mcp_audit.go", "users.recovery_codes": "newline-joined bcrypt hashes of server-generated codes; looks like JSON, is not", "workspace_members.collection_access": "validated enum all/selected", + "decision_jobs.claimed_by": "runner id minted by the server's decision tick, never request-derived", + "decision_jobs.last_error": "Go error string, server-composed", + "item_decisions.answer": "json.Marshal of decision.Answer: a NUL anywhere in it is written as the six-byte escape, never a raw byte, and the column is TEXT on both dialects so no jsonb parser decodes it (migration 090)", "item_yjs_updates.update_data": "BINARY (BLOB/BYTEA), the only such column in either schema. Raw Yjs updates legitimately contain NUL bytes; Layer A exempts it for the same reason and TestBinaryColumnCensus pins that. Surfaced here when the census's type filter was widened to include BLOB affinity, which is correct — the decision to exclude it is a judgement, not an oversight.", } diff --git a/internal/store/pgmigrations/067_item_decisions.sql b/internal/store/pgmigrations/067_item_decisions.sql new file mode 100644 index 000000000..dd212d368 --- /dev/null +++ b/internal/store/pgmigrations/067_item_decisions.sql @@ -0,0 +1,53 @@ +-- Typed-decision storage and its job queue — Postgres mirror of +-- internal/store/migrations/090_item_decisions.sql. See the SQLite migration +-- for the rationale (why a private queue rather than the event outbox; why a +-- state hash rather than items.seq; why generation exists; why answer is TEXT). +-- +-- Dialect notes: timestamps stay TEXT, matching every other table here and +-- the lexical comparison the claim predicate depends on. confidence is DOUBLE +-- PRECISION, the Postgres spelling of SQLite's REAL. answer is TEXT, not +-- JSONB: no SQL reads inside it, and jsonb would refuse an escape the Go +-- marshaller legitimately writes. +CREATE TABLE IF NOT EXISTS item_decisions ( + id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + item_id TEXT NOT NULL, + question_set TEXT NOT NULL, + question_key TEXT NOT NULL, + kind TEXT NOT NULL, + answer TEXT NOT NULL, + confidence DOUBLE PRECISION, + provider TEXT NOT NULL, + model TEXT NOT NULL, + state_hash TEXT NOT NULL, + question_hash TEXT NOT NULL, + item_seq BIGINT NOT NULL, + state_truncated INTEGER NOT NULL DEFAULT 0, + evaluated_at TEXT NOT NULL, + + FOREIGN KEY (item_id, workspace_id) REFERENCES items(id, workspace_id) ON DELETE CASCADE +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_item_decisions_state + ON item_decisions(item_id, question_set, question_key, state_hash, question_hash); + +CREATE INDEX IF NOT EXISTS idx_item_decisions_latest + ON item_decisions(item_id, question_set, question_key, evaluated_at); + +CREATE TABLE IF NOT EXISTS decision_jobs ( + item_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + question_set TEXT NOT NULL, + generation BIGINT NOT NULL DEFAULT 1, + enqueued_at TEXT NOT NULL, + claimed_by TEXT, + lease_expires_at TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + + PRIMARY KEY (item_id, question_set), + FOREIGN KEY (item_id, workspace_id) REFERENCES items(id, workspace_id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_decision_jobs_enqueued + ON decision_jobs(enqueued_at); diff --git a/internal/store/store.go b/internal/store/store.go index 3337e9c53..66749c66b 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -62,6 +62,11 @@ type Store struct { // every test that sets it does so synchronously and none is parallel). afterDebounceRead func() + // decisionSets names the typed-decision question sets that apply to an + // item (TASK-3117). Empty until the server installs a resolver, which it + // does only when a decision provider is configured; see decisions.go. + decisionSets decisionSetsPtr + // outboxRowCapOverride and outboxClaimBudgetOverride are TEST-ONLY seams, // zero in production. They lower MaxOutboxPayloadBytes and // maxOutboxClaimBytes so a test can cross those bounds with kilobytes diff --git a/internal/store/wiki_links.go b/internal/store/wiki_links.go index 0bd604179..bffb85d5f 100644 --- a/internal/store/wiki_links.go +++ b/internal/store/wiki_links.go @@ -789,6 +789,11 @@ func (s *Store) cascadeTitleRename(tx *sql.Tx, renamedItemID, workspaceID, oldTi } continue } + // NOT a decision enqueue door (TASK-3117 ruling 3). This is a system + // fan-out rewrite: one title rename touches every matching item, so an enqueue + // per row would be an unbounded provider bill for one action. The rewrite + // DOES change hashed decision state, so each touched item's stored + // answers read current=false until its next direct write re-evaluates. if _, err := tx.Exec(s.q(` UPDATE items SET content = ?,