Agent-System-Backend/internal/store/project_work_outbox.go

package store

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	core "agent-system-core"
)

// ProjectWorkObservationRow is a private host delivery record. A project ID is
// a trusted locator here, not authorization to reveal the original message to
// other project members. The source body is fetched from the message row only
// for the current provider delivery and is never part of project audit DTOs.
type ProjectWorkObservationRow struct {
	EventID        string
	Owner          string
	ProjectID      string
	ConversationID string
	RunID          string
	SourceKind     string
	SourceID       string
	SourceVersion  string
	SourceSHA256   string
	PartIndex      int
	SpanStart      int
	SpanEnd        int
	Delivery       string
	Scope          string
	ObservedAt     time.Time
	State          string
	ReceiptID      string
	Text           string
	Metadata       map[string]string
	Reason         string
}

// RunWorkObservationInput identifies an actual retrieval/tool/visible-analysis
// result seen by a trusted run. Owner, project, and conversation are resolved
// from the stored run; callers cannot supply or override those identities.
type RunWorkObservationInput struct {
	RunID         string
	SourceKind    string // retrieval-hit|tool-result|visible-analysis
	SourceID      string
	SourceVersion string
	Text          string
	Delivery      string // complete|summary-only|partial
	PartIndex     int
	Metadata      map[string]string
}

func (s *Store) AppendRunWorkObservation(input RunWorkObservationInput) (string, error) {
	if input.RunID == "" || input.SourceID == "" || input.SourceVersion == "" || input.PartIndex < 0 {
		return "", errors.New("project work source: missing trusted source identity")
	}
	switch input.SourceKind {
	case "retrieval-hit", "tool-result", "visible-analysis":
	default:
		return "", errors.New("project work source: unsupported kind")
	}
	if input.Delivery != "complete" && input.Delivery != "summary-only" && input.Delivery != "partial" {
		return "", errors.New("project work source: unsupported delivery state")
	}
	if len(input.Text) > 64*1024 {
		return "", errors.New("project work source: fragment exceeds bounded intake")
	}
	meta, err := json.Marshal(input.Metadata)
	if err != nil || len(meta) > 8*1024 {
		return "", errors.New("project work source: invalid or oversized metadata")
	}
	tx, err := s.db.Begin()
	if err != nil {
		return "", err
	}
	defer tx.Rollback()
	var owner, projectID, conversationID, convOwner, convProject string
	if err := tx.QueryRow(s.q(`SELECT r.owner,r.project_id,r.conversation_id,c.owner,c.project_id FROM runs r JOIN conversations c ON c.id=r.conversation_id WHERE r.id=?`), input.RunID).
		Scan(&owner, &projectID, &conversationID, &convOwner, &convProject); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return "", ErrNotFound
		}
		return "", err
	}
	if owner != convOwner || projectID == "" || projectID != convProject {
		return "", errors.New("project work source: run binding does not match current conversation")
	}
	hash := sha256.Sum256([]byte(input.Text))
	eventID := newID("obs")
	observed := time.Now().UTC().Format(time.RFC3339Nano)
	if _, err := tx.Exec(s.q(`INSERT INTO project_work_outbox(event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,payload_text,metadata_json)
		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`), eventID, owner, projectID, conversationID, input.RunID,
		input.SourceKind, input.SourceID, input.SourceVersion, hex.EncodeToString(hash[:]), input.PartIndex,
		0, len(input.Text), input.Delivery, "private", observed, input.Text, string(meta)); err != nil {
		return "", err
	}
	if err := tx.Commit(); err != nil {
		return "", err
	}
	return eventID, nil
}

