Agent-System-Backend/internal/app/project_work_observation.go

package app

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"sync"

	"agent-system-backend/internal/runmgr"
	"agent-system-backend/internal/store"
	"agent-system-backend/pkg/extension"
	core "agent-system-core"
)

type projectWorkObserverSlot struct {
	mu       sync.RWMutex
	provider *extension.ProjectWorkObserver
}

func (s *projectWorkObserverSlot) get() *extension.ProjectWorkObserver {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	if s.provider == nil {
		return nil
	}
	copy := *s.provider
	return &copy
}

func (s *projectWorkObserverSlot) active() bool { return s.get() != nil }

func wireProjectWorkGate(mgr *runmgr.Manager, st *store.Store, caps hostcaps, slot *projectWorkObserverSlot,
	work func(context.Context, string, core.ID) (bool, error),
) {
	provider := slot.get()
	if provider == nil || work == nil {
		return
	}
	verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error {
		switch row.SourceKind {
		case "retrieval-hit":
			return verifyProjectKnowledgeObservation(ctx, caps, row)
		case "tool-result", "visible-analysis":
			return verifyProjectTransientObservation(ctx, st, row)
		default:
			return fmt.Errorf("project work source %s: unsupported run source kind %q", row.EventID, row.SourceKind)
		}
	}
	mgr.SetProjectWorkGate(
		func(ctx context.Context, runID, action string) (extension.ProjectWorkDecision, error) {
			return preflightProjectRun(ctx, st, slot, work, verify, runID, action)
		},
		provider.Respond,
	)
}

func (r compiledRegistry) RegisterProjectWorkObserver(extension.ProjectWorkObserver) error {
	return errors.New("extension: declarative plugins cannot register project work observers")
}

func (r pluginRegistry) RegisterProjectWorkObserver(p extension.ProjectWorkObserver) error {
	if r.pluginID != "rdm" || r.caps.workObserver == nil {
		return errors.New("extension: project work observer requires trusted RDM registration")
	}
	if p.Observe == nil || p.Preflight == nil || p.Respond == nil {
		return errors.New("extension: project work observer requires all callbacks")
	}
	s := r.caps.workObserver
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.provider != nil {
		return errors.New("extension: project work observer already registered")
	}
	copy := p
	s.provider = &copy
	return nil
}

func (r recordRegistry) RegisterProjectWorkObserver(p extension.ProjectWorkObserver) error {
	reg, ok := r.Registry.(extension.ProjectWorkObservationRegistry)
	if !ok {
		return errors.New("extension: project work observation registration unavailable")
	}
	return reg.RegisterProjectWorkObserver(p)
}

// deliverProjectWorkObservations replays only committed source parts, rechecking
// their owner and current project work authority before disclosing text to RDM.
// An Observe ACK is an intake receipt, not a preflight or publication decision.
// Failures leave the row pending for a later idempotent retry.
func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot,
	work func(context.Context, string, core.ID) (bool, error), limit int, verify ...func(context.Context, store.ProjectWorkObservationRow) error,
) (int, error) {
	provider := slot.get()
	if provider == nil || work == nil {
		return 0, errors.New("extension: trusted project work observer unavailable")
	}
	rows, err := st.PendingProjectWorkObservations(limit)
	if err != nil {
		return 0, err
	}
	return deliverProjectWorkRows(ctx, st, provider, work, rows, verify...)
}

