Class: Aidp::Database::Repositories::WorkstreamRepository

Inherits:
Aidp::Database::Repository show all
Defined in:
lib/aidp/database/repositories/workstream_repository.rb

Overview

Repository for workstreams and workstream_events tables Replaces workstreams/*/state.json and history.jsonl

Instance Attribute Summary

Attributes inherited from Aidp::Database::Repository

#project_dir, #table_name

Instance Method Summary collapse

Constructor Details

#initialize(project_dir: Dir.pwd) ⇒ WorkstreamRepository

Returns a new instance of WorkstreamRepository.



11
12
13
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 11

def initialize(project_dir: Dir.pwd)
  super(project_dir: project_dir, table_name: "workstreams")
end

Instance Method Details

#append_event(slug:, type:, data: {}) ⇒ Object

Append event to workstream history

Parameters:

  • slug (String) —

    Workstream slug

  • type (String) —

    Event type

  • data (Hash) (defaults to: {}) —

    Event data



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 124

def append_event(slug:, type:, data: {})
  now = current_timestamp

  execute(
    <<~SQL,
      INSERT INTO workstream_events
        (project_dir, workstream_slug, event_type, event_data, timestamp)
      VALUES (?, ?, ?, ?, ?)
    SQL
    [project_dir, slug, type, serialize_json(data), now]
  )

  Aidp.log_debug("workstream_repository", "event_appended",
    slug: slug, type: type)
end

#complete(slug:) ⇒ Hash

Complete workstream

Parameters:

  • slug (String) —

    Workstream slug

Returns:

  • (Hash) —

    Result with status or error



195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 195

def complete(slug:)
  state = read(slug: slug)
  return {error: "Workstream not found"} unless state
  return {error: "Already completed"} if state[:status] == "completed"

  now = current_timestamp
  update(slug: slug, status: "completed", completed_at: now)
  append_event(slug: slug, type: "completed", data: {iterations: state[:iteration]})

  {status: "completed"}
end

#elapsed_seconds(slug:) ⇒ Integer

Get elapsed time for workstream

Parameters:

  • slug (String) —

    Workstream slug

Returns:

  • (Integer) —

    Elapsed seconds



242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 242

def elapsed_seconds(slug:)
  state = read(slug: slug)
  return 0 unless state

  metadata = state[:metadata] || {}
  return 0 unless metadata[:started_at]

  started = Time.parse(metadata[:started_at])
  (Time.now.utc - started).to_i
rescue ArgumentError
  0
end

#increment_iteration(slug:) ⇒ Hash

Increment iteration counter

Parameters:

  • slug (String) —

    Workstream slug

Returns:

  • (Hash) —

    Updated workstream state



105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 105

def increment_iteration(slug:)
  existing = read(slug: slug)
  existing ||= init(slug: slug)

  new_iteration = (existing[:iteration] || 0) + 1
  new_status = (existing[:status] == "paused") ? "active" : existing[:status]

  state = update(slug: slug, iteration: new_iteration, status: new_status)

  append_event(slug: slug, type: "iteration", data: {count: new_iteration})

  state
end

#init(slug:, task: nil) ⇒ Hash

Initialize a new workstream

Parameters:

  • slug (String) —

    Workstream slug

  • task (String, nil) (defaults to: nil) —

    Task description

Returns:

  • (Hash) —

    Created workstream state



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 20

def init(slug:, task: nil)
  now = current_timestamp

  execute(
    insert_sql([
      :project_dir, :slug, :status, :iteration, :metadata,
      :created_at, :updated_at
    ]),
    [
      project_dir,
      slug,
      "active",
      0,
      serialize_json({task: task, started_at: now}),
      now,
      now
    ]
  )

  append_event(slug: slug, type: "created", data: {task: task})

  Aidp.log_debug("workstream_repository", "initialized", slug: slug)

  read(slug: slug)
end

#list(status: nil) ⇒ Array<Hash>

List all workstreams for project

Parameters:

  • status (String, nil) (defaults to: nil) —

    Filter by status

Returns:

  • (Array<Hash>) —

    Workstreams



259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 259

def list(status: nil)
  rows = if status
    query(
      "SELECT * FROM workstreams WHERE project_dir = ? AND status = ? ORDER BY updated_at DESC",
      [project_dir, status]
    )
  else
    query(
      "SELECT * FROM workstreams WHERE project_dir = ? ORDER BY updated_at DESC",
      [project_dir]
    )
  end

  rows.map { |row| deserialize_workstream(row) }
end

#mark_removed(slug:) ⇒ Object

Mark workstream as removed

Parameters:

  • slug (String) —

    Workstream slug



210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 210

def mark_removed(slug:)
  state = read(slug: slug)

  # Auto-complete if active
  complete(slug: slug) if state && state[:status] == "active"

  update(slug: slug, status: "removed")
  append_event(slug: slug, type: "removed", data: {})

  Aidp.log_debug("workstream_repository", "removed", slug: slug)
end

#pause(slug:) ⇒ Hash

Pause workstream

Parameters:

  • slug (String) —

    Workstream slug

Returns:

  • (Hash) —

    Result with status or error



163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 163

def pause(slug:)
  state = read(slug: slug)
  return {error: "Workstream not found"} unless state
  return {error: "Already paused"} if state[:status] == "paused"

  now = current_timestamp
  update(slug: slug, status: "paused", paused_at: now)
  append_event(slug: slug, type: "paused", data: {})

  {status: "paused"}
end

#read(slug:) ⇒ Hash?

Read workstream state

Parameters:

  • slug (String) —

    Workstream slug

Returns:

  • (Hash, nil) —

    Workstream state or nil



50
51
52
53
54
55
56
57
58
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 50

def read(slug:)
  row = query_one(
    "SELECT * FROM workstreams WHERE project_dir = ? AND slug = ?",
    [project_dir, slug]
  )
  return nil unless row

  deserialize_workstream(row)
end

#recent_events(slug:, limit: 5) ⇒ Array<Hash>

Get recent events for workstream

Parameters:

  • slug (String) —

    Workstream slug

  • limit (Integer) (defaults to: 5) —

    Maximum events to return

Returns:

  • (Array<Hash>) —

    Recent events



145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 145

def recent_events(slug:, limit: 5)
  rows = query(
    <<~SQL,
      SELECT * FROM workstream_events
      WHERE project_dir = ? AND workstream_slug = ?
      ORDER BY timestamp DESC
      LIMIT ?
    SQL
    [project_dir, slug, limit]
  )

  rows.map { |row| deserialize_event(row) }.reverse
end

#resume(slug:) ⇒ Hash

Resume workstream

Parameters:

  • slug (String) —

    Workstream slug

Returns:

  • (Hash) —

    Result with status or error



179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 179

def resume(slug:)
  state = read(slug: slug)
  return {error: "Workstream not found"} unless state
  return {error: "Not paused"} unless state[:status] == "paused"

  now = current_timestamp
  update(slug: slug, status: "active", resumed_at: now)
  append_event(slug: slug, type: "resumed", data: {})

  {status: "active"}
end

#stalled?(slug:, threshold_seconds: 3600) ⇒ Boolean

Check if workstream is stalled

Parameters:

  • slug (String) —

    Workstream slug

  • threshold_seconds (Integer) (defaults to: 3600) —

    Stall threshold in seconds

Returns:

  • (Boolean) 


227
228
229
230
231
232
233
234
235
236
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 227

def stalled?(slug:, threshold_seconds: 3600)
  state = read(slug: slug)
  return false unless state && state[:updated_at]
  return false if state[:status] != "active"

  updated = Time.parse(state[:updated_at])
  (Time.now.utc - updated).to_i > threshold_seconds
rescue ArgumentError
  false
end

#update(slug:, **attrs) ⇒ Hash

Update workstream attributes

Parameters:

  • slug (String) —

    Workstream slug

  • attrs (Hash) —

    Attributes to update

Returns:

  • (Hash) —

    Updated workstream state



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/aidp/database/repositories/workstream_repository.rb', line 65

def update(slug:, **attrs)
  existing = read(slug: slug)
  existing ||= init(slug: slug)

  now = current_timestamp
  metadata = existing[:metadata] || {}
  metadata = metadata.merge(attrs.except(:status, :iteration, :branch, :worktree_path))

  execute(
    <<~SQL,
      UPDATE workstreams SET
        status = ?,
        iteration = ?,
        branch = ?,
        worktree_path = ?,
        metadata = ?,
        updated_at = ?
      WHERE project_dir = ? AND slug = ?
    SQL
    [
      attrs[:status] || existing[:status],
      attrs[:iteration] || existing[:iteration],
      attrs[:branch] || existing[:branch],
      attrs[:worktree_path] || existing[:worktree_path],
      serialize_json(metadata),
      now,
      project_dir,
      slug
    ]
  )

  Aidp.log_debug("workstream_repository", "updated", slug: slug)

  read(slug: slug)
end