func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message) error {
	if m == nil || (m.Role != core.RoleUser && m.Role != core.RoleAssistant) {
		return nil
	}
	var owner, projectID string
	if err := tx.QueryRow(s.q(`SELECT owner,project_id FROM conversations WHERE id=?`), m.ConversationID).Scan(&owner, &projectID); err != nil {
		return err
	}
	if projectID == "" {
		return nil
	}
	kind := string(m.Role) + "-message"
	observed := m.CreatedAt.UTC().Format(time.RFC3339Nano)
	for i, part := range m.Parts {
		// A tool summary, artifact reference, card, or hidden thinking part is not
		// evidence of the original resource body. The separate retrieval/tool
		// adapters will supply their actual source IDs and spans later.
		if part.Type == core.PartThinking {
			continue
		}
		delivery := "partial"
		var material []byte
		switch part.Type {
		case core.PartText:
			material = []byte(part.Text)
		case core.PartPlan, core.PartCheck, core.PartCard:
			material = []byte(marshalJSON(part)) // saved, visible analysis artefact
		default:
			material = []byte(marshalJSON(part))
		}
		if (part.Type == core.PartText || part.Type == core.PartPlan || part.Type == core.PartCheck || part.Type == core.PartCard) && len(material) > 0 && len(material) <= 64*1024 {
			delivery = "complete"
		}
		hash := sha256.Sum256(material)
		idDigest := sha256.Sum256([]byte(fmt.Sprintf("%s:%s:%d", kind, m.ID, i)))
		eventID := "obs_" + hex.EncodeToString(idDigest[:16])
		meta, _ := json.Marshal(map[string]string{"partType": string(part.Type)})
		if _, err := tx.Exec(s.q(`INSERT INTO project_work_outbox(event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,metadata_json)
			VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`), eventID, owner, projectID, m.ConversationID, m.RunID,
			kind, m.ID, m.ID, hex.EncodeToString(hash[:]), i, 0, len(material), delivery, "private", observed, string(meta)); err != nil {
			return err
		}
	}
	return nil
}

// PendingProjectWorkObservations returns committed, not-yet-ACKed source parts.
// It rechecks that the immutable source bytes still match the recorded hash.
// Missing or changed source bytes become partial, never invented complete text.
func (s *Store) PendingProjectWorkObservations(limit int) ([]ProjectWorkObservationRow, error) {
	return s.projectWorkObservations("", "", limit)
}

// ProjectMessageObservations returns every committed part of one message,
// including already-ACKed parts, for an action's exact source manifest.
func (s *Store) ProjectMessageObservations(messageID string) ([]ProjectWorkObservationRow, error) {
	if messageID == "" {
		return nil, errors.New("project work observation: empty message ID")
	}
	var count int
	if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM project_work_outbox WHERE source_id=? AND source_kind IN ('user-message','assistant-message')`), messageID).Scan(&count); err != nil {
		return nil, err
	}
	if count > 200 {
		return nil, errors.New("project work observation: message parts exceed safe preflight limit")
	}
	return s.projectWorkObservations(messageID, "", 200)
}

// ProjectRunObservations is the exact committed source set linked to a run.
// An over-limit set is an explicit incomplete manifest, never a truncated pass.
func (s *Store) ProjectRunObservations(runID string) ([]ProjectWorkObservationRow, error) {
	if runID == "" {
		return nil, errors.New("project work observation: empty run ID")
	}
	var count int
	if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM project_work_outbox WHERE run_id=?`), runID).Scan(&count); err != nil {
		return nil, err
	}
	if count > 200 {
		return nil, errors.New("project work observation: run sources exceed safe preflight limit")
	}
	return s.projectWorkObservations("", runID, 200)
}

