186 lines
11 KiB
Go
186 lines
11 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/contracts/epistemics"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type EpistemicsRepository struct{ pool *pgxpool.Pool }
|
|
|
|
func OpenEpistemics(ctx context.Context, config ContinuityConfig) (*EpistemicsRepository, error) {
|
|
pool, err := openPool(ctx, config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &EpistemicsRepository{pool: pool}, nil
|
|
}
|
|
|
|
func (r *EpistemicsRepository) Close() { r.pool.Close() }
|
|
|
|
func (r *EpistemicsRepository) Ping(ctx context.Context) error {
|
|
return r.pool.Ping(ctx)
|
|
}
|
|
|
|
func (r *EpistemicsRepository) SaveSystemExpectation(ctx context.Context, value epistemics.SystemExpectation) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO epistemics.system_expectations (
|
|
expectation_id, original_intent_id, source_utterance_id, expected_outcome, expected_effect,
|
|
confidence_millis, evidence_basis, recorded_at_unix_millis)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT DO NOTHING`,
|
|
value.ID, value.OriginalIntentID, value.SourceUtteranceID, value.ExpectedOutcome, value.ExpectedEffect,
|
|
value.ConfidenceMillis, value.EvidenceBasis, value.RecordedAtUnixMillis)
|
|
return err
|
|
}
|
|
|
|
func (r *EpistemicsRepository) RecentExperienceRecords(ctx context.Context, limit int) ([]epistemics.ExperienceRecord, error) {
|
|
if limit < 1 || limit > 100 {
|
|
return nil, fmt.Errorf("invalid experience limit")
|
|
}
|
|
rows, err := r.pool.Query(ctx, `SELECT
|
|
experience_id,source_kind,original_intent_id,source_utterance_id,context_refs,help_method,help_method_ref,
|
|
capability_implementation_ref,capability_usage_observation_ref,capability_gap_ref,skill_ref,expectation_id,
|
|
expected_outcome,actual_outcome,outcome_ref,outcome_detail,human_response_ref,correction_utterance_id,
|
|
verification_evidence,confidence_millis,recorded_at_unix_millis
|
|
FROM epistemics.experience_records ORDER BY recorded_at_unix_millis DESC, experience_id DESC LIMIT $1`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []epistemics.ExperienceRecord
|
|
for rows.Next() {
|
|
var value epistemics.ExperienceRecord
|
|
if err := rows.Scan(&value.ID, &value.SourceKind, &value.OriginalIntentID, &value.SourceUtteranceID, &value.ContextRefs, &value.HelpMethod, &value.HelpMethodRef,
|
|
&value.CapabilityImplementationRef, &value.CapabilityUsageObservationRef, &value.CapabilityGapRef, &value.SkillRef, &value.ExpectationID,
|
|
&value.ExpectedOutcome, &value.ActualOutcome, &value.OutcomeRef, &value.OutcomeDetail, &value.HumanResponseRef, &value.CorrectionUtteranceID,
|
|
&value.VerificationEvidence, &value.ConfidenceMillis, &value.RecordedAtUnixMillis); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, value)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *EpistemicsRepository) RecentReflectionCandidates(ctx context.Context, limit int) ([]epistemics.ReflectionCandidate, error) {
|
|
if limit < 1 || limit > 100 {
|
|
return nil, fmt.Errorf("invalid reflection limit")
|
|
}
|
|
rows, err := r.pool.Query(ctx, `SELECT
|
|
reflection_candidate_id,trigger,state,source_experience_id,original_intent_id,source_utterance_id,
|
|
summary,evidence_refs,requires_cognition,recorded_at_unix_millis,revision,claimed_by,claim_expires_at_unix_millis
|
|
FROM epistemics.reflection_candidates ORDER BY recorded_at_unix_millis DESC, reflection_candidate_id DESC LIMIT $1`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []epistemics.ReflectionCandidate
|
|
for rows.Next() {
|
|
var value epistemics.ReflectionCandidate
|
|
if err := rows.Scan(&value.ID, &value.Trigger, &value.State, &value.SourceExperienceID, &value.OriginalIntentID, &value.SourceUtteranceID,
|
|
&value.Summary, &value.EvidenceRefs, &value.RequiresCognition, &value.RecordedAtUnixMillis, &value.Revision, &value.ClaimedBy, &value.ClaimExpiresAtUnixMillis); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, value)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *EpistemicsRepository) ClaimReflectionCandidate(ctx context.Context, workerID string, nowUnixMillis, leaseMillis int64) (epistemics.ReflectionCandidate, error) {
|
|
if workerID == "" || leaseMillis < 1 {
|
|
return epistemics.ReflectionCandidate{}, fmt.Errorf("invalid reflection claim")
|
|
}
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return epistemics.ReflectionCandidate{}, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
var value epistemics.ReflectionCandidate
|
|
err = tx.QueryRow(ctx, `SELECT reflection_candidate_id,trigger,state,source_experience_id,original_intent_id,source_utterance_id,summary,evidence_refs,requires_cognition,recorded_at_unix_millis,revision,claimed_by,claim_expires_at_unix_millis
|
|
FROM epistemics.reflection_candidates
|
|
WHERE state='OPEN' AND (claim_expires_at_unix_millis=0 OR claim_expires_at_unix_millis < $1)
|
|
ORDER BY recorded_at_unix_millis,reflection_candidate_id
|
|
FOR UPDATE SKIP LOCKED LIMIT 1`, nowUnixMillis).Scan(&value.ID, &value.Trigger, &value.State, &value.SourceExperienceID, &value.OriginalIntentID, &value.SourceUtteranceID, &value.Summary, &value.EvidenceRefs, &value.RequiresCognition, &value.RecordedAtUnixMillis, &value.Revision, &value.ClaimedBy, &value.ClaimExpiresAtUnixMillis)
|
|
if err != nil {
|
|
return epistemics.ReflectionCandidate{}, err
|
|
}
|
|
value.State = epistemics.ReflectionCandidateEvaluating
|
|
value.ClaimedBy = workerID
|
|
value.ClaimExpiresAtUnixMillis = nowUnixMillis + leaseMillis
|
|
_, err = tx.Exec(ctx, `UPDATE epistemics.reflection_candidates SET state=$1,claimed_by=$2,claim_expires_at_unix_millis=$3 WHERE reflection_candidate_id=$4 AND state='OPEN'`, value.State, value.ClaimedBy, value.ClaimExpiresAtUnixMillis, value.ID)
|
|
if err != nil {
|
|
return epistemics.ReflectionCandidate{}, err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return epistemics.ReflectionCandidate{}, err
|
|
}
|
|
return value, nil
|
|
}
|
|
|
|
func (r *EpistemicsRepository) RecoverReflectionClaims(ctx context.Context, nowUnixMillis int64) (int64, error) {
|
|
result, err := r.pool.Exec(ctx, `UPDATE epistemics.reflection_candidates SET state='OPEN',claimed_by='',claim_expires_at_unix_millis=0 WHERE state='EVALUATING' AND claim_expires_at_unix_millis > 0 AND claim_expires_at_unix_millis < $1`, nowUnixMillis)
|
|
return result.RowsAffected(), err
|
|
}
|
|
|
|
func (r *EpistemicsRepository) ResolveReflectionCandidate(ctx context.Context, id epistemics.ReflectionCandidateID, workerID string, nowUnixMillis int64) error {
|
|
if id == "" || workerID == "" {
|
|
return fmt.Errorf("invalid reflection resolution")
|
|
}
|
|
result, err := r.pool.Exec(ctx, `UPDATE epistemics.reflection_candidates SET state='RESOLVED',claimed_by='',claim_expires_at_unix_millis=0,revision=revision+1 WHERE reflection_candidate_id=$1 AND state='EVALUATING' AND claimed_by=$2 AND claim_expires_at_unix_millis >= $3`, id, workerID, nowUnixMillis)
|
|
if err == nil && result.RowsAffected() != 1 {
|
|
return fmt.Errorf("reflection candidate claim lost")
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (r *EpistemicsRepository) WaitForReflectionCandidate(ctx context.Context, interval time.Duration) error {
|
|
if interval <= 0 {
|
|
return fmt.Errorf("invalid reflection poll interval")
|
|
}
|
|
_, err := r.pool.Exec(ctx, `SELECT pg_sleep($1)`, interval.Seconds())
|
|
return err
|
|
}
|
|
|
|
func (r *EpistemicsRepository) SaveExperienceRecord(ctx context.Context, value epistemics.ExperienceRecord) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO epistemics.experience_records (
|
|
experience_id, source_kind, original_intent_id, source_utterance_id, context_refs, help_method, help_method_ref,
|
|
capability_implementation_ref, capability_usage_observation_ref, capability_gap_ref, skill_ref, expectation_id,
|
|
expected_outcome, actual_outcome, outcome_ref, outcome_detail, human_response_ref, correction_utterance_id,
|
|
verification_evidence, confidence_millis, recorded_at_unix_millis)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21) ON CONFLICT DO NOTHING`,
|
|
value.ID, value.SourceKind, value.OriginalIntentID, value.SourceUtteranceID, value.ContextRefs, value.HelpMethod, value.HelpMethodRef,
|
|
value.CapabilityImplementationRef, value.CapabilityUsageObservationRef, value.CapabilityGapRef, value.SkillRef, value.ExpectationID,
|
|
value.ExpectedOutcome, value.ActualOutcome, value.OutcomeRef, value.OutcomeDetail, value.HumanResponseRef, value.CorrectionUtteranceID,
|
|
value.VerificationEvidence, value.ConfidenceMillis, value.RecordedAtUnixMillis)
|
|
return err
|
|
}
|
|
|
|
func (r *EpistemicsRepository) SaveReflectionCandidate(ctx context.Context, value epistemics.ReflectionCandidate) error {
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO epistemics.reflection_candidates (
|
|
reflection_candidate_id, trigger, state, source_experience_id, original_intent_id, source_utterance_id,
|
|
summary, evidence_refs, requires_cognition, recorded_at_unix_millis, revision)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT DO NOTHING`,
|
|
value.ID, value.Trigger, value.State, value.SourceExperienceID, value.OriginalIntentID, value.SourceUtteranceID,
|
|
value.Summary, value.EvidenceRefs, value.RequiresCognition, value.RecordedAtUnixMillis, value.Revision)
|
|
return err
|
|
}
|
|
|
|
func (r *EpistemicsRepository) SaveTruthBundle(ctx context.Context, value epistemics.TruthBundle, recordedAtUnixMillis int64) error {
|
|
if value.Source.ID == "" || value.Snapshot.ID == "" || value.Excerpt.ID == "" || value.Claim.ID == "" || value.Belief.ID == "" || value.Citation.ID == "" || recordedAtUnixMillis < 1 {
|
|
return fmt.Errorf("invalid truth bundle")
|
|
}
|
|
_, err := r.pool.Exec(ctx, `INSERT INTO epistemics.truth_bundles (
|
|
bundle_id,source_id,source_kind,source_locator,source_title,source_publisher,snapshot_id,snapshot_source_id,snapshot_content_digest,snapshot_captured_at_unix_millis,snapshot_freshness,
|
|
excerpt_id,excerpt_snapshot_id,excerpt_text,excerpt_start,excerpt_end,observation_id,observation_observed_at_unix_millis,claim_id,claim_statement,claim_observation_id,claim_citation_id,claim_verification,
|
|
belief_id,belief_claim_id,belief_state,citation_id,citation_excerpt_id,citation_evidence,citation_coverage,citation_conflict,citation_resolved_at_unix_millis,recorded_at_unix_millis)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33)
|
|
ON CONFLICT (bundle_id) DO UPDATE SET recorded_at_unix_millis=EXCLUDED.recorded_at_unix_millis`,
|
|
"bundle:"+string(value.Citation.ID), value.Source.ID, value.Source.Kind, value.Source.Locator, value.Source.Title, value.Source.Publisher,
|
|
value.Snapshot.ID, value.Snapshot.SourceID, value.Snapshot.ContentDigest, value.Snapshot.CapturedAtUnixMillis, value.Snapshot.Freshness,
|
|
value.Excerpt.ID, value.Excerpt.SnapshotID, value.Excerpt.Text, value.Excerpt.Start, value.Excerpt.End,
|
|
value.Observation.ID, value.Observation.ObservedAt, value.Claim.ID, value.Claim.Statement, value.Claim.ObservationID, value.Claim.CitationID, value.Claim.Verification,
|
|
value.Belief.ID, value.Belief.ClaimID, value.Belief.State, value.Citation.ID, value.Citation.ExcerptID, value.Citation.Evidence, value.Citation.Coverage, value.Citation.Conflict, value.Citation.ResolvedAtUnixMillis, recordedAtUnixMillis)
|
|
return err
|
|
}
|