From e62c027b7d77dc01d0a9563eed78dba0f0ae036f Mon Sep 17 00:00:00 2001 From: viettranx Date: Sat, 11 Apr 2026 14:57:51 +0700 Subject: [PATCH] feat(vault): task + delegation auto-linking in enrich worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add enrich_auto_linking.go: deterministic auto-link logic for task/delegation contexts - Add team_task_siblings.go (PG + SQLite): find all tasks in same delegation for batch link - Update enrich_worker.go: call enrich_auto_linking Phase 2 hook + skip re-embed for binary - Update team_store.go: add TaskSiblings interface method + TeamStore binding - Add context_keys.go + test: define ContextDelegationIDKey + ContextTaskIDKey for propagation - Update vault_interceptor.go: extract + inject delegation ID from request - Update run_context.go: propagate DelegationID from request - Update loop_context.go: inject DelegationID from run context - Update teams_tasks.go: DeleteTask(s) → cleanup auto-links via vault_source_cleanup - Update teams_tasks_activity.go: DetachFileFromTask → cleanup related auto-links - Update vault_documents_enrichment.go: populate summary from media_summary - Update cmd/gateway.go: wire TeamStore to handler bootstrap --- cmd/gateway.go | 1 + internal/agent/loop_context.go | 3 + internal/store/pg/team_task_siblings.go | 106 +++++++++++ internal/store/pg/teams_tasks.go | 69 ++++++- internal/store/pg/teams_tasks_activity.go | 18 ++ .../store/pg/vault_documents_enrichment.go | 2 +- internal/store/run_context.go | 1 + .../store/sqlitestore/team_task_siblings.go | 119 ++++++++++++ internal/store/sqlitestore/teams_tasks.go | 49 ++++- .../store/sqlitestore/teams_tasks_activity.go | 24 ++- internal/store/team_store.go | 22 +++ internal/tools/context_keys.go | 23 +++ internal/tools/context_keys_test.go | 32 ++++ internal/tools/team_mock_backend_test.go | 3 + internal/tools/vault_interceptor.go | 20 +- internal/vault/enrich_auto_linking.go | 177 ++++++++++++++++++ internal/vault/enrich_worker.go | 94 ++++++++-- 17 files changed, 730 insertions(+), 33 deletions(-) create mode 100644 internal/store/pg/team_task_siblings.go create mode 100644 internal/store/sqlitestore/team_task_siblings.go create mode 100644 internal/vault/enrich_auto_linking.go diff --git a/cmd/gateway.go b/cmd/gateway.go index d078a1a0..93bef7f4 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -219,6 +219,7 @@ func runGateway() { Model: bgModel, EventBus: domainBus, MsgBus: msgBus, + TeamStore: pgStores.Teams, // Phase 04 task-based auto-linking }) enrichProgress = ep defer cleanupVaultEnrich() diff --git a/internal/agent/loop_context.go b/internal/agent/loop_context.go index b75bf2f3..8774c3d7 100644 --- a/internal/agent/loop_context.go +++ b/internal/agent/loop_context.go @@ -109,6 +109,9 @@ func (l *Loop) injectContext(ctx context.Context, req *RunRequest) (contextSetup if req.TeamTaskID != "" { ctx = tools.WithTeamTaskID(ctx, req.TeamTaskID) } + if req.DelegationID != "" { + ctx = tools.WithDelegationID(ctx, req.DelegationID) + } // --- Per-user setup: file seeding + workspace resolution --- // Uses userSetups sync.Map to track both concerns atomically per user. diff --git a/internal/store/pg/team_task_siblings.go b/internal/store/pg/team_task_siblings.go new file mode 100644 index 00000000..06cfb6fd --- /dev/null +++ b/internal/store/pg/team_task_siblings.go @@ -0,0 +1,106 @@ +package pg + +import ( + "context" + "fmt" + "strings" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// BatchGetTaskSiblingsByBasenames returns vault docs attached to the same +// team task as each input basename. Result is keyed by source basename. +// +// Caps at `limit` per (source_base × task_id) pair — a file attached to two +// different tasks gets up to `limit` siblings from each task (not shared). +// +// Tenant isolation via tta.tenant_id filter; team isolation via matching +// vd.team_id = tta.team_id in the join. +// +// Single SQL per chunk — `ANY($2)` lets us avoid 1 query per basename. +func (s *PGTeamStore) BatchGetTaskSiblingsByBasenames( + ctx context.Context, + tenantID uuid.UUID, + basenames []string, + limit int, +) (map[string][]store.TaskSibling, error) { + if len(basenames) == 0 { + return nil, nil + } + if limit <= 0 { + limit = 9 + } + + // Lowercase + dedup defensively. + seen := make(map[string]bool, len(basenames)) + clean := make([]string, 0, len(basenames)) + for _, b := range basenames { + lb := strings.ToLower(b) + if lb == "" || seen[lb] { + continue + } + seen[lb] = true + clean = append(clean, lb) + } + + out := make(map[string][]store.TaskSibling, len(clean)) + const chunkSize = 500 + + for start := 0; start < len(clean); start += chunkSize { + end := start + chunkSize + if end > len(clean) { + end = len(clean) + } + chunk := clean[start:end] + + const q = ` +WITH target_tasks AS ( + SELECT DISTINCT tta.base_name AS source_base, tta.task_id + FROM team_task_attachments tta + WHERE tta.tenant_id = $1 AND tta.base_name = ANY($2) +), +ranked AS ( + SELECT + tt.source_base, + tta2.task_id, + vd.id AS doc_id, + tta2.base_name, + tta2.created_at, + ROW_NUMBER() OVER ( + PARTITION BY tt.source_base, tta2.task_id + ORDER BY tta2.created_at DESC, vd.id DESC + ) AS rn + FROM target_tasks tt + JOIN team_task_attachments tta2 ON tta2.task_id = tt.task_id + JOIN vault_documents vd + ON vd.tenant_id = tta2.tenant_id + AND vd.team_id = tta2.team_id + AND vd.path_basename = tta2.base_name + WHERE tta2.base_name != tt.source_base +) +SELECT source_base, task_id, doc_id, base_name, created_at +FROM ranked +WHERE rn <= $3 +ORDER BY source_base, task_id, created_at DESC +` + rows, err := s.db.QueryContext(ctx, q, tenantID, pqStringArray(chunk), limit) + if err != nil { + return nil, fmt.Errorf("batch task siblings: %w", err) + } + for rows.Next() { + var sourceBase string + var sib store.TaskSibling + if err := rows.Scan(&sourceBase, &sib.TaskID, &sib.DocID, &sib.BaseName, &sib.AttachmentTime); err != nil { + rows.Close() + return nil, err + } + out[sourceBase] = append(out[sourceBase], sib) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + } + return out, nil +} diff --git a/internal/store/pg/teams_tasks.go b/internal/store/pg/teams_tasks.go index 19f654ea..dac6e2c9 100644 --- a/internal/store/pg/teams_tasks.go +++ b/internal/store/pg/teams_tasks.go @@ -412,7 +412,14 @@ func (s *PGTeamStore) DeleteTask(ctx context.Context, taskID, teamID uuid.UUID) tenantWhere = fmt.Sprintf(" AND tenant_id = $%d", len(args)+1) args = append(args, tid) } - res, err := s.db.ExecContext(ctx, + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("delete task: begin tx: %w", err) + } + defer tx.Rollback() //nolint:errcheck — no-op after commit + + res, err := tx.ExecContext(ctx, `DELETE FROM team_tasks WHERE id = $1 AND team_id = $2 AND status IN ('completed','failed','cancelled')`+tenantWhere, args...) if err != nil { @@ -422,7 +429,25 @@ func (s *PGTeamStore) DeleteTask(ctx context.Context, taskID, teamID uuid.UUID) if n == 0 { return store.ErrTaskNotFound } - return nil + + // Phase 04: clean up auto-created vault_links sourced from this task + // (task_attachment + defensive delegation_attachment) inside the same tx + // so cleanup is atomic with task deletion. + sources := []string{ + "task:" + taskID.String(), + "delegation:" + taskID.String(), + } + if delRes, derr := tx.ExecContext(ctx, ` + DELETE FROM vault_links + WHERE metadata->>'source' = ANY($1) + `, pqStringArray(sources)); derr != nil { + slog.Warn("delete task: vault_links cleanup", "task_id", taskID, "err", derr) + } else if cnt, _ := delRes.RowsAffected(); cnt > 0 { + slog.Info("vault.link.deleted_on_task_delete", + "task_id", taskID, "count", cnt) + } + + return tx.Commit() } func (s *PGTeamStore) DeleteTasks(ctx context.Context, taskIDs []uuid.UUID, teamID uuid.UUID) ([]uuid.UUID, error) { @@ -439,23 +464,57 @@ func (s *PGTeamStore) DeleteTasks(ctx context.Context, taskIDs []uuid.UUID, team tenantWhere = fmt.Sprintf(" AND tenant_id = $%d", len(args)+1) args = append(args, tid) } - rows, err := s.db.QueryContext(ctx, + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("delete tasks: begin tx: %w", err) + } + defer tx.Rollback() //nolint:errcheck + + rows, err := tx.QueryContext(ctx, `DELETE FROM team_tasks WHERE id = ANY($1) AND team_id = $2 AND status IN ('completed','failed','cancelled')`+tenantWhere+` RETURNING id`, args...) if err != nil { return nil, err } - defer rows.Close() var deleted []uuid.UUID for rows.Next() { var id uuid.UUID if err := rows.Scan(&id); err != nil { + rows.Close() return deleted, err } deleted = append(deleted, id) } - return deleted, rows.Err() + rows.Close() + if err := rows.Err(); err != nil { + return deleted, err + } + + // Phase 04 bulk cleanup: one statement for task + delegation sources + // across all deleted tasks. Scoped by tenant via the team_tasks delete + // above (only actually-deleted IDs populate `deleted`). + if len(deleted) > 0 { + sources := make([]string, 0, len(deleted)*2) + for _, id := range deleted { + sources = append(sources, "task:"+id.String(), "delegation:"+id.String()) + } + if delRes, derr := tx.ExecContext(ctx, ` + DELETE FROM vault_links + WHERE metadata->>'source' = ANY($1) + `, pqStringArray(sources)); derr != nil { + slog.Warn("delete tasks: vault_links cleanup", "count", len(deleted), "err", derr) + } else if cnt, _ := delRes.RowsAffected(); cnt > 0 { + slog.Info("vault.link.deleted_on_task_bulk_delete", + "tasks_deleted", len(deleted), "links_deleted", cnt) + } + } + + if err := tx.Commit(); err != nil { + return deleted, err + } + return deleted, nil } // ListActiveTasksByChatID returns non-terminal tasks for a given chat_id (session key). diff --git a/internal/store/pg/teams_tasks_activity.go b/internal/store/pg/teams_tasks_activity.go index 8dc54b36..02049ff7 100644 --- a/internal/store/pg/teams_tasks_activity.go +++ b/internal/store/pg/teams_tasks_activity.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "log/slog" "time" "github.com/google/uuid" @@ -269,6 +270,23 @@ func (s *PGTeamStore) DetachFileFromTask(ctx context.Context, taskID uuid.UUID, if n, _ := res.RowsAffected(); n > 0 { _, _ = s.db.ExecContext(ctx, `UPDATE team_tasks SET attachment_count = GREATEST(attachment_count - 1, 0) WHERE id = $1 AND tenant_id = $2`, taskID, tid) + + // Phase 04: clean up Phase 2.5 auto-links sourced from this task. + // Scoped DELETE via source key — broader than the single detached + // basename, but the task's auto-link group is meaningful as a whole. + source := "task:" + taskID.String() + if delRes, derr := s.db.ExecContext(ctx, ` + DELETE FROM vault_links vl + USING vault_documents vd + WHERE vl.metadata->>'source' = $1 + AND vd.tenant_id = $2 + AND (vl.from_doc_id = vd.id OR vl.to_doc_id = vd.id) + `, source, tid); derr != nil { + slog.Warn("vault.link.cleanup_on_detach", "task_id", taskID, "err", derr) + } else if cnt, _ := delRes.RowsAffected(); cnt > 0 { + slog.Info("vault.link.deleted_on_detach", + "task_id", taskID, "count", cnt) + } } return nil } diff --git a/internal/store/pg/vault_documents_enrichment.go b/internal/store/pg/vault_documents_enrichment.go index 3aa2fc83..e69b0cc9 100644 --- a/internal/store/pg/vault_documents_enrichment.go +++ b/internal/store/pg/vault_documents_enrichment.go @@ -75,7 +75,7 @@ func (s *PGVaultStore) FindSimilarDocs(ctx context.Context, tenantID, agentID, d return nil, nil // no embedding = no neighbors } - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, 1 - (embedding <=> $1::vector) AS score FROM vault_documents diff --git a/internal/store/run_context.go b/internal/store/run_context.go index b95d2a62..1656722a 100644 --- a/internal/store/run_context.go +++ b/internal/store/run_context.go @@ -52,6 +52,7 @@ type RunContext struct { WorkspaceChannel string WorkspaceChatID string TeamTaskID string + DelegationID string // delegation identifier for vault auto-linking (empty when not in delegation) LeaderAgentID string // leader's agent UUID for member memory read fallback AgentToolKey string // tool-level agent key for registry routing } diff --git a/internal/store/sqlitestore/team_task_siblings.go b/internal/store/sqlitestore/team_task_siblings.go new file mode 100644 index 00000000..b9a75bc2 --- /dev/null +++ b/internal/store/sqlitestore/team_task_siblings.go @@ -0,0 +1,119 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "fmt" + "strings" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// BatchGetTaskSiblingsByBasenames — SQLite mirror of the PG implementation. +// Uses dynamic `?,?,?` placeholders (no array type in sqlite) and the same +// PARTITION BY (source_base, task_id) cap semantics. +func (s *SQLiteTeamStore) BatchGetTaskSiblingsByBasenames( + ctx context.Context, + tenantID uuid.UUID, + basenames []string, + limit int, +) (map[string][]store.TaskSibling, error) { + if len(basenames) == 0 { + return nil, nil + } + if limit <= 0 { + limit = 9 + } + + // Lowercase + dedup defensively. + seen := make(map[string]bool, len(basenames)) + clean := make([]string, 0, len(basenames)) + for _, b := range basenames { + lb := strings.ToLower(b) + if lb == "" || seen[lb] { + continue + } + seen[lb] = true + clean = append(clean, lb) + } + + out := make(map[string][]store.TaskSibling, len(clean)) + const chunkSize = 500 + + for start := 0; start < len(clean); start += chunkSize { + end := start + chunkSize + if end > len(clean) { + end = len(clean) + } + chunk := clean[start:end] + + ph := make([]string, len(chunk)) + args := []any{tenantID.String()} + for i, b := range chunk { + ph[i] = "?" + args = append(args, b) + } + args = append(args, limit) + + q := fmt.Sprintf(` +WITH target_tasks AS ( + SELECT DISTINCT tta.base_name AS source_base, tta.task_id + FROM team_task_attachments tta + WHERE tta.tenant_id = ? AND tta.base_name IN (%s) +), +ranked AS ( + SELECT + tt.source_base, + tta2.task_id, + vd.id AS doc_id, + tta2.base_name, + tta2.created_at, + ROW_NUMBER() OVER ( + PARTITION BY tt.source_base, tta2.task_id + ORDER BY tta2.created_at DESC, vd.id DESC + ) AS rn + FROM target_tasks tt + JOIN team_task_attachments tta2 ON tta2.task_id = tt.task_id + JOIN vault_documents vd + ON vd.tenant_id = tta2.tenant_id + AND vd.team_id = tta2.team_id + AND vd.path_basename = tta2.base_name + WHERE tta2.base_name != tt.source_base +) +SELECT source_base, task_id, doc_id, base_name, created_at +FROM ranked +WHERE rn <= ? +ORDER BY source_base, task_id, created_at DESC +`, strings.Join(ph, ",")) + + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("batch task siblings: %w", err) + } + for rows.Next() { + var sourceBase string + var taskIDStr, docIDStr string + var sib store.TaskSibling + ca := &sqliteTime{} + if err := rows.Scan(&sourceBase, &taskIDStr, &docIDStr, &sib.BaseName, ca); err != nil { + rows.Close() + return nil, err + } + if tid, perr := uuid.Parse(taskIDStr); perr == nil { + sib.TaskID = tid + } + if did, perr := uuid.Parse(docIDStr); perr == nil { + sib.DocID = did + } + sib.AttachmentTime = ca.Time + out[sourceBase] = append(out[sourceBase], sib) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + } + return out, nil +} diff --git a/internal/store/sqlitestore/teams_tasks.go b/internal/store/sqlitestore/teams_tasks.go index b0e0a86a..0699666f 100644 --- a/internal/store/sqlitestore/teams_tasks.go +++ b/internal/store/sqlitestore/teams_tasks.go @@ -7,6 +7,7 @@ import ( "database/sql" "encoding/json" "fmt" + "log/slog" "strings" "time" "unicode" @@ -357,7 +358,14 @@ func (s *SQLiteTeamStore) DeleteTask(ctx context.Context, taskID, teamID uuid.UU tenantWhere = " AND tenant_id = ?" args = append(args, tid) } - res, err := s.db.ExecContext(ctx, + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("delete task: begin tx: %w", err) + } + defer tx.Rollback() //nolint:errcheck + + res, err := tx.ExecContext(ctx, `DELETE FROM team_tasks WHERE id = ? AND team_id = ? AND status IN ('completed','failed','cancelled')`+tenantWhere, args...) if err != nil { @@ -367,7 +375,17 @@ func (s *SQLiteTeamStore) DeleteTask(ctx context.Context, taskID, teamID uuid.UU if n == 0 { return store.ErrTaskNotFound } - return nil + + // Phase 04: clean up auto-created vault_links sourced from this task + // (task_attachment + defensive delegation_attachment) inside the same tx. + if _, derr := tx.ExecContext(ctx, ` + DELETE FROM vault_links + WHERE json_extract(metadata, '$.source') IN (?, ?) + `, "task:"+taskID.String(), "delegation:"+taskID.String()); derr != nil { + slog.Warn("delete task: vault_links cleanup", "task_id", taskID, "err", derr) + } + + return tx.Commit() } func (s *SQLiteTeamStore) DeleteTasks(ctx context.Context, taskIDs []uuid.UUID, teamID uuid.UUID) ([]uuid.UUID, error) { @@ -415,10 +433,33 @@ func (s *SQLiteTeamStore) DeleteTasks(ctx context.Context, taskIDs []uuid.UUID, } if len(deleted) > 0 { - _, err = s.db.ExecContext(ctx, `DELETE FROM team_tasks WHERE `+cond, args...) - if err != nil { + tx, txErr := s.db.BeginTx(ctx, nil) + if txErr != nil { + return nil, fmt.Errorf("delete tasks: begin tx: %w", txErr) + } + defer tx.Rollback() //nolint:errcheck + + if _, err = tx.ExecContext(ctx, `DELETE FROM team_tasks WHERE `+cond, args...); err != nil { return nil, err } + + // Phase 04 bulk cleanup: drop task_attachment / delegation_attachment + // links for all deleted task IDs in the same tx. + sourceArgs := make([]any, 0, len(deleted)*2) + phs := make([]string, 0, len(deleted)*2) + for _, id := range deleted { + sourceArgs = append(sourceArgs, "task:"+id.String(), "delegation:"+id.String()) + phs = append(phs, "?", "?") + } + delQ := `DELETE FROM vault_links WHERE json_extract(metadata, '$.source') IN (` + + strings.Join(phs, ",") + `)` + if _, derr := tx.ExecContext(ctx, delQ, sourceArgs...); derr != nil { + slog.Warn("delete tasks: vault_links cleanup", "count", len(deleted), "err", derr) + } + + if err := tx.Commit(); err != nil { + return deleted, err + } } return deleted, nil } diff --git a/internal/store/sqlitestore/teams_tasks_activity.go b/internal/store/sqlitestore/teams_tasks_activity.go index e5ac69a5..c76f1dfa 100644 --- a/internal/store/sqlitestore/teams_tasks_activity.go +++ b/internal/store/sqlitestore/teams_tasks_activity.go @@ -7,6 +7,7 @@ import ( "database/sql" "encoding/json" "fmt" + "log/slog" "time" "github.com/google/uuid" @@ -184,12 +185,16 @@ func (s *SQLiteTeamStore) AttachFileToTask(ctx context.Context, att *store.TeamT if len(att.Metadata) == 0 { att.Metadata = json.RawMessage(`{}`) } + // SQLite has no GENERATED columns via modernc driver — compute app-side. + if att.BaseName == "" && att.Path != "" { + att.BaseName = ComputeAttachmentBaseName(att.Path) + } tid := tenantIDForInsert(ctx) res, err := s.db.ExecContext(ctx, - `INSERT INTO team_task_attachments (id, task_id, team_id, chat_id, path, file_size, mime_type, created_by_agent_id, created_by_sender_id, metadata, created_at, tenant_id) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `INSERT INTO team_task_attachments (id, task_id, team_id, chat_id, path, base_name, file_size, mime_type, created_by_agent_id, created_by_sender_id, metadata, created_at, tenant_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (task_id, path) DO NOTHING`, - att.ID, att.TaskID, att.TeamID, att.ChatID, att.Path, + att.ID, att.TaskID, att.TeamID, att.ChatID, att.Path, att.BaseName, att.FileSize, att.MimeType, att.CreatedByAgentID, nilStr(att.CreatedBySenderID), att.Metadata, att.CreatedAt, tid, @@ -276,6 +281,19 @@ func (s *SQLiteTeamStore) DetachFileFromTask(ctx context.Context, taskID uuid.UU if n, _ := res.RowsAffected(); n > 0 { _, _ = s.db.ExecContext(ctx, `UPDATE team_tasks SET attachment_count = MAX(attachment_count - 1, 0) WHERE id = ? AND tenant_id = ?`, taskID, tid) + + // Phase 04: clean up Phase 2.5 auto-links sourced from this task. + source := "task:" + taskID.String() + if _, derr := s.db.ExecContext(ctx, ` + DELETE FROM vault_links + WHERE json_extract(metadata, '$.source') = ? + AND ( + from_doc_id IN (SELECT id FROM vault_documents WHERE tenant_id = ?) + OR to_doc_id IN (SELECT id FROM vault_documents WHERE tenant_id = ?) + ) + `, source, tid, tid); derr != nil { + slog.Warn("vault.link.cleanup_on_detach", "task_id", taskID, "err", derr) + } } return nil } diff --git a/internal/store/team_store.go b/internal/store/team_store.go index 1ca009a7..61ec76f7 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -160,6 +160,16 @@ type TeamTaskEventData struct { CreatedAt time.Time `json:"created_at" db:"created_at"` } +// TaskSibling represents a vault document attached to the same team task as +// another file sharing the same basename. Returned by BatchGetTaskSiblingsByBasenames +// for Phase 04 task-based auto-linking. +type TaskSibling struct { + TaskID uuid.UUID `db:"task_id"` + DocID uuid.UUID `db:"doc_id"` + BaseName string `db:"base_name"` + AttachmentTime time.Time `db:"created_at"` +} + // TeamTaskAttachmentData represents a file attached to a team task (path-based, no FK to workspace). type TeamTaskAttachmentData struct { ID uuid.UUID `json:"id" db:"id"` @@ -167,6 +177,7 @@ type TeamTaskAttachmentData struct { TeamID uuid.UUID `json:"team_id" db:"team_id"` ChatID string `json:"chat_id,omitempty" db:"chat_id"` Path string `json:"path" db:"path"` + BaseName string `json:"base_name,omitempty" db:"base_name"` // PG: GENERATED; SQLite: app-populated FileSize int64 `json:"file_size" db:"file_size"` MimeType string `json:"mime_type,omitempty" db:"mime_type"` CreatedByAgentID *uuid.UUID `json:"created_by_agent_id,omitempty" db:"created_by_agent_id"` @@ -246,6 +257,17 @@ type TaskCommentStore interface { GetAttachment(ctx context.Context, attachmentID uuid.UUID) (*TeamTaskAttachmentData, error) ListTaskAttachments(ctx context.Context, taskID uuid.UUID) ([]TeamTaskAttachmentData, error) DetachFileFromTask(ctx context.Context, taskID uuid.UUID, path string) error + + // BatchGetTaskSiblingsByBasenames returns, for each basename, the vault + // documents attached to the SAME team task(s) as that basename. Excludes + // the source basename's own docs. Capped per (source_basename × task_id) + // at `limit` (red-team concern #11). Chunks input at 500. + BatchGetTaskSiblingsByBasenames( + ctx context.Context, + tenantID uuid.UUID, + basenames []string, + limit int, + ) (map[string][]TaskSibling, error) } // TaskRecoveryStore manages stale task detection and recovery. diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index 5ca900f7..f6052275 100644 --- a/internal/tools/context_keys.go +++ b/internal/tools/context_keys.go @@ -356,6 +356,29 @@ func TeamTaskIDFromCtx(ctx context.Context) string { return "" } +// --- Delegation ID propagation (delegate_tool → vault_interceptor) --- + +const ctxDelegationID toolContextKey = "tool_delegation_id" + +// WithDelegationID injects the delegation identifier into context so vault +// documents created during the delegated task can be tagged in metadata for +// Phase 05 auto-linking. +func WithDelegationID(ctx context.Context, id string) context.Context { + return context.WithValue(ctx, ctxDelegationID, id) +} + +// DelegationIDFromCtx returns the active delegation ID. Falls back to +// RunContext when no explicit context key is present. +func DelegationIDFromCtx(ctx context.Context) string { + if v, _ := ctx.Value(ctxDelegationID).(string); v != "" { + return v + } + if rc := store.RunContextFromCtx(ctx); rc != nil { + return rc.DelegationID + } + return "" +} + // --- Leader agent ID propagation (team task dispatch → memory interceptor) --- const ctxLeaderAgentID toolContextKey = "tool_leader_agent_id" diff --git a/internal/tools/context_keys_test.go b/internal/tools/context_keys_test.go index fd980751..f5144e23 100644 --- a/internal/tools/context_keys_test.go +++ b/internal/tools/context_keys_test.go @@ -3,8 +3,40 @@ package tools import ( "context" "testing" + + "github.com/nextlevelbuilder/goclaw/internal/store" ) +func TestWithDelegationID_RoundTrip(t *testing.T) { + ctx := context.Background() + if got := DelegationIDFromCtx(ctx); got != "" { + t.Errorf("empty context: want empty, got %q", got) + } + + ctx = WithDelegationID(ctx, "deleg-123") + if got := DelegationIDFromCtx(ctx); got != "deleg-123" { + t.Errorf("round-trip: want deleg-123, got %q", got) + } +} + +func TestDelegationIDFromCtx_RunContextFallback(t *testing.T) { + // When ctxDelegationID is NOT set but RunContext has the ID, + // DelegationIDFromCtx should return the RunContext value. + ctx := context.Background() + rc := &store.RunContext{DelegationID: "from-runcontext"} + ctx = store.WithRunContext(ctx, rc) + + if got := DelegationIDFromCtx(ctx); got != "from-runcontext" { + t.Errorf("fallback: want from-runcontext, got %q", got) + } + + // Explicit value wins over RunContext fallback. + ctx = WithDelegationID(ctx, "explicit") + if got := DelegationIDFromCtx(ctx); got != "explicit" { + t.Errorf("explicit-wins: want explicit, got %q", got) + } +} + func TestToolContextKeys_Channel(t *testing.T) { ctx := context.Background() if v := ToolChannelFromCtx(ctx); v != "" { diff --git a/internal/tools/team_mock_backend_test.go b/internal/tools/team_mock_backend_test.go index 36b02b97..c98e8e93 100644 --- a/internal/tools/team_mock_backend_test.go +++ b/internal/tools/team_mock_backend_test.go @@ -156,6 +156,9 @@ func (b *baseNoopTeamStore) ListTaskAttachments(_ context.Context, _ uuid.UUID) func (b *baseNoopTeamStore) DetachFileFromTask(_ context.Context, _ uuid.UUID, _ string) error { return fmt.Errorf("not implemented: DetachFileFromTask") } +func (b *baseNoopTeamStore) BatchGetTaskSiblingsByBasenames(_ context.Context, _ uuid.UUID, _ []string, _ int) (map[string][]store.TaskSibling, error) { + return nil, nil +} // TaskRecoveryStore func (b *baseNoopTeamStore) RecoverAllStaleTasks(_ context.Context) ([]store.RecoveredTaskInfo, error) { diff --git a/internal/tools/vault_interceptor.go b/internal/tools/vault_interceptor.go index e9d6597f..0e627e5f 100644 --- a/internal/tools/vault_interceptor.go +++ b/internal/tools/vault_interceptor.go @@ -79,6 +79,15 @@ func (v *VaultInterceptor) AfterWrite(ctx context.Context, resolvedPath, content DocType: docType, ContentHash: hash, } + // Phase 05: tag with delegation_id when write happens inside a delegated + // task so Phase 2.6 auto-linking can sibling-link the docs later. + if delegID := DelegationIDFromCtx(ctx); delegID != "" { + if doc.Metadata == nil { + doc.Metadata = make(map[string]any) + } + doc.Metadata["delegation_id"] = delegID + doc.Metadata["created_in"] = "delegation" + } if err := v.vaultStore.UpsertDocument(ctx, doc); err != nil { slog.Warn("vault.after_write", "path", relPath, "err", err) return @@ -142,6 +151,15 @@ func (v *VaultInterceptor) AfterWriteMedia(ctx context.Context, resolvedPath, su eventAgentID = agentID } + // Build metadata carefully: mime_type always set, plus optional + // delegation_id / created_in when running inside a delegation. + // Using explicit assignment preserves any future caller-supplied keys + // added via a metadata-capable variant — red-team concern #18. + meta := map[string]any{"mime_type": mimeType} + if delegID := DelegationIDFromCtx(ctx); delegID != "" { + meta["delegation_id"] = delegID + meta["created_in"] = "delegation" + } doc := &store.VaultDocument{ TenantID: tenantID, AgentID: agentIDPtr, @@ -152,7 +170,7 @@ func (v *VaultInterceptor) AfterWriteMedia(ctx context.Context, resolvedPath, su DocType: "media", ContentHash: hash, Summary: summary, - Metadata: map[string]any{"mime_type": mimeType}, + Metadata: meta, } if err := v.vaultStore.UpsertDocument(ctx, doc); err != nil { slog.Warn("vault.after_write_media", "path", relPath, "err", err) diff --git a/internal/vault/enrich_auto_linking.go b/internal/vault/enrich_auto_linking.go new file mode 100644 index 00000000..7ae811d6 --- /dev/null +++ b/internal/vault/enrich_auto_linking.go @@ -0,0 +1,177 @@ +package vault + +import ( + "context" + "log/slog" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// phase25TaskLinking implements task-based auto-linking (Phase 04). +// Runs after Phase 2 embed, before Phase 3 classify. Uses dedicated +// link_type LinkTypeTaskAttachment (outside validClassifyTypes) so +// DeleteDocLinksByTypes in classify cannot wipe these links. +// +// Single batched query to BatchGetTaskSiblingsByBasenames + single +// batched CreateLinks call — O(1) DB round trips per chunk. +// +// Nil-tolerant: teamStore may be unset (e.g. during tests), in which +// case the phase is a silent no-op. +func (w *enrichWorker) phase25TaskLinking(ctx context.Context, embedded []enriched, docMap map[string]*store.VaultDocument) { + if w.teamStore == nil || len(embedded) == 0 { + return + } + first := embedded[0].payload + tenantUUID, terr := uuid.Parse(first.TenantID) + if terr != nil { + slog.Warn("vault.enrich.phase2_5: parse tenant", "err", terr) + return + } + + // Collect unique basenames from the chunk using Phase 0's docMap. + baseNames := make([]string, 0, len(embedded)) + docByBasename := make(map[string][]*enriched, len(embedded)) + seenBase := make(map[string]bool, len(embedded)) + for i := range embedded { + ex := docMap[embedded[i].payload.DocID] + if ex == nil || ex.PathBasename == "" { + continue + } + if !seenBase[ex.PathBasename] { + seenBase[ex.PathBasename] = true + baseNames = append(baseNames, ex.PathBasename) + } + docByBasename[ex.PathBasename] = append(docByBasename[ex.PathBasename], &embedded[i]) + } + if len(baseNames) == 0 { + return + } + + siblingsByBase, err := w.teamStore.BatchGetTaskSiblingsByBasenames( + ctx, tenantUUID, baseNames, enrichTaskSiblingCap, + ) + if err != nil { + slog.Warn("vault.enrich.phase2_5: batch_task_siblings", "err", err) + return + } + if len(siblingsByBase) == 0 { + return + } + + var links []store.VaultLink + for baseName, docs := range docByBasename { + sibs := siblingsByBase[baseName] + if len(sibs) == 0 { + continue + } + for _, d := range docs { + seen := make(map[string]bool, len(sibs)) + for _, sib := range sibs { + sibID := sib.DocID.String() + if seen[sibID] || sibID == d.payload.DocID { + continue + } + seen[sibID] = true + source := "task:" + sib.TaskID.String() + links = append(links, store.VaultLink{ + FromDocID: d.payload.DocID, + ToDocID: sibID, + LinkType: LinkTypeTaskAttachment, + Context: source, // ONLY task id ref — no subject text (PII safety) + Metadata: map[string]any{"source": source}, + }) + } + } + } + if len(links) == 0 { + return + } + if err := w.vault.CreateLinks(ctx, links); err != nil { + slog.Debug("vault.enrich.phase2_5: create_links", "err", err) + return + } + slog.Info("vault.link.created", + "source_type", "task", + "count", len(links)) +} + +// phase26DelegationLinking implements delegation-based auto-linking +// (Phase 05). Runs after Phase 2.5, before Phase 3 classify. Uses +// dedicated link_type LinkTypeDelegationAttachment. +// +// Batched query via VaultStore.BatchFindByDelegationIDs; single +// CreateLinks call. No-op when no embedded doc carries metadata.delegation_id. +func (w *enrichWorker) phase26DelegationLinking(ctx context.Context, embedded []enriched, docMap map[string]*store.VaultDocument) { + if len(embedded) == 0 { + return + } + first := embedded[0].payload + tenant := first.TenantID + + delegIDs := make([]string, 0, len(embedded)) + docsByDelegID := make(map[string][]*store.VaultDocument, len(embedded)) + excludeDocIDs := make([]string, 0, len(embedded)) + seenDeleg := make(map[string]bool, len(embedded)) + + for i := range embedded { + existing := docMap[embedded[i].payload.DocID] + if existing == nil || len(existing.Metadata) == 0 { + continue + } + delegID, _ := existing.Metadata["delegation_id"].(string) + if delegID == "" { + continue + } + if !seenDeleg[delegID] { + seenDeleg[delegID] = true + delegIDs = append(delegIDs, delegID) + } + docsByDelegID[delegID] = append(docsByDelegID[delegID], existing) + excludeDocIDs = append(excludeDocIDs, existing.ID) + } + if len(delegIDs) == 0 { + return + } + + siblingsByDeleg, err := w.vault.BatchFindByDelegationIDs( + ctx, tenant, delegIDs, enrichTaskSiblingCap, excludeDocIDs, + ) + if err != nil { + slog.Debug("vault.enrich.phase2_6: delegation_siblings_batch", "err", err) + return + } + + var links []store.VaultLink + for delegID, srcDocs := range docsByDelegID { + sibs := siblingsByDeleg[delegID] + if len(sibs) == 0 { + continue + } + source := "delegation:" + delegID + for _, src := range srcDocs { + for _, sib := range sibs { + if sib.ID == src.ID { + continue + } + links = append(links, store.VaultLink{ + FromDocID: src.ID, + ToDocID: sib.ID, + LinkType: LinkTypeDelegationAttachment, + Context: source, + Metadata: map[string]any{"source": source}, + }) + } + } + } + if len(links) == 0 { + return + } + if err := w.vault.CreateLinks(ctx, links); err != nil { + slog.Debug("vault.enrich.phase2_6: create_links", "err", err) + return + } + slog.Info("vault.link.created", + "source_type", "delegation", + "count", len(links)) +} diff --git a/internal/vault/enrich_worker.go b/internal/vault/enrich_worker.go index fe2c36a3..4acaade7 100644 --- a/internal/vault/enrich_worker.go +++ b/internal/vault/enrich_worker.go @@ -5,15 +5,15 @@ import ( "encoding/json" "fmt" "log/slog" - - "github.com/google/uuid" - "github.com/nextlevelbuilder/goclaw/internal/bus" "os" "path/filepath" + "strconv" "strings" "sync" "time" + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/eventbus" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -42,7 +42,8 @@ type EnrichWorkerDeps struct { Provider providers.Provider Model string EventBus eventbus.DomainEventBus - MsgBus bus.EventPublisher // for WS event broadcast + MsgBus bus.EventPublisher // for WS event broadcast + TeamStore store.TaskCommentStore // for Phase 2.5 task-based auto-linking (nil-safe) } // RegisterEnrichWorker subscribes the enrichment worker to vault doc events. @@ -50,12 +51,13 @@ type EnrichWorkerDeps struct { func RegisterEnrichWorker(deps EnrichWorkerDeps) (func(), *EnrichProgress) { progress := NewEnrichProgress(deps.MsgBus) w := &enrichWorker{ - vault: deps.VaultStore, - provider: deps.Provider, - model: deps.Model, - dedup: make(map[string]string), - sem: semaphore.NewWeighted(enrichMaxConcurrent), - progress: progress, + vault: deps.VaultStore, + teamStore: deps.TeamStore, + provider: deps.Provider, + model: deps.Model, + dedup: make(map[string]string), + sem: semaphore.NewWeighted(enrichMaxConcurrent), + progress: progress, } unsub := deps.EventBus.Subscribe(eventbus.EventVaultDocUpserted, w.Handle) return unsub, progress @@ -64,11 +66,12 @@ func RegisterEnrichWorker(deps EnrichWorkerDeps) (func(), *EnrichProgress) { // enrichWorker processes vault document upsert events to generate summaries, // embeddings, and semantic links between related documents. type enrichWorker struct { - vault store.VaultStore - provider providers.Provider - model string - queue enrichBatchQueue - progress *EnrichProgress + vault store.VaultStore + teamStore store.TaskCommentStore // nil-tolerant — Phase 2.5 disabled when nil + provider providers.Provider + model string + queue enrichBatchQueue + progress *EnrichProgress // Bounded dedup: docID → content_hash. Prevents re-processing unchanged files. dedupMu sync.Mutex @@ -76,6 +79,18 @@ type enrichWorker struct { sem *semaphore.Weighted // limits concurrent LLM summarize calls } +// enrichTaskSiblingCap bounds the number of auto-linked siblings per +// (source_doc × task) pair. Tunable via VAULT_TASK_SIBLING_CAP env var so +// operators can raise/lower without a rebuild. +var enrichTaskSiblingCap = func() int { + if v := os.Getenv("VAULT_TASK_SIBLING_CAP"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return 9 +}() + // Handle is the EventBus handler for vault.doc_upserted events. func (w *enrichWorker) Handle(ctx context.Context, event eventbus.DomainEvent) error { payload, ok := event.Payload.(eventbus.VaultDocUpsertedPayload) @@ -208,12 +223,24 @@ func (w *enrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD var all []prepared for _, item := range pending { existing := docMap[item.DocID] + // Preceding short-circuit: existing non-empty summary wins. This + // protects LLM- or user-authored summaries from being clobbered by + // the deterministic synthesizer below on subsequent rescans. Phase 6 + // idempotent embedding relies on this stability. if existing != nil && existing.Summary != "" { all = append(all, prepared{payload: item, summary: existing.Summary, title: existing.Title}) continue } - if existing != nil && existing.DocType == "media" { - all = append(all, prepared{payload: item, title: existing.Title}) + // Phase 02: deterministic pseudo-summary for media + new `document` + // docType. Pure function — no file read, no LLM, works for any binary. + if existing != nil && (existing.DocType == "media" || existing.DocType == "document") { + mime, _ := existing.Metadata["mime_type"].(string) + summary := SynthesizeMediaSummary(existing.Path, mime) + all = append(all, prepared{ + payload: item, + title: existing.Title, + summary: summary, + }) continue } @@ -266,8 +293,23 @@ func (w *enrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD } // Phase 2 — Embed: update summary + embed per doc. + // Idempotent for media/document: if content hash and synthesized summary + // are unchanged AND non-empty, skip the DB write + embedding call. Text + // docs still re-embed every tick since their summary is LLM-generated. var embedded []enriched for _, r := range results { + existing := docMap[r.payload.DocID] + if existing != nil && + (existing.DocType == "media" || existing.DocType == "document") && + existing.ContentHash == r.payload.ContentHash && + existing.Summary == r.summary && + r.summary != "" { + slog.Debug("vault.enrich: skip_reembed_unchanged", + "doc", r.payload.DocID, + "doc_type", existing.DocType) + embedded = append(embedded, r) + continue + } if err := w.vault.UpdateSummaryAndReembed(ctx, r.payload.TenantID, r.payload.DocID, r.summary); err != nil { slog.Warn("vault.enrich: update_summary", "doc", r.payload.DocID, "err", err) continue @@ -275,6 +317,18 @@ func (w *enrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD embedded = append(embedded, r) } + // Phase 2.5 — Task-based auto-linking (deterministic, no LLM). + // Runs BEFORE classify so the dedicated link_type `task_attachment` + // (outside validClassifyTypes) survives DeleteDocLinksByTypes. + // Piggybacks on docMap from Phase 0 — O(n) basename collection, one + // batched query to BatchGetTaskSiblingsByBasenames, one batched + // CreateLinks call. Nil-tolerant: disabled when teamStore is nil. + w.phase25TaskLinking(ctx, embedded, docMap) + + // Phase 2.6 — Delegation-based auto-linking (deterministic, no LLM). + // Same invariants as Phase 2.5 but keyed on metadata.delegation_id. + w.phase26DelegationLinking(ctx, embedded, docMap) + // Phase 3 — Classify links for this chunk. if len(embedded) > 0 { first := embedded[0].payload @@ -373,13 +427,15 @@ func (w *enrichWorker) chatWithRetry(ctx context.Context, logPrefix string, req } // syncWikilinks extracts [[wikilinks]] from document content and syncs them as vault links. -// Skips binary/media files to avoid parsing garbage data as wikilinks. +// Skips binary/media/document files to avoid parsing garbage data as wikilinks +// (PDFs and office docs are binary and would produce garbage [[...]] matches +// while wasting a 4MB read buffer). func (w *enrichWorker) syncWikilinks(ctx context.Context, p eventbus.VaultDocUpsertedPayload) { doc, err := w.vault.GetDocumentByID(ctx, p.TenantID, p.DocID) if err != nil || doc == nil { return } - if doc.DocType == "media" { + if doc.DocType == "media" || doc.DocType == "document" { return }