func (s *Store) projectWorkObservations(messageID, runID string, limit int) ([]ProjectWorkObservationRow, error) {
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	query := `SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id,payload_text,metadata_json
		FROM project_work_outbox WHERE state='pending' ORDER BY observed_at,event_id LIMIT ?`
	args := []any{limit}
	if messageID != "" {
		query = `SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id,payload_text,metadata_json
			FROM project_work_outbox WHERE source_id=? AND source_kind IN ('user-message','assistant-message') ORDER BY part_index LIMIT ?`
		args = []any{messageID, limit}
	} else if runID != "" {
		query = `SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id,payload_text,metadata_json
			FROM project_work_outbox WHERE run_id=? ORDER BY observed_at,event_id LIMIT ?`
		args = []any{runID, limit}
	}
	rows, err := s.db.Query(s.q(query), args...)
	if err != nil {
		return nil, err
	}
	out := make([]ProjectWorkObservationRow, 0, limit)
	for rows.Next() {
		var row ProjectWorkObservationRow
		var observed, payload, metadata string
		if err := rows.Scan(&row.EventID, &row.Owner, &row.ProjectID, &row.ConversationID, &row.RunID,
			&row.SourceKind, &row.SourceID, &row.SourceVersion, &row.SourceSHA256, &row.PartIndex,
			&row.SpanStart, &row.SpanEnd, &row.Delivery, &row.Scope, &observed, &row.State, &row.ReceiptID, &payload, &metadata); err != nil {
			rows.Close()
			return nil, err
		}
		if err := json.Unmarshal([]byte(metadata), &row.Metadata); err != nil {
			rows.Close()
			return nil, err
		}
		row.Text = payload
		row.ObservedAt, err = time.Parse(time.RFC3339Nano, observed)
		if err != nil {
			rows.Close()
			return nil, err
		}
		out = append(out, row)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return nil, err
	}
	for i := range out {
		row := &out[i]
		if row.Delivery != "complete" {
			continue
		}
		if row.SourceKind != "user-message" && row.SourceKind != "assistant-message" {
			hash := sha256.Sum256([]byte(row.Text))
			if hex.EncodeToString(hash[:]) != row.SourceSHA256 || row.SpanEnd != len(row.Text) {
				row.Delivery, row.Reason, row.Text = "partial", "source-changed", ""
			}
			continue
		}
		msg, err := s.MessageByID(row.SourceID)
		if errors.Is(err, ErrNotFound) {
			row.Delivery, row.Reason = "partial", "source-missing"
			continue
		}
		if err != nil {
			return nil, err
		}
		if msg.ConversationID != core.ID(row.ConversationID) || row.PartIndex < 0 || row.PartIndex >= len(msg.Parts) {
			row.Delivery, row.Reason = "partial", "source-changed"
			continue
		}
		part := msg.Parts[row.PartIndex]
		var text string
		switch part.Type {
		case core.PartText:
			text = part.Text
		case core.PartPlan, core.PartCheck, core.PartCard:
			text = marshalJSON(part)
		default:
			row.Delivery, row.Reason = "partial", "source-changed"
			continue
		}
		hash := sha256.Sum256([]byte(text))
		if hex.EncodeToString(hash[:]) != row.SourceSHA256 || row.SpanEnd != len(text) {
			row.Delivery, row.Reason = "partial", "source-changed"
			continue
		}
		row.Text = text
	}
	return out, nil
}

// AckProjectWorkObservation records only the provider's durable intake receipt.
// It does not assert that source analysis, risk checks, or project publication
// finished. A different receipt for the same event is a conflict, not a retry.
func (s *Store) AckProjectWorkObservation(eventID, receiptID string) error {
	return s.FinishProjectWorkObservation(eventID, receiptID, "accepted")
}

// FinishProjectWorkObservation records the provider's terminal intake result.
// "accepted" is only durable receipt; "partial" and "denied" are truthful
// terminal treatment of this source part, not a project fact publication.
func (s *Store) FinishProjectWorkObservation(eventID, receiptID, state string) error {
	if eventID == "" || receiptID == "" {
		return errors.New("project work observation: empty event or receipt")
	}
	if state != "accepted" && state != "partial" && state != "denied" {
		return errors.New("project work observation: invalid terminal state")
	}
	res, err := s.db.Exec(s.q(`UPDATE project_work_outbox SET state=?,receipt_id=? WHERE event_id=? AND state='pending'`), state, receiptID, eventID)
	if err != nil {
		return err
	}
	changed, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if changed == 1 {
		return nil
	}
	var currentReceipt, currentState string
	if err := s.db.QueryRow(s.q(`SELECT receipt_id,state FROM project_work_outbox WHERE event_id=?`), eventID).Scan(&currentReceipt, &currentState); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		return err
	}
	if currentReceipt != receiptID || currentState != state {
		return fmt.Errorf("project work observation: receipt conflict for %s", eventID)
	}
	return nil
}