2026-08-26 06:41:07 +08:00
package extraction
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
2026-10-08 10:06:44 +08:00
"time"
2026-08-26 06:41:07 +08:00
"unicode/utf8"
"data-terminal-cloud/internal/contentpolicy"
"data-terminal-cloud/internal/outbox"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var ErrConversationContentNotReady = errors . New ( "conversation content is not ready for extraction" )
var ErrConversationContentChanged = errors . New ( "conversation content changed while extraction was running" )
type Processor struct {
pool * pgxpool . Pool
generator Generator
safety contentpolicy . Classifier
}
type turn struct {
id , publicID , speaker , text string
interrupted bool
2026-10-08 10:06:44 +08:00
sourceRevision int64
2026-08-26 06:41:07 +08:00
}
type extractionSnapshot struct {
exists bool
tenantID , subjectID , status string
storyExtractionAllowed bool
turns [] turn
2026-08-29 21:56:51 +08:00
maxSequence int64
2026-08-26 06:41:07 +08:00
digest string
}
type queryer interface {
QueryRow ( context . Context , string , ... any ) pgx . Row
Query ( context . Context , string , ... any ) ( pgx . Rows , error )
}
func NewProcessor ( pool * pgxpool . Pool , generators ... Generator ) ( * Processor , error ) {
if pool == nil {
return nil , errors . New ( "database pool is required" )
}
var generator Generator = DeterministicGenerator {}
if len ( generators ) > 1 || ( len ( generators ) == 1 && generators [ 0 ] == nil ) {
return nil , errors . New ( "one valid extraction generator is required" )
}
if len ( generators ) == 1 {
generator = generators [ 0 ]
}
return & Processor { pool : pool , generator : generator , safety : contentpolicy . LocalClassifier {}}, nil
}
func NewProcessorWithSafety ( pool * pgxpool . Pool , generator Generator , safety contentpolicy . Classifier ) ( * Processor , error ) {
if generator == nil || safety == nil {
return nil , errors . New ( "valid extraction generator and content safety classifier are required" )
}
processor , err := NewProcessor ( pool , generator )
if err != nil {
return nil , err
}
processor . safety = safety
return processor , nil
}
func ( p * Processor ) Handle ( ctx context . Context , event outbox . Event ) error {
snapshot , err := loadExtractionSnapshot ( ctx , p . pool , event . AggregateID , false )
if err != nil {
return err
}
if ! snapshot . exists {
return nil
}
if snapshot . status != "COMPLETED" && snapshot . status != "FAILED" {
return outbox . Retryable ( ErrConversationContentNotReady )
}
if ! snapshot . storyExtractionAllowed {
return nil
}
2026-08-29 21:56:51 +08:00
generationVersion := requestedGenerationVersion ( event . Payload )
if generationVersion < snapshot . maxSequence {
return nil
}
2026-08-26 06:41:07 +08:00
sourceTurns := subjectTurns ( snapshot . turns )
if len ( sourceTurns ) == 0 {
2026-08-31 11:29:56 +08:00
// Terminal conversations cannot receive more subject turns. Treat the
// extraction request as complete so permanently empty conversations do
// not cycle through retry and dead-letter queues forever.
return nil
2026-08-26 06:41:07 +08:00
}
2026-09-06 20:16:01 +08:00
// Only the subject's final turns can become a summary, story, fact, or
// continuation cue. Agent turns may contain questions or generated wording
// and must never be promoted into the subject's personal history.
2026-10-08 10:06:44 +08:00
generated , previous , err := p . generateEventsSafely ( ctx , snapshot , time . Now (). In ( time . FixedZone ( "CST" , 8 * 3600 )). Format ( "2006-01-02" ))
2026-08-26 06:41:07 +08:00
if err != nil {
return err
}
// The model call above is deliberately outside a database transaction. The
// conversation row is locked only for the short revalidation and write phase.
tx , err := p . pool . BeginTx ( ctx , pgx . TxOptions {})
if err != nil {
return err
}
defer tx . Rollback ( context . Background ())
2026-10-08 10:06:44 +08:00
if generated . Stories != nil || generated . DailyEvents != nil {
if err = lockStoryContext ( ctx , tx , snapshot , previous ); err != nil {
return err
}
}
2026-08-26 06:41:07 +08:00
current , err := loadExtractionSnapshot ( ctx , tx , event . AggregateID , true )
if err != nil {
return err
}
if ! current . exists {
return tx . Commit ( ctx )
}
if ! current . storyExtractionAllowed {
return tx . Commit ( ctx )
}
if current . status != "COMPLETED" && current . status != "FAILED" {
return outbox . Retryable ( ErrConversationContentNotReady )
}
if current . digest != snapshot . digest {
return outbox . Retryable ( ErrConversationContentChanged )
}
2026-10-08 10:06:44 +08:00
return storeGeneration ( ctx , tx , event . AggregateID , current , generated , generationVersion , previous )
2026-08-26 06:41:07 +08:00
}
2026-08-30 10:51:32 +08:00
func ( p * Processor ) generateSafely ( ctx context . Context , content string , tenantIDs ... string ) ( GenerationResult , error ) {
tenantID := ""
if len ( tenantIDs ) > 0 {
tenantID = tenantIDs [ 0 ]
}
2026-08-26 06:41:07 +08:00
decision , err := p . safety . Classify ( ctx , content )
if err != nil {
return GenerationResult {}, err
}
if decision . AllowOrdinaryStory {
2026-08-30 10:51:32 +08:00
generated , generateErr := p . generator . Generate ( ctx , GenerationInput { TenantID : tenantID , Content : content })
2026-08-26 06:41:07 +08:00
if generateErr != nil {
return GenerationResult {}, generateErr
}
outputDecision , safetyErr := p . safety . Classify ( ctx , generatedSafetyContent ( generated ))
if safetyErr != nil {
return GenerationResult {}, safetyErr
}
if ! outputDecision . AllowOrdinaryStory {
return sensitiveGeneration ( outputDecision ), nil
}
return generated , nil
}
return sensitiveGeneration ( decision ), nil
}
func generatedSafetyContent ( result GenerationResult ) string {
2026-10-08 10:06:44 +08:00
parts := [] string { result . Summary , strings . Join ( result . Keywords , "、" ), result . OpenTopic , result . StoryTitle , result . StoryContent , result . FactContent }
for _ , s := range result . Stories {
parts = append ( parts , s . Title , s . Content , s . FactContent , strings . Join ( s . NewDetails , "\n" ))
}
for _ , d := range result . DailyEvents {
parts = append ( parts , d . Title , d . Content )
}
return strings . Join ( parts , "\n" )
2026-08-26 06:41:07 +08:00
}
func loadExtractionSnapshot ( ctx context . Context , q queryer , conversationID string , lock bool ) ( extractionSnapshot , error ) {
lockClause := ""
if lock {
lockClause = " FOR UPDATE OF c"
}
var snapshot extractionSnapshot
var consentsJSON [] byte
err := q . QueryRow ( ctx , `
SELECT c.tenant_id::text,c.subject_id::text,c.status,cs.effective_consents
FROM conversation c JOIN consent_snapshot cs ON cs.id=c.consent_snapshot_id AND cs.tenant_id=c.tenant_id
JOIN subject_profile sp ON sp.id=c.subject_id AND sp.tenant_id=c.tenant_id AND sp.status IN ('active','suspended')
WHERE c.id=$1::uuid` + lockClause , conversationID ). Scan ( & snapshot . tenantID , & snapshot . subjectID , & snapshot . status , & consentsJSON )
if errors . Is ( err , pgx . ErrNoRows ) {
return extractionSnapshot {}, nil
}
if err != nil {
return extractionSnapshot {}, fmt . Errorf ( "read extraction conversation: %w" , err )
}
snapshot . exists = true
consents := map [ string ] bool {}
if err = json . Unmarshal ( consentsJSON , & consents ); err != nil {
return extractionSnapshot {}, fmt . Errorf ( "decode extraction consents: %w" , err )
}
snapshot . storyExtractionAllowed = consents [ "story_extraction" ]
rows , err := q . Query ( ctx , `
2026-08-29 21:56:51 +08:00
SELECT id::text,public_id::text,speaker,final_text,interrupted,sequence_no
2026-10-08 10:06:44 +08:00
FROM conversation_turn ct WHERE conversation_id=$1::uuid AND final_text<>'[deleted]'
AND NOT EXISTS(SELECT 1 FROM memory_source_suppression ms CROSS JOIN LATERAL jsonb_array_elements(ms.source_refs) r
JOIN story_draft sd ON sd.public_id::text=r->>'story_id' JOIN story_revision sr ON sr.story_id=sd.id
JOIN story_source ss ON ss.story_revision_id=sr.id WHERE ss.conversation_turn_id=ct.id)
ORDER BY sequence_no,id` , conversationID )
2026-08-26 06:41:07 +08:00
if err != nil {
return extractionSnapshot {}, fmt . Errorf ( "read extraction turns: %w" , err )
}
defer rows . Close ()
for rows . Next () {
var item turn
2026-08-29 21:56:51 +08:00
var sequence int64
if err = rows . Scan ( & item . id , & item . publicID , & item . speaker , & item . text , & item . interrupted , & sequence ); err != nil {
2026-08-26 06:41:07 +08:00
return extractionSnapshot {}, err
}
2026-08-29 21:56:51 +08:00
if sequence > snapshot . maxSequence {
snapshot . maxSequence = sequence
}
2026-08-26 06:41:07 +08:00
snapshot . turns = append ( snapshot . turns , item )
}
if err = rows . Err (); err != nil {
return extractionSnapshot {}, err
}
snapshot . digest = snapshotDigest ( snapshot )
return snapshot , nil
}
2026-08-29 21:56:51 +08:00
func requestedGenerationVersion ( payload json . RawMessage ) int64 {
var value struct {
GenerationVersion int64 `json:"generation_version"`
}
if json . Unmarshal ( payload , & value ) == nil && value . GenerationVersion > 0 {
return value . GenerationVersion
}
return 1
}
2026-08-26 06:41:07 +08:00
func snapshotDigest ( snapshot extractionSnapshot ) string {
type digestTurn struct {
2026-10-08 10:06:44 +08:00
ID string `json:"id"`
PublicID string `json:"public_id"`
Speaker string `json:"speaker"`
Text string `json:"text"`
Interrupted bool `json:"interrupted"`
SourceRevision int64 `json:"source_revision"`
2026-08-26 06:41:07 +08:00
}
digestTurns := make ([] digestTurn , 0 , len ( snapshot . turns ))
for _ , item := range snapshot . turns {
2026-10-08 10:06:44 +08:00
digestTurns = append ( digestTurns , digestTurn { item . id , item . publicID , item . speaker , item . text , item . interrupted , item . sourceRevision })
2026-08-26 06:41:07 +08:00
}
digestInput := struct {
Exists bool
TenantID , SubjectID , Status string
Allowed bool
Turns [] digestTurn
}{ snapshot . exists , snapshot . tenantID , snapshot . subjectID , snapshot . status , snapshot . storyExtractionAllowed , digestTurns }
encoded , _ := json . Marshal ( digestInput )
digest := sha256 . Sum256 ( encoded )
return hex . EncodeToString ( digest [:])
}
func subjectTurns ( turns [] turn ) [] turn {
items := make ([] turn , 0 , len ( turns ))
for _ , item := range turns {
if item . speaker == "subject" {
items = append ( items , item )
}
}
return items
}
2026-09-06 20:16:01 +08:00
func joinedSubjectContent ( turns [] turn ) string {
2026-08-26 06:41:07 +08:00
parts := make ([] string , 0 , len ( turns ))
for _ , item := range turns {
2026-09-06 20:16:01 +08:00
if item . speaker != "subject" {
continue
2026-08-30 15:20:08 +08:00
}
2026-09-06 20:16:01 +08:00
parts = append ( parts , "长者：" + item . text )
2026-08-26 06:41:07 +08:00
}
2026-08-30 15:20:08 +08:00
return strings . Join ( parts , "\n" )
2026-08-26 06:41:07 +08:00
}
2026-10-08 10:06:44 +08:00
func storeGeneration ( ctx context . Context , tx pgx . Tx , conversationID string , snapshot extractionSnapshot , generated GenerationResult , generationVersion int64 , previous [] PreviousStory ) error {
2026-08-26 06:41:07 +08:00
sourceTurns := subjectTurns ( snapshot . turns )
sourcePublicIDs := make ([] string , 0 , len ( sourceTurns ))
for _ , item := range sourceTurns {
sourcePublicIDs = append ( sourcePublicIDs , item . publicID )
}
sourceJSON , _ := json . Marshal ( sourcePublicIDs )
2026-08-30 15:20:08 +08:00
if _ , err := tx . Exec ( ctx , `
INSERT INTO conversation_content_assessment (tenant_id,subject_id,conversation_id,generation_version,meaningful,story_worthy,decision_reason,model_ref,prompt_version,confidence)
VALUES ($1::uuid,$2::uuid,$3::uuid,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (conversation_id,generation_version) DO NOTHING` , snapshot . tenantID , snapshot . subjectID , conversationID , generationVersion , generated . Meaningful , generated . StoryWorthy , generated . DecisionReason , generated . ModelRef , generated . PromptVersion , generated . Confidence ); err != nil {
return fmt . Errorf ( "store conversation content assessment: %w" , err )
}
if ! generated . Meaningful {
return tx . Commit ( ctx )
}
2026-08-26 06:41:07 +08:00
if _ , err := tx . Exec ( ctx , `
INSERT INTO conversation_summary (tenant_id,subject_id,conversation_id,summary_type,content,source_turn_ids,model_ref,prompt_version,generation_version,confidence,status)
VALUES ($1::uuid,$2::uuid,$3::uuid,'conversation_summary',$4,$5::jsonb,$6,$7,$8,$9,'READY')
2026-08-29 21:56:51 +08:00
ON CONFLICT (conversation_id,summary_type,generation_version) DO NOTHING` , snapshot . tenantID , snapshot . subjectID , conversationID , summaryWithKeywords ( generated . Summary , generated . Keywords ), sourceJSON , generated . ModelRef , generated . PromptVersion , generationVersion , generated . Confidence ); err != nil {
2026-08-26 06:41:07 +08:00
return fmt . Errorf ( "store conversation summary: %w" , err )
}
2026-09-06 20:16:01 +08:00
if strings . TrimSpace ( generated . OpenTopic ) != "" {
if _ , err := tx . Exec ( ctx , `
INSERT INTO conversation_summary (tenant_id,subject_id,conversation_id,summary_type,content,source_turn_ids,model_ref,prompt_version,generation_version,confidence,status)
VALUES ($1::uuid,$2::uuid,$3::uuid,'open_topic',$4,$5::jsonb,$6,$7,$8,$9,'READY')
ON CONFLICT (conversation_id,summary_type,generation_version) DO NOTHING` , snapshot . tenantID , snapshot . subjectID , conversationID , generated . OpenTopic , sourceJSON , generated . ModelRef , generated . PromptVersion , generationVersion , generated . Confidence ); err != nil {
return fmt . Errorf ( "store open topic: %w" , err )
}
2026-08-26 06:41:07 +08:00
}
2026-10-08 10:06:44 +08:00
if generated . Stories != nil || generated . DailyEvents != nil {
if err := storeEventStories ( ctx , tx , conversationID , "" , "" , 0 , snapshot , generated , generationVersion , previous ); err != nil {
return err
}
return tx . Commit ( ctx )
}
2026-08-26 06:41:07 +08:00
if ! generated . AllowStory {
return tx . Commit ( ctx )
}
2026-10-08 10:06:44 +08:00
if err := storeStoryCandidate ( ctx , tx , conversationID , "" , 0 , snapshot , generated , generationVersion ); err != nil {
return err
}
return tx . Commit ( ctx )
}
// Both runtimes use the same story/revision/fact writer and review workflow.
// Direct calls retain their native utterance identifiers and source revisions.
func storeStoryCandidate ( ctx context . Context , tx pgx . Tx , conversationID , callSessionID string , epoch int64 , snapshot extractionSnapshot , generated GenerationResult , generationVersion int64 ) error {
sourceTurns := subjectTurns ( snapshot . turns )
sourceIDs := make ([] string , 0 , len ( sourceTurns ))
for _ , item := range sourceTurns {
sourceIDs = append ( sourceIDs , item . publicID )
}
sourceJSON , _ := json . Marshal ( sourceIDs )
2026-08-26 06:41:07 +08:00
var storyID string
2026-10-08 10:06:44 +08:00
var err error
if callSessionID != "" {
err = tx . QueryRow ( ctx , `INSERT INTO story_draft(tenant_id,subject_id,call_session_id,call_privacy_epoch,title,status,confidence,sensitivity,generation_version)
VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,'CANDIDATE',$6,'normal',$7)
ON CONFLICT(call_session_id,generation_version,event_index) DO UPDATE SET updated_at=story_draft.updated_at RETURNING id::text` , snapshot . tenantID , snapshot . subjectID , callSessionID , epoch , generated . StoryTitle , generated . Confidence , generationVersion ). Scan ( & storyID )
} else {
err = tx . QueryRow ( ctx , `
2026-08-26 06:41:07 +08:00
INSERT INTO story_draft (tenant_id,subject_id,conversation_id,title,status,confidence,sensitivity,generation_version)
VALUES ($1::uuid,$2::uuid,$3::uuid,$4,'CANDIDATE',$5,'normal',$6)
2026-10-08 10:06:44 +08:00
ON CONFLICT (conversation_id,generation_version,event_index) DO UPDATE SET updated_at=story_draft.updated_at
2026-08-26 06:41:07 +08:00
RETURNING id::text` , snapshot . tenantID , snapshot . subjectID , conversationID , generated . StoryTitle , generated . Confidence , generationVersion ). Scan ( & storyID )
2026-10-08 10:06:44 +08:00
}
2026-08-26 06:41:07 +08:00
if err != nil {
return fmt . Errorf ( "store story candidate: %w" , err )
}
var revisionID string
err = tx . QueryRow ( ctx , `
INSERT INTO story_revision (story_id,revision_no,content,author_type,model_ref,prompt_version)
VALUES ($1::uuid,1,$2,'model',$3,$4)
ON CONFLICT (story_id,revision_no) DO UPDATE SET content=story_revision.content
RETURNING id::text` , storyID , generated . StoryContent , generated . ModelRef , generated . PromptVersion ). Scan ( & revisionID )
if err != nil {
return fmt . Errorf ( "store story revision: %w" , err )
}
if _ , err = tx . Exec ( ctx , `UPDATE story_draft SET current_revision_id=$1::uuid WHERE id=$2::uuid AND current_revision_id IS NULL` , revisionID , storyID ); err != nil {
return err
}
for _ , item := range sourceTurns {
2026-10-08 10:06:44 +08:00
if callSessionID != "" {
_ , err = tx . Exec ( ctx , `INSERT INTO story_source(story_revision_id,call_session_id,call_utterance_id,call_source_revision,source_start_offset,source_end_offset,source_role)
VALUES($1::uuid,$2::uuid,$3,$4,0,$5,'subject') ON CONFLICT DO NOTHING` , revisionID , callSessionID , item . id , item . sourceRevision , utf8 . RuneCountInString ( item . text ))
} else {
_ , err = tx . Exec ( ctx , `
2026-08-26 06:41:07 +08:00
INSERT INTO story_source (story_revision_id,conversation_turn_id,source_start_offset,source_end_offset,source_role)
2026-10-08 10:06:44 +08:00
VALUES ($1::uuid,$2::uuid,0,$3,'subject') ON CONFLICT DO NOTHING` , revisionID , item . id , utf8 . RuneCountInString ( item . text ))
}
if err != nil {
2026-08-26 06:41:07 +08:00
return fmt . Errorf ( "store story source: %w" , err )
}
}
if _ , err = tx . Exec ( ctx , `
INSERT INTO fact_candidate (tenant_id,subject_id,story_id,fact_type,content,statement_basis,time_range,confidence,sensitivity,source_turn_ids,status,generation_version)
SELECT $1::uuid,$2::uuid,$3::uuid,'life_event',$4,'subject_statement','{}'::jsonb,$5,'normal',$6::jsonb,'CANDIDATE',$7
WHERE NOT EXISTS (SELECT 1 FROM fact_candidate WHERE story_id=$3::uuid AND generation_version=$7)` , snapshot . tenantID , snapshot . subjectID , storyID , generated . FactContent , generated . Confidence , sourceJSON , generationVersion ); err != nil {
return fmt . Errorf ( "store fact candidate: %w" , err )
}
2026-10-08 10:06:44 +08:00
return nil
2026-08-26 06:41:07 +08:00
}
2026-08-29 21:56:51 +08:00
func summaryWithKeywords ( summary string , keywords [] string ) string {
keywords = normalizedKeywords ( keywords )
if len ( keywords ) == 0 {
return summary
}
return summary + "\n关键词：" + strings . Join ( keywords , "、" )
}
2026-08-26 06:41:07 +08:00
func truncateRunes ( value string , limit int ) string {
characters := [] rune ( value )
if len ( characters ) <= limit {
return value
}
return string ( characters [: limit ]) + "…"
}
