sqlite

func (*Store) AppendResponseEvent ¶

func (s *Store) AppendResponseEvent(
	ctx context.Context,
	key storeapi.RequestKey,
	kind string,
	data []byte,
) (storeapi.ResponseEvent, error)

func (*Store) AppendWorkerResponseEvent ¶

func (s *Store) AppendWorkerResponseEvent(
	ctx context.Context,
	key storeapi.RequestKey,
	kind string,
	eventID string,
	eventDigest string,
	data []byte,
) (storeapi.ResponseEvent, error)

AppendWorkerResponseEvent appends one durable stage per worker event ID and kind. Exact retries return the original row; reusing an ID with different payload or digest fails closed. This is the online counterpart of the worker-event receipt and prevents a transient later-stage failure from poisoning recovery with duplicate step/applied rows.

func (*Store) BeginRequest ¶

func (s *Store) BeginRequest(ctx context.Context, input storeapi.BeginRequestInput) (storeapi.BeginRequestResult, error)

func (*Store) BeginResponse ¶

func (s *Store) BeginResponse(ctx context.Context, key storeapi.RequestKey) (storeapi.Request, error)

BeginResponse durably commits the 200/SSE boundary before an assignment can become visible to a worker. A retry that races the original request waits for this decision instead of guessing from the existence of response frames.

func (*Store) BeginToolExecution ¶

func (s *Store) BeginToolExecution(
	ctx context.Context,
	key storeapi.ToolExecutionKey,
	requestDigest string,
) (storeapi.BeginToolExecutionResult, error)

func (*Store) Close ¶

func (s *Store) Close() error

func (*Store) CompleteAuditMetadata ¶

func (s *Store) CompleteAuditMetadata(
	ctx context.Context,
	id string,
	pendingMS int64,
	genMS int64,
	errorMessage string,
) (storeapi.AuditMetadata, error)

func (*Store) CompleteNonStreamingResponse ¶

func (s *Store) CompleteNonStreamingResponse(
	ctx context.Context,
	key storeapi.RequestKey,
	response storeapi.ResponseDecision,
) (storeapi.Request, error)

CompleteNonStreamingResponse atomically chooses the final HTTP status/body and closes a non-streaming request. Until this operation commits, duplicate callers observe an undecided response and wait; afterward every replay gets the exact same bytes. Canonical Stream=false is the durable response-mode marker, so no assignment can be mistaken for a streaming response on restart.

func (*Store) CompleteRequest ¶

func (s *Store) CompleteRequest(ctx context.Context, key storeapi.RequestKey) error

func (*Store) CompleteToolExecution ¶

func (s *Store) CompleteToolExecution(
	ctx context.Context,
	key storeapi.ToolExecutionKey,
	resultPayload []byte,
	isError bool,
) (storeapi.ToolExecution, error)

func (*Store) CreateAPIToken ¶

func (s *Store) CreateAPIToken(ctx context.Context, token storeapi.APIToken) error

func (*Store) CreateAuditMetadata ¶

func (s *Store) CreateAuditMetadata(
	ctx context.Context,
	metadata storeapi.AuditMetadata,
) (storeapi.AuditMetadata, error)

func (*Store) FailRequest ¶

func (s *Store) FailRequest(
	ctx context.Context,
	key storeapi.RequestKey,
	expected completion.State,
	response storeapi.ResponseDecision,
) (storeapi.Request, error)

FailRequest atomically chooses a terminal pre-stream HTTP response and completes only that request. The task remains admitted/reconciled and keeps the failed canonical digest as a narrow retry grant: a new idempotency key may retry that exact request, but cannot smuggle a different continuation into a turn which never crossed its response boundary.

func (*Store) FindAPITokenByHash ¶

func (s *Store) FindAPITokenByHash(ctx context.Context, tokenHash []byte) (storeapi.APIToken, error)

func (*Store) FindOpenHarnessTask ¶

func (s *Store) FindOpenHarnessTask(
	ctx context.Context,
	affinity storeapi.TaskAffinity,
) (storeapi.Task, error)

func (*Store) GetAuditMetadata ¶

func (s *Store) GetAuditMetadata(ctx context.Context, id string) (storeapi.AuditMetadata, error)

func (*Store) GetAuditPayload ¶

func (s *Store) GetAuditPayload(ctx context.Context, auditID, kind string) (storeapi.AuditPayload, error)

func (*Store) GetTask ¶

