| 1 | package readcoord |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "sort" |
| 6 | "sync" |
| 7 | "time" |
| 8 | |
| 9 | "reasonix/internal/tool" |
| 10 | ) |
| 11 | |
| 12 | // Advice is a non-blocking steering signal for a stalled read. |
| 13 | type Advice string |
| 14 | |
| 15 | const ( |
| 16 | // AdvicePivot tells the caller to change approach once before pausing. |
| 17 | AdvicePivot Advice = "pivot" |
| 18 | ) |
| 19 | |
| 20 | // Policy bounds automatic continuation of one logical read. The zero value |
| 21 | // imposes no bound. |
| 22 | type Policy struct { |
| 23 | MaxPages int |
| 24 | MaxActiveTime time.Duration |
| 25 | // PivotAfter is the consecutive no-progress count that triggers one |
| 26 | // strategy change; PauseAfter more stops the read. |
| 27 | PivotAfter int |
| 28 | PauseAfter int |
| 29 | } |
| 30 | |
| 31 | // DefaultPolicy is the internal continuation bound: 64 pages or 120 seconds of |
| 32 | // active reading per logical read, with a single strategy change after two |
| 33 | // stalled pages and a pause two stalled pages later. |
| 34 | func DefaultPolicy() Policy { |
| 35 | return Policy{MaxPages: 64, MaxActiveTime: 120 * time.Second, PivotAfter: 2, PauseAfter: 2} |
| 36 | } |
| 37 | |
| 38 | // Transition reports what one observation changed. Callers commit progress |
| 39 | // from it; nothing else mutates an obligation. |
| 40 | type Transition struct { |
| 41 | Key string |
| 42 | Scope Scope |
| 43 | From, To State |
| 44 | Generation uint64 |
| 45 | Sequence uint64 |
| 46 | // Added is the coverage this delivery contributed that was not already |
| 47 | // known for the current version. |
| 48 | Added []tool.ReadRange |
| 49 | // Missing is what the requirement still lacks after the delivery. |
| 50 | Missing []tool.ReadRange |
| 51 | // Covered is the accumulated coverage on the current content version. |
| 52 | Covered []tool.ReadRange |
| 53 | SourceEnd *int |
| 54 | Stale bool |
| 55 | Progress bool |
| 56 | Advice Advice |
| 57 | Stop *Block |
| 58 | } |
| 59 | |
| 60 | // Coordinator owns every obligation. It is safe for concurrent use, but the |
| 61 | // agent feeds it from the single mutation-ordered finalizer so decisions |
| 62 | // follow provider order. |
| 63 | type Coordinator struct { |
| 64 | mu sync.Mutex |
| 65 | byKey map[string]*Obligation |
| 66 | sequence uint64 |
| 67 | policy Policy |
| 68 | } |
| 69 | |
| 70 | // New returns an empty coordinator with the default continuation bound. |
| 71 | func New() *Coordinator { return NewWithPolicy(DefaultPolicy()) } |
| 72 | |
| 73 | // NewWithPolicy returns a coordinator bounded by policy. |
| 74 | func NewWithPolicy(policy Policy) *Coordinator { |
| 75 | return &Coordinator{byKey: map[string]*Obligation{}, policy: policy} |
| 76 | } |
| 77 | |
| 78 | // Begin registers a requirement before its first call runs. Re-registering a |
| 79 | // key refreshes the requirement and keeps accumulated coverage. |
| 80 | func (c *Coordinator) Begin(key string, scope Scope, req Requirement) Obligation { |
| 81 | c.mu.Lock() |
| 82 | defer c.mu.Unlock() |
| 83 | |
| 84 | ob := c.byKey[key] |
| 85 | if ob == nil { |
| 86 | ob = &Obligation{Key: key, Scope: scope} |
| 87 | c.byKey[key] = ob |
| 88 | } |
| 89 | ob.Scope = scope |
| 90 | ob.Requirement = Requirement{Intent: req.Intent, Ranges: append([]tool.ReadRange(nil), req.Ranges...), WholeFile: req.WholeFile} |
| 91 | // A new requirement revives a finished obligation: coverage stays valid |
| 92 | // because it is scoped to one content version. |
| 93 | if ob.State == StateCreated || ob.State.Terminal() { |
| 94 | ob.State = StateFetching |
| 95 | } |
| 96 | return ob.clone() |
| 97 | } |
| 98 | |
| 99 | // Observe folds one delivered envelope into its obligation. ok=false means the |
| 100 | // envelope carried no identity or the obligation was already terminal, so a |
| 101 | // cancelled or satisfied read is never resurrected by a late delivery. |
| 102 | func (c *Coordinator) Observe(env tool.ReadResultEnvelope, activeMillis int64) (Transition, bool) { |
| 103 | if env.ReadID == "" || env.Source.CanonicalPath == "" { |
| 104 | return Transition{}, false |
| 105 | } |
| 106 | c.mu.Lock() |
| 107 | defer c.mu.Unlock() |
| 108 | |
| 109 | ob := c.byKey[env.ReadID] |
| 110 | if ob == nil { |
| 111 | ob = &Obligation{ |
| 112 | Key: env.ReadID, |
| 113 | Scope: Scope{WorkspaceID: env.Source.WorkspaceID, CanonicalPath: env.Source.CanonicalPath}, |
| 114 | Requirement: requirementFor(env), |
| 115 | } |
| 116 | c.byKey[ob.Key] = ob |
| 117 | } |
| 118 | if ob.State.Terminal() { |
| 119 | // A verified repeat is accounting, not a new incomplete requirement. |
| 120 | if ob.State == StateSatisfied && env.Source.Identity != "" && ob.Source == env.Source { |
| 121 | ob.Pages++ |
| 122 | ob.ActiveTime += time.Duration(activeMillis) * time.Millisecond |
| 123 | } |
| 124 | return Transition{}, false |
| 125 | } |
| 126 | if env.Source.Identity != "" { |
| 127 | ob.Source = env.Source |
| 128 | } |
| 129 | |
| 130 | c.sequence++ |
| 131 | ob.Sequence = c.sequence |
| 132 | tr := Transition{Key: ob.Key, Scope: ob.Scope, From: ob.State, Sequence: c.sequence} |
| 133 | |
| 134 | // Fragments of two content versions must never be stitched into one |
| 135 | // coverage claim, so a version change discards what was accumulated. |
| 136 | if (ob.Version != "" && env.Source.Snapshot != ob.Version) || (env.Source.Snapshot == "" && ob.Pages > 0) { |
| 137 | ob.Covered = nil |
| 138 | ob.SawEOF = false |
| 139 | ob.SourceEnd = nil |
| 140 | ob.Generation++ |
| 141 | tr.Stale = true |
| 142 | } |
| 143 | ob.Version = env.Source.Snapshot |
| 144 | ob.SawEOF = ob.SawEOF || env.EOF |
| 145 | if env.SourceEnd != nil { |
| 146 | end := *env.SourceEnd |
| 147 | ob.SourceEnd = &end |
| 148 | } |
| 149 | // A delivery supersedes an earlier stop reason: whatever blocked the read |
| 150 | // no longer explains its state. |
| 151 | ob.Stop = nil |
| 152 | before := ob.Covered |
| 153 | ob.Covered = Normalize(append(append([]tool.ReadRange(nil), ob.Covered...), env.DeliveredRanges...)) |
| 154 | tr.Added = Subtract(ob.Covered, before) |
| 155 | tr.Progress = len(tr.Added) > 0 |
| 156 | ob.Pages++ |
| 157 | if tr.Progress { |
| 158 | ob.Stagnant = 0 |
| 159 | } else { |
| 160 | ob.Stagnant++ |
| 161 | } |
| 162 | |
| 163 | ob.ActiveTime += time.Duration(activeMillis) * time.Millisecond |
| 164 | ob.State = evaluate(ob, env) |
| 165 | tr.Advice = c.enforcePolicy(ob, tr) |
| 166 | tr.To = ob.State |
| 167 | tr.Generation = ob.Generation |
| 168 | tr.Missing = missingFor(ob) |
| 169 | tr.Covered = append([]tool.ReadRange(nil), ob.Covered...) |
| 170 | if ob.SourceEnd != nil { |
| 171 | end := *ob.SourceEnd |
| 172 | tr.SourceEnd = &end |
| 173 | } |
| 174 | tr.Stop = ob.Stop |
| 175 | return tr, true |
| 176 | } |
| 177 | |
| 178 | // enforcePolicy applies the hard budget and the no-progress ladder. A content |
| 179 | // change never resets the budget: only satisfied or cancelled ends it. |
| 180 | func (c *Coordinator) enforcePolicy(ob *Obligation, tr Transition) Advice { |
| 181 | if ob.State.Terminal() || ob.Stop != nil { |
| 182 | return "" |
| 183 | } |
| 184 | switch { |
| 185 | case c.policy.MaxPages > 0 && ob.Pages >= c.policy.MaxPages: |
| 186 | ob.State = StateBlocked |
| 187 | ob.Stop = &Block{ |
| 188 | Code: "page_budget", |
| 189 | Detail: fmt.Sprintf("automatic continuation stopped after %d pages", ob.Pages), |
| 190 | Recovery: "read the remaining lines explicitly, or work on an independent item", |
| 191 | } |
| 192 | return "" |
| 193 | case c.policy.MaxActiveTime > 0 && ob.ActiveTime >= c.policy.MaxActiveTime: |
| 194 | ob.State = StateBlocked |
| 195 | ob.Stop = &Block{ |
| 196 | Code: "time_budget", |
| 197 | Detail: fmt.Sprintf("automatic continuation used %s of active read time", ob.ActiveTime.Round(time.Second)), |
| 198 | Recovery: "read the remaining lines explicitly, or work on an independent item", |
| 199 | } |
| 200 | return "" |
| 201 | case !tr.Progress && c.policy.PivotAfter > 0 && ob.Stagnant >= c.policy.PivotAfter && !ob.Pivoted: |
| 202 | ob.Pivoted = true |
| 203 | return AdvicePivot |
| 204 | case ob.Pivoted && c.policy.PauseAfter > 0 && ob.Stagnant >= c.policy.PivotAfter+c.policy.PauseAfter: |
| 205 | ob.State = StateBlocked |
| 206 | ob.Stop = &Block{ |
| 207 | Code: "no_progress", |
| 208 | Detail: fmt.Sprintf("%d consecutive pages added no new content", ob.Stagnant), |
| 209 | Recovery: "change approach or read a narrower window; the read stays paused until new content arrives", |
| 210 | } |
| 211 | } |
| 212 | return "" |
| 213 | } |
| 214 | |
| 215 | // Fail records a read that could not deliver at all. |
| 216 | func (c *Coordinator) Fail(key string, block Block) (Transition, bool) { |
| 217 | return c.stop(key, StateBlocked, block) |
| 218 | } |
| 219 | |
| 220 | // Narrow records that the requirement cannot be met within the current budget. |
| 221 | // Only a local requirement may narrow; a whole-file requirement reports |
| 222 | // needs_scope instead of silently downgrading. |
| 223 | func (c *Coordinator) Narrow(key string, block Block) (Transition, bool) { |
| 224 | return c.stop(key, StateNeedsScope, block) |
| 225 | } |
| 226 | |
| 227 | func (c *Coordinator) stop(key string, state State, block Block) (Transition, bool) { |
| 228 | c.mu.Lock() |
| 229 | defer c.mu.Unlock() |
| 230 | |
| 231 | ob := c.byKey[key] |
| 232 | if ob == nil || ob.State.Terminal() { |
| 233 | return Transition{}, false |
| 234 | } |
| 235 | c.sequence++ |
| 236 | ob.Sequence = c.sequence |
| 237 | tr := Transition{Key: key, Scope: ob.Scope, From: ob.State, To: state, Sequence: c.sequence, Generation: ob.Generation} |
| 238 | ob.State = state |
| 239 | ob.Stop = &block |
| 240 | tr.Stop = ob.Stop |
| 241 | tr.Missing = missingFor(ob) |
| 242 | tr.Covered = append([]tool.ReadRange(nil), ob.Covered...) |
| 243 | if ob.SourceEnd != nil { |
| 244 | end := *ob.SourceEnd |
| 245 | tr.SourceEnd = &end |
| 246 | } |
| 247 | return tr, true |
| 248 | } |
| 249 | |
| 250 | // Cancel marks an obligation cancelled. Its state is terminal, so a later |
| 251 | // delivery for the same key is ignored. |
| 252 | func (c *Coordinator) Cancel(key string) (Transition, bool) { |
| 253 | c.mu.Lock() |
| 254 | defer c.mu.Unlock() |
| 255 | |
| 256 | ob := c.byKey[key] |
| 257 | if ob == nil || ob.State.Terminal() { |
| 258 | return Transition{}, false |
| 259 | } |
| 260 | c.sequence++ |
| 261 | ob.Sequence = c.sequence |
| 262 | tr := Transition{Key: key, Scope: ob.Scope, From: ob.State, To: StateCancelled, Sequence: c.sequence, Generation: ob.Generation} |
| 263 | ob.State = StateCancelled |
| 264 | return tr, true |
| 265 | } |
| 266 | |
| 267 | // Get returns a copy of one obligation. |
| 268 | func (c *Coordinator) Get(key string) (Obligation, bool) { |
| 269 | c.mu.Lock() |
| 270 | defer c.mu.Unlock() |
| 271 | |
| 272 | ob, ok := c.byKey[key] |
| 273 | if !ok { |
| 274 | return Obligation{}, false |
| 275 | } |
| 276 | return ob.clone(), true |
| 277 | } |
| 278 | |
| 279 | // Snapshot returns every obligation ordered by key. |
| 280 | func (c *Coordinator) Snapshot() []Obligation { |
| 281 | c.mu.Lock() |
| 282 | defer c.mu.Unlock() |
| 283 | |
| 284 | out := make([]Obligation, 0, len(c.byKey)) |
| 285 | for _, ob := range c.byKey { |
| 286 | out = append(out, ob.clone()) |
| 287 | } |
| 288 | sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key }) |
| 289 | return out |
| 290 | } |
| 291 | |
| 292 | func requirementFor(env tool.ReadResultEnvelope) Requirement { |
| 293 | switch env.Intent { |
| 294 | case tool.ReadIntentFull: |
| 295 | return Requirement{Intent: tool.ReadIntentFull, WholeFile: true} |
| 296 | case tool.ReadIntentRange: |
| 297 | var ranges []tool.ReadRange |
| 298 | if env.RequestedRange != nil { |
| 299 | ranges = []tool.ReadRange{*env.RequestedRange} |
| 300 | } else { |
| 301 | ranges = append(ranges, env.DeliveredRanges...) |
| 302 | } |
| 303 | return Requirement{Intent: tool.ReadIntentRange, Ranges: Normalize(ranges)} |
| 304 | default: |
| 305 | return Requirement{Intent: tool.ReadIntentInspect} |
| 306 | } |
| 307 | } |
| 308 | |
| 309 | func evaluate(ob *Obligation, env tool.ReadResultEnvelope) State { |
| 310 | switch ob.Requirement.Intent { |
| 311 | case tool.ReadIntentInspect: |
| 312 | // One bounded page completes an inspect requirement; content left in |
| 313 | // the file is not an outstanding debt. |
| 314 | return StateSatisfied |
| 315 | case tool.ReadIntentRange: |
| 316 | if len(ob.Requirement.Ranges) == 0 || Covers(ob.Covered, ob.Requirement.Ranges) { |
| 317 | return StateSatisfied |
| 318 | } |
| 319 | // Reaching EOF satisfies a range only when the reader vouched for where |
| 320 | // the source ends and that end is inside the requested window. |
| 321 | if ob.SawEOF && ob.SourceEnd != nil { |
| 322 | var required []tool.ReadRange |
| 323 | for _, r := range ob.Requirement.Ranges { |
| 324 | if end := min(r.End, *ob.SourceEnd); r.Start < end { |
| 325 | required = append(required, tool.ReadRange{Start: r.Start, End: end}) |
| 326 | } |
| 327 | } |
| 328 | if Covers(ob.Covered, required) { |
| 329 | return StateSatisfied |
| 330 | } |
| 331 | } |
| 332 | return StateNeedsMore |
| 333 | case tool.ReadIntentFull: |
| 334 | return evaluateWholeFile(ob, env) |
| 335 | default: |
| 336 | return StateDelivered |
| 337 | } |
| 338 | } |
| 339 | |
| 340 | func evaluateWholeFile(ob *Obligation, _ tool.ReadResultEnvelope) State { |
| 341 | // A whole-file read is only proven by a trustworthy source end plus |
| 342 | // contiguous coverage from line 0 on one version. |
| 343 | if !ob.SawEOF || ob.SourceEnd == nil { |
| 344 | return StateNeedsMore |
| 345 | } |
| 346 | if *ob.SourceEnd == 0 { |
| 347 | return StateSatisfied |
| 348 | } |
| 349 | if len(ob.Covered) == 1 && ob.Covered[0].Start == 0 && ob.Covered[0].End >= *ob.SourceEnd { |
| 350 | return StateSatisfied |
| 351 | } |
| 352 | return StateNeedsMore |
| 353 | } |
| 354 | |
| 355 | func missingFor(ob *Obligation) []tool.ReadRange { |
| 356 | switch ob.Requirement.Intent { |
| 357 | case tool.ReadIntentRange: |
| 358 | return Subtract(ob.Requirement.Ranges, ob.Covered) |
| 359 | case tool.ReadIntentFull: |
| 360 | if ob.SourceEnd != nil { |
| 361 | return Subtract([]tool.ReadRange{{Start: 0, End: *ob.SourceEnd}}, ob.Covered) |
| 362 | } |
| 363 | if len(ob.Covered) == 0 { |
| 364 | return nil |
| 365 | } |
| 366 | return Subtract([]tool.ReadRange{{Start: 0, End: ob.Covered[len(ob.Covered)-1].End}}, ob.Covered) |
| 367 | default: |
| 368 | return nil |
| 369 | } |
| 370 | } |
| 371 |