func deliverProjectWorkRows(ctx context.Context, st *store.Store, provider *extension.ProjectWorkObserver,
	work func(context.Context, string, core.ID) (bool, error), rows []store.ProjectWorkObservationRow,
	verify ...func(context.Context, store.ProjectWorkObservationRow) error,
) (int, error) {
	completed := 0
	for _, row := range rows {
		if row.State != "pending" {
			continue
		}
		if row.SourceKind != "user-message" && row.SourceKind != "assistant-message" {
			if len(verify) != 1 || verify[0] == nil {
				return completed, fmt.Errorf("project work source %s: current resource status verifier unavailable", row.EventID)
			}
			if err := verify[0](ctx, row); err != nil {
				return completed, err
			}
		}
		if err := ctx.Err(); err != nil {
			return completed, err
		}
		conv, err := st.GetConversation(row.Owner, row.ConversationID)
		if err != nil || string(conv.ProjectID) != row.ProjectID {
			return completed, fmt.Errorf("project work source %s: current conversation binding unavailable", row.EventID)
		}
		allowed, err := work(ctx, row.Owner, core.ID(row.ProjectID))
		if err != nil || !allowed {
			return completed, fmt.Errorf("project work source %s: current project authorization denied", row.EventID)
		}
		account, err := st.GetAccountByOwner(row.Owner)
		authorizationEpoch := ""
		if err == nil && account.Enabled {
			authorizationEpoch = fmt.Sprintf("account:%d:project:%s", account.AuthVersion, row.ProjectID)
		} else if row.Owner == store.OwnerDefault {
			// Auth-off local mode has no enabled account model. The project-work
			// callback above is still authoritative for this process.
			authorizationEpoch = "auth-off-local:project:" + row.ProjectID
		} else {
			return completed, fmt.Errorf("project work source %s: current account authorization unavailable", row.EventID)
		}
		observation := extension.WorkObservation{
			EventID: row.EventID, Phase: "source-committed", Actor: row.Owner, Owner: row.Owner,
			ProjectID: core.ID(row.ProjectID), ConversationID: core.ID(row.ConversationID), RunID: core.ID(row.RunID),
			SourceKind: row.SourceKind, SourceID: row.SourceID, SourceVersion: row.SourceVersion,
			SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex, SpanStart: row.SpanStart, SpanEnd: row.SpanEnd,
			Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, DeliveryReason: row.Reason, Metadata: row.Metadata,
			AuthorizationEpoch: authorizationEpoch, ObservedAt: row.ObservedAt,
		}
		result, err := provider.Observe(ctx, observation)
		if err != nil {
			return completed, err
		}
		switch result.State {
		case extension.WorkObservationPending:
			continue
		case extension.WorkObservationAccepted, extension.WorkObservationPartial, extension.WorkObservationDenied:
			if err := st.FinishProjectWorkObservation(row.EventID, result.ReceiptID, string(result.State)); err != nil {
				return completed, err
			}
			completed++
		default:
			return completed, fmt.Errorf("project work source %s: invalid observer result %q", row.EventID, result.State)
		}
	}
	return completed, nil
}

func verifyProjectKnowledgeObservation(_ context.Context, caps hostcaps, row store.ProjectWorkObservationRow) error {
	if row.SourceKind != "retrieval-hit" || caps.st == nil {
		return errors.New("project work source: unsupported resource verification")
	}
	doc, err := caps.st.ProjectWorkKnowledgeSource(row.SourceID)
	if err != nil {
		return errors.New("project work source: current knowledge source unavailable")
	}
	hash := sha256.Sum256([]byte(doc.Chunk))
	actualHash := fmt.Sprintf("%x", hash)
	if actualHash != row.SourceVersion || actualHash != row.SourceSHA256 || doc.Chunk != row.Text ||
		(row.Metadata["ref"] != "" && row.Metadata["ref"] != doc.Ref) ||
		(row.Metadata["source"] != "" && row.Metadata["source"] != doc.Source) ||
		(doc.ProjectID != "" && string(doc.ProjectID) != row.ProjectID) {
		return errors.New("project work source: knowledge version or scope changed")
	}
	if caps.authz != nil && caps.authz.hasV2() {
		allowed, err := caps.authz.AuthorizeResource(extension.ResourceAuthzQuery{
			Actor: row.Owner, Action: "model-context", ResourceType: "knowledge", ResourceID: doc.ID,
			Owner: string(doc.Owner), ProjectID: doc.ProjectID, Source: doc.Source,
		})
		if err != nil || !allowed {
			return errors.New("project work source: current knowledge authorization denied")
		}
	} else if string(doc.Owner) != row.Owner {
		return errors.New("project work source: shared knowledge authorization unavailable")
	}
	return nil
}

