feat(vault): task + delegation auto-linking in enrich worker

- 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
This commit is contained in:
viettranx committed 2026-04-11 21:22:23 +07:00
1 parent 2644a08b2b
commit e62c027b7d
17 files changed
+730 -33

No files matched your search

+1
View File
@@ -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()
+3
View File
@@ -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.
+106
View File
@@ -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
}
+64 -5
View File
@@ -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).
+18
View File
@@ -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
}
@@ -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
+1
View File
@@ -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
}
@@ -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
}
+45 -4
View File
@@ -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
}
@@ -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
}
+22
View File
@@ -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.
+23
View File
@@ -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"
+32
View File
@@ -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 != "" {
+3
View File
@@ -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) {
+19 -1
View File
@@ -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)
+177
View File
@@ -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))
}
+75 -19
View File
@@ -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
}