agent-radar
2ed3415fb7
Publish Claude Code Binaries / Stage and publish binaries (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-arm64 (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build linux-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build windows-amd64 (push) Canceled after 0s
Sync Tutti Agent Skills / sync (push) Canceled after 0s
Windows Agent Adapters / Test Windows agent adapters (push) Canceled after 0s
Windows Daemon Adapters / Test Windows daemon adapters (push) Canceled after 0s
Publish Tutti App Runtime / Publish runtime catalog (push) Canceled after 0s
817 lines
34 KiB
Go
817 lines
34 KiB
Go
package storesqlite
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
)
|
|
|
|
const runtimeOperationSelectSQL = `
|
|
SELECT operation_id, workspace_id, agent_session_id, kind, status, COALESCE(result, ''),
|
|
turn_id, COALESCE(request_id, ''), payload_json, COALESCE(lease_owner, ''),
|
|
COALESCE(lease_expires_at_unix_ms, 0), COALESCE(next_attempt_at_unix_ms, 0),
|
|
attempt, version, last_error,
|
|
created_at_unix_ms, updated_at_unix_ms, COALESCE(completed_at_unix_ms, 0)
|
|
FROM workspace_agent_runtime_operations
|
|
`
|
|
|
|
func (s *Store) PrepareRuntimeOperation(ctx context.Context, input RuntimeOperationPrepare) (RuntimeOperation, bool, error) {
|
|
if s == nil || s.db == nil {
|
|
return RuntimeOperation{}, false, errors.New("workspace database is not initialized")
|
|
}
|
|
input.OperationID = strings.TrimSpace(input.OperationID)
|
|
input.WorkspaceID = strings.TrimSpace(input.WorkspaceID)
|
|
input.AgentSessionID = strings.TrimSpace(input.AgentSessionID)
|
|
input.TurnID = strings.TrimSpace(input.TurnID)
|
|
input.RequestID = strings.TrimSpace(input.RequestID)
|
|
if err := validateRuntimeOperationPrepare(input); err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
payloadJSON, err := marshalJSONMap(input.Payload)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
now := input.OccurredAtMS
|
|
if now <= 0 {
|
|
return RuntimeOperation{}, false, errors.New("runtime operation occurred time is required")
|
|
}
|
|
requestID := any(nil)
|
|
switch input.Kind {
|
|
case RuntimeOperationKindInteractiveResponse, RuntimeOperationKindPlanDecision, RuntimeOperationKindEditRetry:
|
|
requestID = input.RequestID
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("begin prepare runtime operation: %w", err)
|
|
}
|
|
committed := false
|
|
defer func() {
|
|
if !committed {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
existing, found, err := getRuntimeOperationByIdentityTx(ctx, tx, input)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
if found {
|
|
if existing.OperationID != input.OperationID {
|
|
return RuntimeOperation{}, false, ErrRuntimeOperationIdentityMismatch
|
|
}
|
|
if !jsonMapsEqual(existing.Payload, input.Payload) {
|
|
return RuntimeOperation{}, false, ErrRuntimeOperationConflict
|
|
}
|
|
if _, err := s.commitTransaction(ctx, tx, input.WorkspaceID, nil); err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("commit duplicate runtime operation prepare: %w", err)
|
|
}
|
|
committed = true
|
|
return existing, false, nil
|
|
}
|
|
if byID, idFound, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID); err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
} else if idFound {
|
|
if !runtimeOperationIdentityMatches(byID, input) {
|
|
return RuntimeOperation{}, false, ErrRuntimeOperationIdentityMismatch
|
|
}
|
|
return RuntimeOperation{}, false, ErrRuntimeOperationConflict
|
|
}
|
|
if err := requireSessionForkSourceWritableTx(
|
|
ctx, tx, input.WorkspaceID, input.AgentSessionID,
|
|
); err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
if err := validateRuntimeOperationSubjectTx(ctx, tx, input); err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_agent_runtime_operations (
|
|
operation_id, workspace_id, agent_session_id, kind, status, turn_id,
|
|
request_id, payload_json, next_attempt_at_unix_ms, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, input.OperationID, input.WorkspaceID, input.AgentSessionID, input.Kind,
|
|
RuntimeOperationStatusPrepared, input.TurnID, requestID, payloadJSON, now, now, now)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("insert runtime operation: %w", err)
|
|
}
|
|
op, _, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
delta, err := s.commitTransaction(ctx, tx, input.WorkspaceID, []TransactionMutation{
|
|
transactionMutation(input.WorkspaceID, input.AgentSessionID, MutationEntityRuntimeOperation, input.OperationID, "prepare", op.Version),
|
|
})
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("commit runtime operation prepare: %w", err)
|
|
}
|
|
committed = true
|
|
op.CommitTransactionID = delta.TransactionID
|
|
op.CommitDelta = delta
|
|
return op, true, nil
|
|
}
|
|
|
|
// PrepareInteractiveRuntimeOperation atomically establishes the response claim
|
|
// by transitioning the interaction from pending to answered and persists the
|
|
// durable operation that will deliver that answer to the runtime. A terminal
|
|
// interaction returns its stored disposition without creating another
|
|
// operation, allowing competing GUI/CLI responders to normalize the result.
|
|
func (s *Store) PrepareInteractiveRuntimeOperation(
|
|
ctx context.Context,
|
|
input RuntimeOperationPrepare,
|
|
) (RuntimeOperation, Interaction, InteractionTransitionResult, error) {
|
|
if s == nil || s.db == nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, errors.New("workspace database is not initialized")
|
|
}
|
|
input.OperationID = strings.TrimSpace(input.OperationID)
|
|
input.WorkspaceID = strings.TrimSpace(input.WorkspaceID)
|
|
input.AgentSessionID = strings.TrimSpace(input.AgentSessionID)
|
|
input.TurnID = strings.TrimSpace(input.TurnID)
|
|
input.RequestID = strings.TrimSpace(input.RequestID)
|
|
if input.Kind != RuntimeOperationKindInteractiveResponse {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, errors.New("interactive runtime operation kind is required")
|
|
}
|
|
if err := validateRuntimeOperationPrepare(input); err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
now := input.OccurredAtMS
|
|
if now <= 0 {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, errors.New("runtime operation occurred time is required")
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, fmt.Errorf("begin prepare interactive runtime operation: %w", err)
|
|
}
|
|
committed := false
|
|
defer func() {
|
|
if !committed {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
|
|
existingInteraction, found, err := getAgentInteractionTx(ctx, tx, input.WorkspaceID, input.AgentSessionID, input.TurnID, input.RequestID)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
if !found {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, ErrRuntimeOperationSubjectState
|
|
}
|
|
existingOperation, operationFound, err := getRuntimeOperationByIdentityTx(ctx, tx, input)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
if operationFound && existingOperation.OperationID != input.OperationID {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, ErrRuntimeOperationIdentityMismatch
|
|
}
|
|
claimPayload := input.Payload
|
|
if operationFound {
|
|
claimPayload = existingOperation.Payload
|
|
}
|
|
response := InteractionUpsert{
|
|
WorkspaceID: input.WorkspaceID, AgentSessionID: input.AgentSessionID,
|
|
RequestID: input.RequestID, TurnID: input.TurnID,
|
|
Kind: existingInteraction.Kind, Status: InteractionStatusAnswered,
|
|
ToolName: existingInteraction.ToolName, Input: existingInteraction.Input,
|
|
Metadata: existingInteraction.Metadata, Output: interactiveResponseOutput(claimPayload),
|
|
OccurredAtUnixMS: now,
|
|
}
|
|
interaction, transitionResult, err := s.upsertInteractionTx(ctx, tx, response, now)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
mutations := make([]TransactionMutation, 0, 2)
|
|
if transitionResult == InteractionTransitionApplied {
|
|
mutations = append(mutations, transactionMutation(
|
|
input.WorkspaceID, input.AgentSessionID, MutationEntityInteraction,
|
|
interactionMutationEntityID(input.TurnID, input.RequestID), "answer", interaction.UpdatedAtUnixMS,
|
|
))
|
|
}
|
|
if operationFound {
|
|
delta, err := s.commitTransaction(ctx, tx, input.WorkspaceID, mutations)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, fmt.Errorf("commit duplicate interactive runtime operation prepare: %w", err)
|
|
}
|
|
committed = true
|
|
existingOperation.CommitTransactionID = delta.TransactionID
|
|
existingOperation.CommitDelta = delta
|
|
return existingOperation, interaction, transitionResult, nil
|
|
}
|
|
if transitionResult != InteractionTransitionApplied {
|
|
if _, err := s.commitTransaction(ctx, tx, input.WorkspaceID, mutations); err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, fmt.Errorf("commit terminal interactive response observation: %w", err)
|
|
}
|
|
committed = true
|
|
return RuntimeOperation{}, interaction, transitionResult, nil
|
|
}
|
|
if err := validateRuntimeOperationSubjectTx(ctx, tx, input); err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
payloadJSON, err := marshalJSONMap(input.Payload)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
if byID, idFound, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID); err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
} else if idFound {
|
|
if !runtimeOperationIdentityMatches(byID, input) {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, ErrRuntimeOperationIdentityMismatch
|
|
}
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, ErrRuntimeOperationConflict
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `
|
|
INSERT INTO workspace_agent_runtime_operations (
|
|
operation_id, workspace_id, agent_session_id, kind, status, turn_id,
|
|
request_id, payload_json, next_attempt_at_unix_ms, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`, input.OperationID, input.WorkspaceID, input.AgentSessionID, input.Kind,
|
|
RuntimeOperationStatusPrepared, input.TurnID, input.RequestID,
|
|
payloadJSON, now, now, now); err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, fmt.Errorf("insert interactive runtime operation: %w", err)
|
|
}
|
|
operation, _, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, err
|
|
}
|
|
mutations = append(mutations, transactionMutation(
|
|
input.WorkspaceID, input.AgentSessionID, MutationEntityRuntimeOperation,
|
|
input.OperationID, "prepare", operation.Version,
|
|
))
|
|
delta, err := s.commitTransaction(ctx, tx, input.WorkspaceID, mutations)
|
|
if err != nil {
|
|
return RuntimeOperation{}, Interaction{}, InteractionTransitionConflict, fmt.Errorf("commit interactive runtime operation prepare: %w", err)
|
|
}
|
|
committed = true
|
|
operation.CommitTransactionID = delta.TransactionID
|
|
operation.CommitDelta = delta
|
|
return operation, interaction, transitionResult, nil
|
|
}
|
|
|
|
func interactiveResponseOutput(payload map[string]any) map[string]any {
|
|
responsePayload, _ := payload["payload"].(map[string]any)
|
|
return map[string]any{
|
|
"action": payloadString(payload, "action"),
|
|
"optionId": payloadString(payload, "optionId"),
|
|
"payload": cloneJSONMap(responsePayload),
|
|
}
|
|
}
|
|
|
|
func (s *Store) GetRuntimeOperation(ctx context.Context, workspaceID string, operationID string) (RuntimeOperation, bool, error) {
|
|
if s == nil || s.db == nil {
|
|
return RuntimeOperation{}, false, errors.New("workspace database is not initialized")
|
|
}
|
|
return getRuntimeOperation(ctx, s.db, strings.TrimSpace(workspaceID), strings.TrimSpace(operationID))
|
|
}
|
|
|
|
func (s *Store) ListClaimableRuntimeOperations(ctx context.Context, input ListClaimableRuntimeOperationsInput) ([]RuntimeOperation, error) {
|
|
if s == nil || s.db == nil {
|
|
return nil, errors.New("workspace database is not initialized")
|
|
}
|
|
limit := input.Limit
|
|
if limit <= 0 || limit > 1000 {
|
|
limit = 100
|
|
}
|
|
workspaceID := strings.TrimSpace(input.WorkspaceID)
|
|
query := runtimeOperationSelectSQL + `
|
|
WHERE ((status = ? AND next_attempt_at_unix_ms <= ?)
|
|
OR (status = ? AND lease_expires_at_unix_ms <= ?))`
|
|
args := []any{RuntimeOperationStatusPrepared, input.NowUnixMS, RuntimeOperationStatusLeased, input.NowUnixMS}
|
|
if workspaceID != "" {
|
|
query += ` AND workspace_id = ?`
|
|
args = append(args, workspaceID)
|
|
}
|
|
query += ` ORDER BY created_at_unix_ms ASC, operation_id ASC LIMIT ?`
|
|
args = append(args, limit)
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list claimable runtime operations: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
result := make([]RuntimeOperation, 0)
|
|
for rows.Next() {
|
|
op, err := scanRuntimeOperation(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, op)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate claimable runtime operations: %w", err)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) ClaimRuntimeOperationLease(ctx context.Context, input ClaimRuntimeOperationLeaseInput) (RuntimeOperation, bool, error) {
|
|
if s == nil || s.db == nil {
|
|
return RuntimeOperation{}, false, errors.New("workspace database is not initialized")
|
|
}
|
|
input.WorkspaceID = strings.TrimSpace(input.WorkspaceID)
|
|
input.OperationID = strings.TrimSpace(input.OperationID)
|
|
input.LeaseOwner = strings.TrimSpace(input.LeaseOwner)
|
|
if input.WorkspaceID == "" || input.OperationID == "" || input.LeaseOwner == "" || input.NowUnixMS <= 0 || input.LeaseExpiresAtMS <= input.NowUnixMS {
|
|
return RuntimeOperation{}, false, errors.New("valid workspace, operation, owner, now, and future lease expiry are required")
|
|
}
|
|
result, err := s.db.ExecContext(ctx, `
|
|
UPDATE workspace_agent_runtime_operations
|
|
SET status = ?, lease_owner = ?, lease_expires_at_unix_ms = ?, attempt = attempt + 1,
|
|
next_attempt_at_unix_ms = NULL, version = version + 1, last_error = '', updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND operation_id = ?
|
|
AND ((status = ? AND next_attempt_at_unix_ms <= ?)
|
|
OR (status = ? AND lease_expires_at_unix_ms <= ?))
|
|
`, RuntimeOperationStatusLeased, input.LeaseOwner, input.LeaseExpiresAtMS, input.NowUnixMS,
|
|
input.WorkspaceID, input.OperationID, RuntimeOperationStatusPrepared, input.NowUnixMS,
|
|
RuntimeOperationStatusLeased, input.NowUnixMS)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("claim runtime operation lease: %w", err)
|
|
}
|
|
claimed, err := rowsWereAffected(result, "claim runtime operation lease")
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
op, found, err := s.GetRuntimeOperation(ctx, input.WorkspaceID, input.OperationID)
|
|
if err != nil || !found {
|
|
return op, false, err
|
|
}
|
|
return op, claimed, nil
|
|
}
|
|
|
|
func (s *Store) ReleaseOrFailRuntimeOperation(ctx context.Context, input ReleaseOrFailRuntimeOperationInput) (RuntimeOperation, bool, error) {
|
|
if s == nil || s.db == nil {
|
|
return RuntimeOperation{}, false, errors.New("workspace database is not initialized")
|
|
}
|
|
status, resultValue, nextAttemptValue := RuntimeOperationStatusPrepared, any(nil), any(input.NextAttemptAtMS)
|
|
if input.Fail {
|
|
status, resultValue, nextAttemptValue = RuntimeOperationStatusFailed, RuntimeOperationResultFailed, nil
|
|
} else if input.NextAttemptAtMS < input.NowUnixMS {
|
|
return RuntimeOperation{}, false, errors.New("runtime operation retry time must not precede release time")
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
committed := false
|
|
defer func() {
|
|
if !committed {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_agent_runtime_operations
|
|
SET status = ?, result = ?, lease_owner = NULL, lease_expires_at_unix_ms = NULL,
|
|
next_attempt_at_unix_ms = ?, version = version + 1, last_error = ?, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND operation_id = ? AND status = ? AND lease_owner = ?
|
|
`, status, resultValue, nextAttemptValue, strings.TrimSpace(input.LastError), input.NowUnixMS,
|
|
strings.TrimSpace(input.WorkspaceID), strings.TrimSpace(input.OperationID), RuntimeOperationStatusLeased, strings.TrimSpace(input.LeaseOwner))
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("release or fail runtime operation: %w", err)
|
|
}
|
|
changed, err := rowsWereAffected(result, "release or fail runtime operation")
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
op, found, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID)
|
|
if err != nil || !found {
|
|
return op, false, err
|
|
}
|
|
mutations := []TransactionMutation{}
|
|
if changed {
|
|
operation := "release"
|
|
if input.Fail {
|
|
operation = "fail"
|
|
}
|
|
mutations = append(mutations, transactionMutation(
|
|
op.WorkspaceID, op.AgentSessionID, MutationEntityRuntimeOperation,
|
|
op.OperationID, operation, op.Version,
|
|
))
|
|
}
|
|
delta, err := s.commitTransaction(ctx, tx, op.WorkspaceID, mutations)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
committed = true
|
|
op.CommitTransactionID = delta.TransactionID
|
|
op.CommitDelta = delta
|
|
return op, changed, nil
|
|
}
|
|
|
|
func (s *Store) CheckpointRuntimeOperation(ctx context.Context, input CheckpointRuntimeOperationInput) (RuntimeOperation, bool, error) {
|
|
if s == nil || s.db == nil {
|
|
return RuntimeOperation{}, false, errors.New("workspace database is not initialized")
|
|
}
|
|
input.WorkspaceID = strings.TrimSpace(input.WorkspaceID)
|
|
input.OperationID = strings.TrimSpace(input.OperationID)
|
|
input.LeaseOwner = strings.TrimSpace(input.LeaseOwner)
|
|
if input.WorkspaceID == "" || input.OperationID == "" || input.LeaseOwner == "" || input.NowUnixMS <= 0 {
|
|
return RuntimeOperation{}, false, errors.New("workspace, operation, owner, and checkpoint time are required")
|
|
}
|
|
payloadJSON, err := marshalJSONMap(input.Payload)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("begin runtime operation checkpoint: %w", err)
|
|
}
|
|
committed := false
|
|
defer func() {
|
|
if !committed {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
current, found, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
if !found || current.Status != RuntimeOperationStatusLeased || current.LeaseOwner != input.LeaseOwner ||
|
|
current.LeaseExpiresAtMS <= input.NowUnixMS {
|
|
return current, false, ErrRuntimeOperationLeaseLost
|
|
}
|
|
switch current.Kind {
|
|
case RuntimeOperationKindInteractiveResponse:
|
|
if !interactiveResponseCheckpointIdentityEqual(current.Payload, input.Payload) {
|
|
return current, false, ErrRuntimeOperationSubjectState
|
|
}
|
|
case RuntimeOperationKindCancelTurn:
|
|
if !cancelRuntimeOperationCheckpointIdentityEqual(current.Payload, input.Payload) {
|
|
return current, false, ErrRuntimeOperationSubjectState
|
|
}
|
|
case RuntimeOperationKindPlanDecision:
|
|
if err := validatePlanDecisionOperationPayload(input.OperationID, input.Payload); err != nil {
|
|
return current, false, err
|
|
}
|
|
if !planDecisionCheckpointIdentityEqual(current.Payload, input.Payload) ||
|
|
!planDecisionStepCanAdvance(payloadString(current.Payload, "step"), payloadString(input.Payload, "step")) {
|
|
return current, false, ErrRuntimeOperationSubjectState
|
|
}
|
|
case RuntimeOperationKindEditRetry:
|
|
next, err := DecodeEditRetryOperationPayload(input.Payload)
|
|
if err != nil {
|
|
return current, false, err
|
|
}
|
|
if err := next.Validate(input.OperationID); err != nil {
|
|
return current, false, err
|
|
}
|
|
previous, err := DecodeEditRetryOperationPayload(current.Payload)
|
|
if err != nil {
|
|
return current, false, err
|
|
}
|
|
if !editRetryIdentityEqual(previous, next) ||
|
|
!editRetryCheckpointCanAdvance(previous.Checkpoint, next.Checkpoint) {
|
|
return current, false, ErrRuntimeOperationSubjectState
|
|
}
|
|
default:
|
|
return current, false, ErrRuntimeOperationSubjectState
|
|
}
|
|
result, err := tx.ExecContext(ctx, `
|
|
UPDATE workspace_agent_runtime_operations
|
|
SET payload_json = ?, version = version + 1, updated_at_unix_ms = ?
|
|
WHERE workspace_id = ? AND operation_id = ? AND status = ? AND lease_owner = ?
|
|
`, payloadJSON, input.NowUnixMS, input.WorkspaceID, input.OperationID,
|
|
RuntimeOperationStatusLeased, input.LeaseOwner)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("checkpoint runtime operation: %w", err)
|
|
}
|
|
changed, err := rowsWereAffected(result, "checkpoint runtime operation")
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
var pendingEvent RuntimeOperationEvent
|
|
if payloadString(current.Payload, "step") != "send_dispatched" && payloadString(input.Payload, "step") == "send_dispatched" {
|
|
if err := insertPlanDecisionUnknownNoticeTx(ctx, tx, current, input.NowUnixMS); err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
pendingEvent, err = insertRuntimeOperationEventTx(ctx, tx, current, RuntimeOperationEventPlanDecisionPending, map[string]any{
|
|
"turnId": current.TurnID, "requestId": current.RequestID,
|
|
"noticeMessageId": planDecisionNoticeMessageID(current.OperationID),
|
|
}, input.NowUnixMS)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
}
|
|
op, found, err := getRuntimeOperationTx(ctx, tx, input.WorkspaceID, input.OperationID)
|
|
if err != nil || !found {
|
|
return op, false, err
|
|
}
|
|
mutations := []TransactionMutation{
|
|
transactionMutation(input.WorkspaceID, op.AgentSessionID, MutationEntityRuntimeOperation, input.OperationID, "checkpoint", op.Version),
|
|
}
|
|
if pendingEvent.ID > 0 {
|
|
messageID := planDecisionNoticeMessageID(op.OperationID)
|
|
message, found, err := getAgentMessageForUpdate(ctx, tx, input.WorkspaceID, op.AgentSessionID, messageID)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, err
|
|
}
|
|
if !found {
|
|
return RuntimeOperation{}, false, ErrRuntimeOperationSubjectState
|
|
}
|
|
mutations = append(mutations,
|
|
transactionMutation(input.WorkspaceID, op.AgentSessionID, MutationEntityMessage, messageID, "upsert", int64(message.Version)),
|
|
transactionMutation(input.WorkspaceID, op.AgentSessionID, MutationEntityRuntimeEvent, fmt.Sprint(pendingEvent.ID), "insert", pendingEvent.ID),
|
|
)
|
|
}
|
|
delta, err := s.commitTransaction(ctx, tx, input.WorkspaceID, mutations)
|
|
if err != nil {
|
|
return RuntimeOperation{}, false, fmt.Errorf("commit runtime operation checkpoint: %w", err)
|
|
}
|
|
committed = true
|
|
op.CommitTransactionID = delta.TransactionID
|
|
op.CommitDelta = delta
|
|
return op, changed, nil
|
|
}
|
|
|
|
func insertPlanDecisionUnknownNoticeTx(ctx context.Context, tx *sql.Tx, operation RuntimeOperation, now int64) error {
|
|
payloadJSON, err := marshalJSONMap(map[string]any{
|
|
"kind": "agent_system_notice",
|
|
"noticeKind": "plan_implementation_pending_confirmation",
|
|
"severity": "warning",
|
|
"retryable": false,
|
|
"operationId": operation.OperationID,
|
|
"planTurnId": operation.TurnID,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(ctx, `
|
|
INSERT OR IGNORE INTO workspace_agent_messages (
|
|
workspace_id, agent_session_id, message_id, version, turn_id, role, kind,
|
|
status, payload_json, occurred_at_unix_ms, created_at_unix_ms, updated_at_unix_ms
|
|
) VALUES (?, ?, ?,
|
|
COALESCE((SELECT MAX(version) + 1 FROM workspace_agent_messages WHERE workspace_id = ? AND agent_session_id = ?), 1),
|
|
?, 'system', 'system', 'running', ?, ?, ?, ?)
|
|
`, operation.WorkspaceID, operation.AgentSessionID, planDecisionNoticeMessageID(operation.OperationID),
|
|
operation.WorkspaceID, operation.AgentSessionID,
|
|
operation.TurnID, payloadJSON, now, now, now)
|
|
if err != nil {
|
|
return fmt.Errorf("insert plan decision unknown notice: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func planDecisionNoticeMessageID(operationID string) string {
|
|
return "plan-decision:" + strings.TrimSpace(operationID) + ":status"
|
|
}
|
|
|
|
func planDecisionCheckpointIdentityEqual(current, next map[string]any) bool {
|
|
for _, key := range []string{"promptKind", "action", "idempotencyKey", "clientSubmitId"} {
|
|
if payloadString(current, key) != payloadString(next, key) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func planDecisionStepCanAdvance(current, next string) bool {
|
|
if current == next {
|
|
return true
|
|
}
|
|
return (current == "prepared" && next == "settings_applied") ||
|
|
(current == "settings_applied" && next == "send_dispatched") ||
|
|
(current == "send_dispatched" && next == "send_confirmed")
|
|
}
|
|
|
|
func editRetryCheckpointCanAdvance(current, next EditRetryCheckpoint) bool {
|
|
if current == next {
|
|
return true
|
|
}
|
|
return current == EditRetryCheckpointRollbackConfirmed &&
|
|
next == EditRetryCheckpointReplacementDispatched
|
|
}
|
|
|
|
func (s *Store) FindTurnByClientSubmitID(ctx context.Context, workspaceID string, agentSessionID string, clientSubmitID string) (string, bool, error) {
|
|
if s == nil || s.db == nil {
|
|
return "", false, errors.New("workspace database is not initialized")
|
|
}
|
|
var turnID string
|
|
err := s.db.QueryRowContext(ctx, `
|
|
SELECT turn_id
|
|
FROM workspace_agent_messages
|
|
WHERE workspace_id = ? AND agent_session_id = ?
|
|
AND turn_id IS NOT NULL AND length(turn_id) > 0
|
|
AND json_extract(payload_json, '$.clientSubmitId') = ?
|
|
ORDER BY version DESC LIMIT 1
|
|
`, strings.TrimSpace(workspaceID), strings.TrimSpace(agentSessionID), strings.TrimSpace(clientSubmitID)).Scan(&turnID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("find turn by client submit id: %w", err)
|
|
}
|
|
return strings.TrimSpace(turnID), true, nil
|
|
}
|
|
|
|
// RequeueLeasedRuntimeOperationsOnStartup invalidates every lease left by the
|
|
// previous daemon process. This is deliberately global: a new single daemon
|
|
// owner must recover even leases whose wall-clock expiry is still in the
|
|
// future instead of misclassifying their turns as generic stale work.
|
|
func (s *Store) RequeueLeasedRuntimeOperationsOnStartup(ctx context.Context, nowUnixMS int64) (int64, error) {
|
|
if s == nil || s.db == nil {
|
|
return 0, errors.New("workspace database is not initialized")
|
|
}
|
|
if nowUnixMS <= 0 {
|
|
return 0, errors.New("startup recovery time is required")
|
|
}
|
|
result, err := s.db.ExecContext(ctx, `
|
|
UPDATE workspace_agent_runtime_operations
|
|
SET status = ?, lease_owner = NULL, lease_expires_at_unix_ms = NULL,
|
|
next_attempt_at_unix_ms = ?, version = version + 1,
|
|
last_error = CASE
|
|
WHEN TRIM(last_error) = '' THEN 'startup recovery: previous runtime lease invalidated'
|
|
ELSE last_error || '; startup recovery: previous runtime lease invalidated'
|
|
END,
|
|
updated_at_unix_ms = ?
|
|
WHERE status = ?
|
|
`, RuntimeOperationStatusPrepared, nowUnixMS, nowUnixMS, RuntimeOperationStatusLeased)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("requeue leased runtime operations on startup: %w", err)
|
|
}
|
|
count, err := result.RowsAffected()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("requeue startup runtime operation rows affected: %w", err)
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
func validateRuntimeOperationSubjectTx(ctx context.Context, tx *sql.Tx, input RuntimeOperationPrepare) error {
|
|
turn, found, err := getAgentTurnTx(ctx, tx, input.WorkspaceID, input.AgentSessionID, input.TurnID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !found {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
if input.Kind == RuntimeOperationKindCancelTurn {
|
|
targets, err := cancelTargetsFromPayload(input.AgentSessionID, input.TurnID, input.Payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rootAgentSessionID := payloadString(input.Payload, "rootAgentSessionId")
|
|
for _, target := range targets {
|
|
targetTurn, targetFound, err := getAgentTurnTx(ctx, tx, input.WorkspaceID, target.AgentSessionID, target.TurnID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !targetFound {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
var kind string
|
|
var recordedRoot sql.NullString
|
|
if err := tx.QueryRowContext(ctx, `
|
|
SELECT session_kind, root_agent_session_id
|
|
FROM workspace_agent_sessions
|
|
WHERE workspace_id = ? AND agent_session_id = ? AND deleted_at_unix_ms = 0
|
|
`, input.WorkspaceID, target.AgentSessionID).Scan(&kind, &recordedRoot); err != nil {
|
|
return fmt.Errorf("read cancel target session relation: %w", err)
|
|
}
|
|
if (kind == SessionKindRoot && target.AgentSessionID != rootAgentSessionID) ||
|
|
(kind == SessionKindChild && strings.TrimSpace(recordedRoot.String) != rootAgentSessionID) ||
|
|
(kind != SessionKindRoot && kind != SessionKindChild) {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
if targetTurn.Phase == TurnPhaseSettled {
|
|
continue
|
|
}
|
|
var active sql.NullString
|
|
if err := tx.QueryRowContext(ctx, `SELECT active_turn_id FROM workspace_agent_sessions WHERE workspace_id = ? AND agent_session_id = ?`, input.WorkspaceID, target.AgentSessionID).Scan(&active); err != nil {
|
|
return fmt.Errorf("read runtime operation target: %w", err)
|
|
}
|
|
if !active.Valid || active.String != target.TurnID {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
if input.Kind == RuntimeOperationKindPlanDecision && payloadString(input.Payload, "promptKind") == "plan-implementation" {
|
|
if turn.Phase != TurnPhaseSettled || turn.Outcome != TurnOutcomeCompleted {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
var hasPlan int
|
|
err := tx.QueryRowContext(ctx, `
|
|
SELECT EXISTS(
|
|
SELECT 1 FROM workspace_agent_messages
|
|
WHERE workspace_id = ? AND agent_session_id = ? AND turn_id = ?
|
|
AND deleted_at_unix_ms = 0
|
|
AND (kind = 'plan' OR json_extract(payload_json, '$.messageKind') = 'plan')
|
|
)
|
|
`, input.WorkspaceID, input.AgentSessionID, input.TurnID).Scan(&hasPlan)
|
|
if err != nil {
|
|
return fmt.Errorf("validate plan decision evidence: %w", err)
|
|
}
|
|
if hasPlan != 1 {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
return nil
|
|
}
|
|
if input.Kind == RuntimeOperationKindEditRetry {
|
|
return validateEditRetryRuntimeOperationSubjectTx(ctx, tx, input, turn)
|
|
}
|
|
interaction, found, err := getAgentInteractionTx(ctx, tx, input.WorkspaceID, input.AgentSessionID, input.TurnID, input.RequestID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !found || interaction.TurnID != input.TurnID {
|
|
return ErrRuntimeOperationSubjectState
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func payloadString(payload map[string]any, key string) string {
|
|
value, _ := payload[key].(string)
|
|
return strings.TrimSpace(value)
|
|
}
|
|
|
|
func getRuntimeOperation(ctx context.Context, q rowQueryer, workspaceID string, operationID string) (RuntimeOperation, bool, error) {
|
|
return scanRuntimeOperationRow(q.QueryRowContext(ctx, runtimeOperationSelectSQL+` WHERE workspace_id = ? AND operation_id = ?`, workspaceID, operationID))
|
|
}
|
|
|
|
type rowQueryer interface {
|
|
QueryRowContext(context.Context, string, ...any) *sql.Row
|
|
}
|
|
|
|
func getRuntimeOperationTx(ctx context.Context, tx *sql.Tx, workspaceID string, operationID string) (RuntimeOperation, bool, error) {
|
|
return getRuntimeOperation(ctx, tx, workspaceID, operationID)
|
|
}
|
|
|
|
func getRuntimeOperationByIdentityTx(ctx context.Context, tx *sql.Tx, input RuntimeOperationPrepare) (RuntimeOperation, bool, error) {
|
|
query := runtimeOperationSelectSQL + ` WHERE workspace_id = ? AND agent_session_id = ? AND kind = ? AND turn_id = ? AND request_id = ?`
|
|
args := []any{input.WorkspaceID, input.AgentSessionID, input.Kind, input.TurnID, input.RequestID}
|
|
if input.Kind == RuntimeOperationKindCancelTurn {
|
|
query = runtimeOperationSelectSQL + ` WHERE workspace_id = ? AND agent_session_id = ? AND kind = ? AND turn_id = ? AND request_id IS NULL`
|
|
args = []any{input.WorkspaceID, input.AgentSessionID, input.Kind, input.TurnID}
|
|
}
|
|
if input.Kind == RuntimeOperationKindEditRetry {
|
|
query = runtimeOperationSelectSQL + ` WHERE workspace_id = ? AND agent_session_id = ? AND kind = ? AND request_id = ?`
|
|
args = []any{input.WorkspaceID, input.AgentSessionID, input.Kind, input.RequestID}
|
|
}
|
|
return scanRuntimeOperationRow(tx.QueryRowContext(ctx, query, args...))
|
|
}
|
|
|
|
func runtimeOperationIdentityMatches(operation RuntimeOperation, input RuntimeOperationPrepare) bool {
|
|
return operation.WorkspaceID == input.WorkspaceID &&
|
|
operation.AgentSessionID == input.AgentSessionID &&
|
|
operation.Kind == input.Kind &&
|
|
operation.TurnID == input.TurnID &&
|
|
operation.RequestID == input.RequestID
|
|
}
|
|
|
|
func scanRuntimeOperationRow(row *sql.Row) (RuntimeOperation, bool, error) {
|
|
op, err := scanRuntimeOperation(row)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return RuntimeOperation{}, false, nil
|
|
}
|
|
return op, err == nil, err
|
|
}
|
|
|
|
func scanRuntimeOperation(scanner rowScanner) (RuntimeOperation, error) {
|
|
var op RuntimeOperation
|
|
var payloadJSON string
|
|
err := scanner.Scan(&op.OperationID, &op.WorkspaceID, &op.AgentSessionID, &op.Kind, &op.Status, &op.Result,
|
|
&op.TurnID, &op.RequestID, &payloadJSON, &op.LeaseOwner, &op.LeaseExpiresAtMS, &op.NextAttemptAtMS, &op.Attempt,
|
|
&op.Version, &op.LastError, &op.CreatedAtUnixMS, &op.UpdatedAtUnixMS, &op.CompletedAtUnixMS)
|
|
if err != nil {
|
|
return RuntimeOperation{}, err
|
|
}
|
|
op.Payload, err = unmarshalJSONMap(payloadJSON)
|
|
if err != nil {
|
|
return RuntimeOperation{}, fmt.Errorf("decode runtime operation payload: %w", err)
|
|
}
|
|
if op.Kind == RuntimeOperationKindEditRetry {
|
|
if err := validateEditRetryOperationPayload(op.OperationID, op.Payload); err != nil {
|
|
return RuntimeOperation{}, fmt.Errorf("validate stored edit retry operation payload: %w", err)
|
|
}
|
|
}
|
|
return op, nil
|
|
}
|
|
|
|
func jsonMapsEqual(left map[string]any, right map[string]any) bool {
|
|
leftJSON, leftErr := marshalJSONMap(left)
|
|
rightJSON, rightErr := marshalJSONMap(right)
|
|
return leftErr == nil && rightErr == nil && leftJSON == rightJSON
|
|
}
|
|
|
|
func interactiveResponseCheckpointIdentityEqual(previous, next map[string]any) bool {
|
|
previousIdentity := cloneJSONMap(previous)
|
|
nextIdentity := cloneJSONMap(next)
|
|
for _, key := range []string{"followUpPrompt", "followUpClientSubmitId", "followUpDisposition"} {
|
|
delete(previousIdentity, key)
|
|
delete(nextIdentity, key)
|
|
}
|
|
return jsonMapsEqual(previousIdentity, nextIdentity)
|
|
}
|
|
|
|
func cancelRuntimeOperationCheckpointIdentityEqual(previous, next map[string]any) bool {
|
|
nextDeliveryUnconfirmed, ok := next[CancelRuntimeOperationDeliveryUnconfirmedPayloadKey].(bool)
|
|
if !ok || !nextDeliveryUnconfirmed {
|
|
return false
|
|
}
|
|
if existing, exists := previous[CancelRuntimeOperationDeliveryUnconfirmedPayloadKey]; exists {
|
|
deliveryUnconfirmed, ok := existing.(bool)
|
|
if !ok || !deliveryUnconfirmed {
|
|
return false
|
|
}
|
|
}
|
|
previousIdentity := cloneJSONMap(previous)
|
|
nextIdentity := cloneJSONMap(next)
|
|
delete(previousIdentity, CancelRuntimeOperationDeliveryUnconfirmedPayloadKey)
|
|
delete(nextIdentity, CancelRuntimeOperationDeliveryUnconfirmedPayloadKey)
|
|
return jsonMapsEqual(previousIdentity, nextIdentity)
|
|
}
|