// A transient tool result has no external source authority to recheck. Its
// committed private outbox bytes and stored run are the only reliable source;
// RDM must keep it a candidate until a separately verifiable evidence source
// or human confirmation supports a formal project claim.
func verifyProjectTransientObservation(_ context.Context, st *store.Store, row store.ProjectWorkObservationRow) error {
	toolSource := row.SourceKind == "tool-result" && row.Metadata["sourceStatus"] == "transient" && row.Metadata["toolName"] != ""
	visibleAnalysis := row.SourceKind == "visible-analysis" && row.Metadata["stage"] == "assistant-final" && row.Metadata["partType"] != ""
	if (!toolSource && !visibleAnalysis) || row.RunID == "" {
		return errors.New("project work source: transient result lacks trusted provenance")
	}
	run, err := st.GetRunByID(core.ID(row.RunID))
	if err != nil || string(run.Owner) != row.Owner || string(run.ProjectID) != row.ProjectID || string(run.ConversationID) != row.ConversationID {
		return errors.New("project work source: transient run binding unavailable")
	}
	return nil
}

// preflightProjectMessage delivers this committed message's pending parts
// inline, then asks RDM to prove that every exact source part was analyzed for
// the pending action. A durable intake ACK alone never permits the action.
func preflightProjectMessage(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot,
	work func(context.Context, string, core.ID) (bool, error), messageID, action string,
) (extension.ProjectWorkDecision, error) {
	pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "project source awaiting analysis"}
	provider := slot.get()
	if provider == nil || work == nil {
		return pending, errors.New("extension: trusted project work observer unavailable")
	}
	rows, err := st.ProjectMessageObservations(messageID)
	if err != nil {
		return pending, err
	}
	if len(rows) == 0 {
		return pending, errors.New("extension: committed project message has no source manifest")
	}
	if _, err := deliverProjectWorkRows(ctx, st, provider, work, rows); err != nil {
		return pending, err
	}
	rows, err = st.ProjectMessageObservations(messageID)
	if err != nil {
		return pending, err
	}
	manifest := make([]extension.ProjectWorkSource, 0, len(rows))
	for _, row := range rows {
		if row.State != "accepted" || row.Delivery != "complete" {
			return pending, nil
		}
		manifest = append(manifest, extension.ProjectWorkSource{
			EventID: row.EventID, SourceKind: row.SourceKind, SourceID: row.SourceID,
			SourceVersion: row.SourceVersion, SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex,
		})
	}
	first := rows[0]
	for _, row := range rows[1:] {
		if row.Owner != first.Owner || row.ProjectID != first.ProjectID || row.ConversationID != first.ConversationID {
			return pending, errors.New("extension: mixed project source manifest")
		}
	}
	conv, err := st.GetConversation(first.Owner, first.ConversationID)
	if err != nil || string(conv.ProjectID) != first.ProjectID {
		return pending, errors.New("extension: project binding changed before preflight")
	}
	allowed, err := work(ctx, first.Owner, conv.ProjectID)
	if err != nil || !allowed {
		return pending, errors.New("extension: current project work denied before preflight")
	}
	raw, err := json.Marshal(manifest)
	if err != nil {
		return pending, err
	}
	watermark := fmt.Sprintf("%x", sha256.Sum256(raw))
	decision, err := provider.Preflight(ctx, extension.ProjectWorkAction{
		Actor: first.Owner, Owner: first.Owner, ProjectID: conv.ProjectID,
		ConversationID: conv.ID, RunID: core.ID(first.RunID), Action: action,
		Manifest: manifest, SourceWatermark: watermark,
	})
	if err != nil {
		return pending, err
	}
	switch decision.State {
	case extension.ProjectWorkProceed:
		if decision.DecisionID == "" || decision.SourceWatermark != watermark {
			return pending, errors.New("extension: project preflight decision did not bind current sources")
		}
	case extension.ProjectWorkClarify, extension.ProjectWorkPending, extension.ProjectWorkDenied:
		// A non-proceed decision is a real result, never an implied approval.
	default:
		return pending, errors.New("extension: invalid project preflight decision")
	}
	return decision, nil
}