func (s *Store) GetTask(ctx context.Context, key storeapi.TaskKey) (storeapi.Task, error)

func (*Store) ListRecoverableRequests ¶

func (s *Store) ListRecoverableRequests(ctx context.Context) (storeapi.RecoverySnapshot, error)

ListRecoverableRequests returns incomplete responses whose owning task can still make progress, terminal tasks with a durable response step awaiting publication, and completed responses whose final durable receipt was not committed. It reads task identity and canonical request payloads from one snapshot so callers can reconstruct assignments after a restart.

func (*Store) ListResponseEvents ¶

func (s *Store) ListResponseEvents(ctx context.Context, key storeapi.RequestKey, after int64) ([]storeapi.ResponseEvent, error)

func (*Store) ListWorkerEventStages ¶

func (s *Store) ListWorkerEventStages(
	ctx context.Context,
	key storeapi.RequestKey,
	eventID string,
) ([]storeapi.ResponseEvent, error)

ListWorkerEventStages uses the event-id index for online step/applied reconciliation.

func (*Store) LookupRequest ¶

func (s *Store) LookupRequest(
	ctx context.Context,
	key storeapi.RequestKey,
	requestDigest string,
) (storeapi.BeginRequestResult, error)

func (*Store) LookupToolExecution ¶

func (s *Store) LookupToolExecution(
	ctx context.Context,
	key storeapi.ToolExecutionKey,
) (storeapi.ToolExecution, error)

func (*Store) LookupWorkerEventReceipt ¶

func (s *Store) LookupWorkerEventReceipt(
	ctx context.Context,
	key storeapi.RequestKey,
	eventID string,
) (storeapi.WorkerEventReceipt, error)

func (*Store) Ping ¶

func (s *Store) Ping(ctx context.Context) error

Ping proves that the gateway database can query its current schema now. It is deliberately stronger than merely checking that sql.DB still owns a driver handle (or evaluating SELECT 1 without touching the database file): readiness must fail when SQLite becomes unavailable or damaged at runtime.

func (*Store) PurgeExpiredAuditPayloads ¶

func (s *Store) PurgeExpiredAuditPayloads(ctx context.Context, before time.Time) (int64, error)

func (*Store) PurgeExpiredCompletionPayloads ¶

func (s *Store) PurgeExpiredCompletionPayloads(ctx context.Context, completedBefore time.Time) (int64, error)

PurgeExpiredCompletionPayloads replaces completed request replay data with immutable idempotency tombstones. Active and otherwise recoverable requests are deliberately excluded: a completed response with an unacknowledged worker step still needs its canonical request and event log after restart.

The caller supplies the completion cutoff (normally now minus the configured replay grace). A malformed completed row without completed_at conservatively uses created_at rather than retaining payload forever. Audit retention is separate.

func (*Store) QuarantineRecoveryRequest ¶

func (s *Store) QuarantineRecoveryRequest(
	ctx context.Context,
	quarantine storeapi.RecoveryQuarantine,
) error

QuarantineRecoveryRequest turns one corrupt recovery record into a durable, finite replay without decoding canonical_request. The request digest remains untouched, so the existing idempotency key still rejects a different body. A committed 200 is irreversible: in that case the transaction appends one explicit terminal SSE frame and completes the response instead of rewriting the HTTP decision.

func (*Store) ReadResponse ¶

func (s *Store) ReadResponse(
	ctx context.Context,
	key storeapi.RequestKey,
	after int64,
) (storeapi.ResponseRead, error)

ReadResponse returns response state and cursor events from one SQLite read transaction. The streaming handler relies on this snapshot guarantee: once response_complete is visible, the terminal applied event must be visible in the same result.

func (*Store) RecordWorkerEventReceipt ¶

func (s *Store) RecordWorkerEventReceipt(
	ctx context.Context,
	key storeapi.RequestKey,
	eventID string,
	workerID string,
	digest string,
) (storeapi.WorkerEventReceipt, error)

func (*Store) RevokeAPIToken ¶

func (s *Store) RevokeAPIToken(ctx context.Context, keyID string) error

func (*Store) StoreAuditPayload ¶

func (s *Store) StoreAuditPayload(
	ctx context.Context,
	payload storeapi.AuditPayload,
) (storeapi.AuditPayload, error)

func (*Store) TransitionTask ¶

func (s *Store) TransitionTask(
	ctx context.Context,
	key storeapi.TaskKey,
	expected completion.State,
	next completion.State,
	workerID string,
) (storeapi.Task, error)