Class: Insika::CheckpointStore

Inherits:
Object
  • Object
show all
Includes:
Coercion
Defined in:
lib/insika/checkpoint_store.rb

Overview

Domain store for checkpoints. A per-turn snapshot written in an ALL-OR-NOTHING transaction (invariant: a checkpoint is either fully valid or does not exist), a record of non-idempotent side effects in a spill key during the turn, and prune to bound growth.

Two key families in the "checkpoints" scope:

"checkpoint:<task_id>:turn:<n>"  -> Checkpoint JSON
"sideeffects:<task_id>:turn:<n>" -> ["tool_call_id", ...] (spill key)

The spill key exists because the turn's checkpoint does not exist yet when the tool calls run (it is only saved at stage 8): it is written BEFORE the tool result goes back to the model, and the following save consolidates it into completed_side_effects and deletes it in the SAME transaction.

Constant Summary collapse

SCOPE = 
"checkpoints"

Instance Method Summary collapse

Methods included from Coercion

blank?, deep_stringify, presence, present?, utf8

Constructor Details

#initialize(store:) ⇒ CheckpointStore

Returns a new instance of CheckpointStore.



24
25
26
 
# File 'lib/insika/checkpoint_store.rb', line 24

def initialize(store:)
  @store = store
end

Instance Method Details

#find(task_id, turn:) ⇒ Object

-> Checkpoint | nil



67
68
69
70
 
# File 'lib/insika/checkpoint_store.rb', line 67

def find(task_id, turn:)
  record = @store.get(SCOPE, checkpoint_key(task_id, turn))
  record && to_checkpoint(record)
end

#latest(task_id) ⇒ Object

-> Checkpoint | nil (highest turn). NUMERIC ordering: list sorts lexicographically and "turn:9" > "turn:10" — parse n as an Integer.



59
60
61
62
63
64
 
# File 'lib/insika/checkpoint_store.rb', line 59

def latest(task_id)
  turns = checkpoint_turns(task_id)
  return nil if turns.empty?

  find(task_id, turn: turns.max)
end

#prune(task_id, keep: 1) ⇒ Object

-> void. Keeps the keep checkpoints with the highest turn (numeric); deletes the rest. Also clears spill keys of turns strictly smaller than the smallest kept turn (unreachable garbage after consolidation). In a transaction so it never leaves a partial prune. No-op if there are <= keep checkpoints.



97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/insika/checkpoint_store.rb', line 97

def prune(task_id, keep: 1)
  @store.transaction do
    turns = checkpoint_turns(task_id).sort
    next if turns.size <= keep

    kept = turns.last(keep)
    smallest_kept = kept.first
    (turns - kept).each { |n| @store.delete(SCOPE, checkpoint_key(task_id, n)) }
    sideeffect_turns(task_id).each do |n|
      @store.delete(SCOPE, sideeffects_key(task_id, n)) if n < smallest_kept
    end
  end
  nil
end

#record_side_effect(task_id, turn:, tool_call_id:) ⇒ Object

-> nil; idempotent (recording twice = one entry). In a transaction (written before the tool goes back to the model).



74
75
76
77
78
79
80
81
82
 
# File 'lib/insika/checkpoint_store.rb', line 74

def record_side_effect(task_id, turn:, tool_call_id:)
  @store.transaction do
    key = sideeffects_key(task_id, turn)
    ids = @store.get(SCOPE, key) || []
    id = tool_call_id.to_s
    @store.set(SCOPE, key, ids + [id]) unless ids.include?(id)
  end
  nil
end

#save(checkpoint) ⇒ Object

-> Checkpoint (with the consolidated side-effect list). ALWAYS in a transaction. Order: validate monotonicity -> consolidate the previous turn's spill key -> write the checkpoint -> delete the absorbed spill key. Any exception in the middle -> full rollback (neither a partial checkpoint nor a lost spill key).



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/insika/checkpoint_store.rb', line 33

def save(checkpoint)
  @store.transaction do
    current = latest(checkpoint.task_id)
    if current && current.turn >= checkpoint.turn
      raise ArgumentError,
            "checkpoint with non-monotonic turn: #{checkpoint.turn} <= #{current.turn}"
    end

    # Stage 8 of turn n saves turn n+1's checkpoint:
    # the spill key to absorb is that of the turn that just executed (n).
    spill_key = sideeffects_key(checkpoint.task_id, checkpoint.turn - 1)
    spilled = @store.get(SCOPE, spill_key) || []
    consolidated = Array(checkpoint.completed_side_effects).map(&:to_s) | spilled

    record = deep_stringify(checkpoint.to_h)
    record["completed_side_effects"] = consolidated
    record["created_at"] ||= timestamp
    @store.set(SCOPE, checkpoint_key(checkpoint.task_id, checkpoint.turn), record)
    @store.delete(SCOPE, spill_key)

    to_checkpoint(record)
  end
end

#side_effects(task_id, turn:) ⇒ Object

-> [tool_call_id] = spill key ∪ checkpoint of the same turn. Covers both places where an id may live during the cycle; since tool_call_id is globally unique, the union never causes an improper skip.



87
88
89
90
91
 
# File 'lib/insika/checkpoint_store.rb', line 87

def side_effects(task_id, turn:)
  spilled = @store.get(SCOPE, sideeffects_key(task_id, turn)) || []
  from_checkpoint = find(task_id, turn: turn)&.completed_side_effects || []
  spilled | from_checkpoint
end