You've already forked agentic-coding-workflow
357 lines
8.6 KiB
Go
357 lines
8.6 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: steps.sql
|
|
|
|
package orchestratorsqlc
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const countActiveChildSteps = `-- name: CountActiveChildSteps :one
|
|
SELECT COUNT(*) FROM orchestrator_steps
|
|
WHERE parent_step_id = $1
|
|
AND status IN ('pending','spawning','running','awaiting_gate')
|
|
`
|
|
|
|
func (q *Queries) CountActiveChildSteps(ctx context.Context, parentStepID pgtype.UUID) (int64, error) {
|
|
row := q.db.QueryRow(ctx, countActiveChildSteps, parentStepID)
|
|
var count int64
|
|
err := row.Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
const getStep = `-- name: GetStep :one
|
|
SELECT id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id, created_at, updated_at, ended_at
|
|
FROM orchestrator_steps
|
|
WHERE id = $1
|
|
`
|
|
|
|
func (q *Queries) GetStep(ctx context.Context, id pgtype.UUID) (OrchestratorStep, error) {
|
|
row := q.db.QueryRow(ctx, getStep, id)
|
|
var i OrchestratorStep
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getStepBySession = `-- name: GetStepBySession :one
|
|
SELECT id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id, created_at, updated_at, ended_at
|
|
FROM orchestrator_steps
|
|
WHERE acp_session_id = $1
|
|
ORDER BY created_at DESC
|
|
LIMIT 1
|
|
`
|
|
|
|
func (q *Queries) GetStepBySession(ctx context.Context, acpSessionID pgtype.UUID) (OrchestratorStep, error) {
|
|
row := q.db.QueryRow(ctx, getStepBySession, acpSessionID)
|
|
var i OrchestratorStep
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const incrementStepAttempt = `-- name: IncrementStepAttempt :one
|
|
UPDATE orchestrator_steps
|
|
SET attempt = attempt + 1,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id, created_at, updated_at, ended_at
|
|
`
|
|
|
|
func (q *Queries) IncrementStepAttempt(ctx context.Context, id pgtype.UUID) (OrchestratorStep, error) {
|
|
row := q.db.QueryRow(ctx, incrementStepAttempt, id)
|
|
var i OrchestratorStep
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const insertStep = `-- name: InsertStep :one
|
|
INSERT INTO orchestrator_steps (
|
|
id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id
|
|
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
|
|
RETURNING id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id, created_at, updated_at, ended_at
|
|
`
|
|
|
|
type InsertStepParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
RunID pgtype.UUID `json:"run_id"`
|
|
Phase string `json:"phase"`
|
|
Attempt int32 `json:"attempt"`
|
|
ParentStepID pgtype.UUID `json:"parent_step_id"`
|
|
Depth int32 `json:"depth"`
|
|
Status string `json:"status"`
|
|
AgentKindID pgtype.UUID `json:"agent_kind_id"`
|
|
AcpSessionID pgtype.UUID `json:"acp_session_id"`
|
|
Prompt string `json:"prompt"`
|
|
StopReason *string `json:"stop_reason"`
|
|
GateResult []byte `json:"gate_result"`
|
|
LastError *string `json:"last_error"`
|
|
JobID pgtype.UUID `json:"job_id"`
|
|
}
|
|
|
|
func (q *Queries) InsertStep(ctx context.Context, arg InsertStepParams) (OrchestratorStep, error) {
|
|
row := q.db.QueryRow(ctx, insertStep,
|
|
arg.ID,
|
|
arg.RunID,
|
|
arg.Phase,
|
|
arg.Attempt,
|
|
arg.ParentStepID,
|
|
arg.Depth,
|
|
arg.Status,
|
|
arg.AgentKindID,
|
|
arg.AcpSessionID,
|
|
arg.Prompt,
|
|
arg.StopReason,
|
|
arg.GateResult,
|
|
arg.LastError,
|
|
arg.JobID,
|
|
)
|
|
var i OrchestratorStep
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const listStepsByRun = `-- name: ListStepsByRun :many
|
|
SELECT id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id, created_at, updated_at, ended_at
|
|
FROM orchestrator_steps
|
|
WHERE run_id = $1
|
|
ORDER BY created_at ASC
|
|
`
|
|
|
|
func (q *Queries) ListStepsByRun(ctx context.Context, runID pgtype.UUID) ([]OrchestratorStep, error) {
|
|
rows, err := q.db.Query(ctx, listStepsByRun, runID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var items []OrchestratorStep
|
|
for rows.Next() {
|
|
var i OrchestratorStep
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const resetStuckStepsOnRestart = `-- name: ResetStuckStepsOnRestart :many
|
|
UPDATE orchestrator_steps s
|
|
SET status = 'pending',
|
|
updated_at = now()
|
|
FROM orchestrator_runs r
|
|
WHERE s.run_id = r.id
|
|
AND r.status IN ('pending','running','paused')
|
|
AND s.status IN ('spawning','running','awaiting_gate')
|
|
RETURNING s.id, s.run_id, s.phase, s.attempt, s.parent_step_id, s.depth,
|
|
s.status, s.agent_kind_id, s.acp_session_id, s.prompt, s.stop_reason,
|
|
s.gate_result, s.last_error, s.job_id, s.created_at, s.updated_at, s.ended_at
|
|
`
|
|
|
|
// 把活跃 run 下卡在 spawning/running/awaiting_gate 的 step 重置为 pending，便于
|
|
// 重新入队恢复（镜像 acp ResetStuckSessionsOnRestart）。succeeded/canceled/failed
|
|
// run 下的 step 不动。
|
|
func (q *Queries) ResetStuckStepsOnRestart(ctx context.Context) ([]OrchestratorStep, error) {
|
|
rows, err := q.db.Query(ctx, resetStuckStepsOnRestart)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var items []OrchestratorStep
|
|
for rows.Next() {
|
|
var i OrchestratorStep
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const updateStepStatus = `-- name: UpdateStepStatus :one
|
|
UPDATE orchestrator_steps
|
|
SET status = $2,
|
|
acp_session_id = $3,
|
|
stop_reason = $4,
|
|
gate_result = $5,
|
|
last_error = $6,
|
|
job_id = $7,
|
|
ended_at = CASE WHEN $2 IN ('succeeded','failed','dead') THEN now() ELSE ended_at END,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING id, run_id, phase, attempt, parent_step_id, depth,
|
|
status, agent_kind_id, acp_session_id, prompt, stop_reason,
|
|
gate_result, last_error, job_id, created_at, updated_at, ended_at
|
|
`
|
|
|
|
type UpdateStepStatusParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
Status string `json:"status"`
|
|
AcpSessionID pgtype.UUID `json:"acp_session_id"`
|
|
StopReason *string `json:"stop_reason"`
|
|
GateResult []byte `json:"gate_result"`
|
|
LastError *string `json:"last_error"`
|
|
JobID pgtype.UUID `json:"job_id"`
|
|
}
|
|
|
|
func (q *Queries) UpdateStepStatus(ctx context.Context, arg UpdateStepStatusParams) (OrchestratorStep, error) {
|
|
row := q.db.QueryRow(ctx, updateStepStatus,
|
|
arg.ID,
|
|
arg.Status,
|
|
arg.AcpSessionID,
|
|
arg.StopReason,
|
|
arg.GateResult,
|
|
arg.LastError,
|
|
arg.JobID,
|
|
)
|
|
var i OrchestratorStep
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.RunID,
|
|
&i.Phase,
|
|
&i.Attempt,
|
|
&i.ParentStepID,
|
|
&i.Depth,
|
|
&i.Status,
|
|
&i.AgentKindID,
|
|
&i.AcpSessionID,
|
|
&i.Prompt,
|
|
&i.StopReason,
|
|
&i.GateResult,
|
|
&i.LastError,
|
|
&i.JobID,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
&i.EndedAt,
|
|
)
|
|
return i, err
|
|
}
|