Class: Insika::TaskStore

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

Overview

Domain store for tasks. Persists Tasks over an injected Insika::Store, with the STATE MACHINE validated here: the store is the only place status is written, so the invariants live where the writes live. An invalid transition is a bug and raises ArgumentError loud and early — this is how logical races are detected without a lock.

Each Execution is ONE attempt; retry/resume opens a new entry, never overwrites.

Defined Under Namespace

Classes: Execution, Task

Constant Summary collapse

SCOPE = 
"tasks"
KEY_PREFIX = 
"task:"
STATUSES = 
%i[queued running waiting paused completed failed cancelled].freeze
TRANSITIONS =

Valid transitions — anything outside this is a bug -> ArgumentError.

{
  # queued -> failed: a turn queued in the SessionActor may fail
  # at STARTUP (spawn error before the fiber) without ever running.
  queued: %i[running cancelled failed],
  running: %i[waiting paused completed failed cancelled],
  waiting: %i[running cancelled failed],
  paused: %i[running cancelled],
  completed: [], failed: [], cancelled: [] # terminal
}.freeze

Constants included from Coercion

Coercion::TRUTHY

Instance Method Summary collapse

Methods included from Coercion

blank?, deep_stringify, presence, present?, truthy?, utf8

Constructor Details

#initialize(store:) ⇒ TaskStore

Returns a new instance of TaskStore.



39
40
41
 
# File 'lib/insika/task_store.rb', line 39

def initialize(store:)
  @store = store
end

Instance Method Details

#append_message(id, text, separator: "\n") ⇒ Object

(collect): appends a fragment to a task's message while it is still waiting at the door. -> Task.

ONLY on :queued, and that guard is the whole safety of the feature: once a turn is :running its input has been read into the Chat, seeded into the context and possibly sent to the provider — rewriting it there would mean the transcript disagrees with what the model actually saw. ArgumentError on any other status, so a lost race fails loudly instead of corrupting a turn.



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/insika/task_store.rb', line 142

def append_message(id, text, separator: "\n")
  fragment = presence(text)
  return to_task(fetch!(id)) if fragment.nil?

  record = fetch!(id)
  unless record["status"] == "queued"
    raise ArgumentError, "task #{id} is #{record['status']}, not queued: its message is already in flight"
  end

  payload = (record["command"]["payload"] ||= {})
  current = presence(payload["message"])
  payload["message"] = current ? "#{current}#{separator}#{fragment}" : fragment
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_task(record)
end

#begin_execution(id) ⇒ Object

-> Task; opens an Execution (attempt N+1). ArgumentError if one is already open (a double attempt is a bug — one owner per task). Append-only.

Raises:

  • (ArgumentError) 


104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/insika/task_store.rb', line 104

def begin_execution(id)
  record = fetch!(id)
  raise ArgumentError, "an open Execution already exists on task #{id}" if open_execution(record)

  record["executions"] += [{
    "attempt" => record["executions"].size + 1,
    "started_at" => timestamp,
    "finished_at" => nil,
    "outcome" => nil,
    "error" => nil
  }]
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_task(record)
end

#create(command:, session_id: nil, id: SecureRandom.uuid, at: nil) ⇒ Object

-> Task (status :queued). command: Hash (payload:, meta:) or any object that responds to to_h (e.g. Insika::Command). ArgumentError if the id already exists. at (ISO8601) is injectable for deterministic tests — the timestamp has SECOND precision, so two tasks created in the same second are indistinguishable by time to any reader ordering by it (same rationale as MemoryStore#add_note).

Raises:

  • (ArgumentError) 


49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/insika/task_store.rb', line 49

def create(command:, session_id: nil, id: SecureRandom.uuid, at: nil)
  key = key_for(id)
  raise ArgumentError, "task already exists: #{id}" unless @store.get(SCOPE, key).nil?

  now = at || timestamp
  record = {
    "id" => id.to_s,
    "status" => "queued",
    "command" => deep_stringify(command.respond_to?(:to_h) ? command.to_h : command),
    "session_id" => session_id&.to_s,
    "executions" => [],
    "mailbox_state" => { "pending" => [] },
    "created_at" => now,
    "updated_at" => now
  }
  @store.set(SCOPE, key, record)
  to_task(record)
end

#each_id ⇒ Object

-> enumerates ids without the "task:" prefix; without a block returns an Enumerator.



180
181
182
183
184
185
186
 
# File 'lib/insika/task_store.rb', line 180

def each_id
  return enum_for(:each_id) unless block_given?

  @store.list(SCOPE, KEY_PREFIX).each do |key|
    yield key.delete_prefix(KEY_PREFIX)
  end
end

#find(id) ⇒ Object

-> Task | nil



69
70
71
72
 
# File 'lib/insika/task_store.rb', line 69

def find(id)
  record = @store.get(SCOPE, key_for(id))
  record && to_task(record)
end

#finish_execution(id, outcome:) ⇒ Object

-> Task; closes the current Execution. ArgumentError if none is open. Does NOT touch status (that is transition's job).

Raises:

  • (ArgumentError) 


122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/insika/task_store.rb', line 122

def finish_execution(id, outcome:)
  record = fetch!(id)
  open = open_execution(record)
  raise ArgumentError, "no open Execution on task #{id}" if open.nil?

  open["finished_at"] = timestamp
  open["outcome"] = outcome.to_s
  record["updated_at"] = timestamp
  @store.set(SCOPE, key_for(id), record)
  to_task(record)
end

#queued ⇒ Object

Queued but never started (turn in the SessionActor queue at the crash) — no checkpoint; recovering = RUN from scratch (Recovery/ResumeTask).



177
 
# File 'lib/insika/task_store.rb', line 177

def queued = with_status(:queued)

#running_or_interrupted ⇒ Object

Interrupted (crashed mid-turn): have a checkpoint -> resume.



173
 
# File 'lib/insika/task_store.rb', line 173

def running_or_interrupted = with_status(:running, :waiting, :paused)

#transition(id, to:, error: nil) ⇒ Object

-> Task; validates the state machine. NotFoundError if absent, ArgumentError for a status outside the enum or an invalid transition. If error: is provided AND there is an open Execution, it closes it in the same write (Recovery path).

The read-check-write rides Store#transaction because this is also the dispatch CLAIM: two workers racing queued -> running serialize on the backend's lock, the loser re-reads :running and gets the loud ArgumentError instead of a second silent owner.

Raises:

  • (ArgumentError) 


83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/insika/task_store.rb', line 83

def transition(id, to:, error: nil)
  target = to.to_sym
  raise ArgumentError, "invalid status: #{to}" unless STATUSES.include?(target)

  @store.transaction do
    record = fetch!(id)
    from = record["status"].to_sym
    unless TRANSITIONS.fetch(from).include?(target)
      raise ArgumentError, "invalid transition: #{from} -> #{target}"
    end

    close_open_execution(record, outcome: target.to_s, error: error) if error
    record["status"] = target.to_s
    record["updated_at"] = timestamp
    @store.set(SCOPE, key_for(id), record)
    to_task(record)
  end
end

#with_status(*statuses) ⇒ Object

-> [Task] with one of the given statuses. O(n) scan at boot; acceptable (one node, local SQLite).



161
162
163
164
165
166
167
168
169
170
 
# File 'lib/insika/task_store.rb', line 161

def with_status(*statuses)
  wanted = statuses.flatten
  @store.list(SCOPE, KEY_PREFIX).filter_map do |key|
    record = @store.get(SCOPE, key)
    next if record.nil?

    task = to_task(record)
    task if wanted.include?(task.status)
  end
end