// preflightProjectRun binds all committed user, retrieval, tool, and visible
// analysis parts of one run before a later model or business action consumes
// them. Missing/partial parts keep the action pending. The domain provider
// decides whether the complete source set contains a blocking risk.
func preflightProjectRun(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot,
	work func(context.Context, string, core.ID) (bool, error), verify func(context.Context, store.ProjectWorkObservationRow) error,
	runID, action string,
) (extension.ProjectWorkDecision, error) {
	pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "run sources awaiting analysis"}
	provider := slot.get()
	if provider == nil || work == nil || verify == nil {
		return pending, errors.New("extension: trusted project run observer unavailable")
	}
	run, err := st.GetRunByID(core.ID(runID))
	if err != nil || run.ProjectID == "" {
		return pending, errors.New("extension: trusted project run unavailable")
	}
	rows, err := st.ProjectRunObservations(runID)
	if err != nil {
		return pending, err
	}
	if len(rows) == 0 {
		return pending, errors.New("extension: project run has no trusted source manifest")
	}
	if _, err := deliverProjectWorkRows(ctx, st, provider, work, rows, verify); err != nil {
		return pending, err
	}
	rows, err = st.ProjectRunObservations(runID)
	if err != nil {
		return pending, err
	}
	manifest := make([]extension.ProjectWorkSource, 0, len(rows))
	for _, row := range rows {
		if row.State != "accepted" || row.Delivery != "complete" {
			return pending, nil
		}
		if row.RunID != runID || row.Owner != string(run.Owner) || row.ProjectID != string(run.ProjectID) || row.ConversationID != string(run.ConversationID) {
			return pending, errors.New("extension: mixed run source manifest")
		}
		if row.SourceKind != "user-message" && row.SourceKind != "assistant-message" {
			if err := verify(ctx, row); err != nil {
				return pending, err
			}
		}
		manifest = append(manifest, extension.ProjectWorkSource{
			EventID: row.EventID, SourceKind: row.SourceKind, SourceID: row.SourceID,
			SourceVersion: row.SourceVersion, SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex,
		})
	}
	conv, err := st.GetConversation(string(run.Owner), string(run.ConversationID))
	if err != nil || conv.ProjectID != run.ProjectID {
		return pending, errors.New("extension: run project binding changed before preflight")
	}
	allowed, err := work(ctx, string(run.Owner), run.ProjectID)
	if err != nil || !allowed {
		return pending, errors.New("extension: current run project work denied")
	}
	raw, err := json.Marshal(manifest)
	if err != nil {
		return pending, err
	}
	watermark := fmt.Sprintf("%x", sha256.Sum256(raw))
	decision, err := provider.Preflight(ctx, extension.ProjectWorkAction{
		Actor: string(run.Owner), Owner: string(run.Owner), ProjectID: run.ProjectID,
		ConversationID: run.ConversationID, RunID: run.ID, Action: action,
		Manifest: manifest, SourceWatermark: watermark,
	})
	if err != nil {
		return pending, err
	}
	switch decision.State {
	case extension.ProjectWorkProceed:
		if decision.DecisionID == "" || decision.SourceWatermark != watermark {
			return pending, errors.New("extension: run preflight decision did not bind current sources")
		}
	case extension.ProjectWorkClarify, extension.ProjectWorkPending, extension.ProjectWorkDenied:
	default:
		return pending, errors.New("extension: invalid project run preflight decision")
	}
	return decision, nil
}