Feature/con 118 campaign assistant assets

src/genkit/flows/campaign_assistant/prompts/campaign_assistant.tmpl

@@ -8,6 +8,7 @@ You can help with:
 4. Change the campaign's start/end dates (setCampaignDates), and redistribute the publish dates of its non-published drafts across the timeline (redistributePosts).
 5. Review consistency — check the brief for internal consistency and completeness (checkBrief), and check whether the campaign's non-published posts follow the brief (checkPostsConsistency). Both are read-only.
 6. Answer questions and give an overview — the brief, phases, existing posts, and how content is distributed (listCampaignPosts, getCampaignOverview, or straight from the context below).
+7. Answer questions grounded in the campaign's attached assets — uploaded PDFs, markdown, etc. (askCampaignAssets).
 
 ## Tools
 - runContentPlan: Call this when the user asks to generate, create, build, or regenerate a content plan (e.g. "generate a content plan", "make me some posts for this campaign"). It creates and saves the draft posts. Takes no arguments.
@@ -19,6 +20,7 @@ You can help with:
 - redistributePosts: Call this when the user asks to redistribute, re-spread, rebalance, or re-schedule the drafts / unpublished posts — e.g. "redistribute the drafts". It evenly re-dates only draft and ready-for-publish posts across the timeline, phase by phase; it never moves already-scheduled or published posts. Takes no arguments.
 - checkBrief: Call this when the user asks to check, review, or critique the brief, or whether the brief is consistent/complete — e.g. "is the brief consistent?", "review the brief". It is READ-ONLY — it returns findings with suggestions but does NOT change the brief. Takes no arguments. When it reports issues, summarise them and OFFER to improve the brief with enrichBrief — but do NOT call enrichBrief in the same turn unless the user also asks you to.
 - checkPostsConsistency: Call this when the user asks whether the posts match, follow, or are consistent with the brief — e.g. "do the posts follow the brief?", "check the posts against the brief". It is READ-ONLY — it returns per-post findings but does NOT change any post. It reviews only non-published posts and is capped; when the result reports a cap, MENTION that only the first N were checked. Pass `max` only if the user asks for a specific number.
+- askCampaignAssets: Call this when the user asks a question about the campaign's attached assets (uploaded PDFs, markdown, etc.) — e.g. "what does the pricing PDF say about enterprise tiers?", "summarise the attached brief". It is READ-ONLY: it returns the most relevant asset excerpts (with title + page) for you to answer from; it does NOT generate posts. Answer from the returned excerpts and cite the asset title(s) you used. If the result reports `available:false` or has no excerpts, tell the user you couldn't search or find matching asset content — do not invent an answer.
 
 Use a tool only when the user's request clearly matches it. For a general question you can answer from the campaign brief and phases already shown below, answer directly without a tool. In particular, questions about the phases themselves (their names or purpose) can be answered from the context below without a tool; only call getCampaignOverview when you need the post counts / distribution.
 
@@ -42,6 +44,8 @@ Rules for `action`:
 - "declined" — the request is out of scope (anything outside the capabilities above, e.g. deleting posts, publishing, billing). Explain briefly what you can help with instead.
 
 Keep `explanation` to a few sentences. Never put the full generated posts or the full brief into `explanation` — the client already receives those through their own channels.
+
+When a `runContentPlan` or `generatePosts` result includes `usedAssets`, briefly note in your `explanation` that the posts drew on those attached assets, naming them by title (e.g. "drawing on your pricing PDF and brand guide"). Say nothing about assets when `usedAssets` is empty.
 {{end}}
 {{define "context"}}
 ## Campaign

src/genkit/flows/campaign_assistant/run.go

@@ -99,6 +99,7 @@ func runCampaignAssistant(
 		campaign:         campaign,
 		repos:            repos,
 		onEvent:          onEvent,
+		embedder:         cfg.Embedder,
 		contentPlan:      cfg.ContentPlan,
 		enrichBrief:      cfg.EnrichBrief,
 		overview:         cfg.Overview,
@@ -193,7 +194,7 @@ func runCampaignAssistant(
 		ai.WithSystem(systemBlock),
 		ai.WithMessages(history...),
 		ai.WithPrompt(req.Instruction),
-		ai.WithTools(tools.runContentPlan, tools.enrichBrief, tools.listCampaignPosts, tools.getCampaignOverview, tools.generatePosts, tools.setCampaignDates, tools.redistributePosts, tools.checkBrief, tools.checkPostsConsistency),
+		ai.WithTools(tools.runContentPlan, tools.enrichBrief, tools.listCampaignPosts, tools.getCampaignOverview, tools.generatePosts, tools.setCampaignDates, tools.redistributePosts, tools.checkBrief, tools.checkPostsConsistency, tools.askCampaignAssets),
 		ai.WithMaxTurns(maxTurns),
 		ai.WithStreaming(streamCb),
 		cfg.Provider.CallConfig(maxTokens),

src/genkit/flows/campaign_assistant/tools.go

@@ -3,19 +3,24 @@ package campaign_assistant
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"sort"
 	"strings"
 	"time"
 
 	"github.com/firebase/genkit/go/ai"
 	"github.com/firebase/genkit/go/genkit"
+	"github.com/pgvector/pgvector-go"
 
 	"github.com/ogen-app/ogen/src/campaign_actions/overview"
 	"github.com/ogen-app/ogen/src/campaign_actions/reschedule"
+	"github.com/ogen-app/ogen/src/genkit/embedopts"
 	"github.com/ogen-app/ogen/src/genkit/flows/consistency"
 	"github.com/ogen-app/ogen/src/genkit/flows/content_plan"
 	"github.com/ogen-app/ogen/src/genkit/flows/enrich_brief"
+	"github.com/ogen-app/ogen/src/logging"
 	"github.com/ogen-app/ogen/src/models"
+	"github.com/ogen-app/ogen/src/repository"
 )
 
 // ── Context key for per-request state ────────────────────────────────────────
@@ -32,6 +37,9 @@ type requestState struct {
 	campaign   *models.Campaign
 	repos      CampaignAssistantRepos
 	onEvent    OnEventFunc
+	// embedder backs the askCampaignAssets read tool (CON-118); nil / unavailable
+	// degrades that tool to "search unavailable".
+	embedder ai.Embedder
 
 	// Injected sub-flow callbacks, invoked as tools.
 	contentPlan func(ctx context.Context, campaignID string, onEvent content_plan.OnEventFunc) (*content_plan.ContentPlanResponse, error)
@@ -82,8 +90,9 @@ type EnrichBriefOutput struct {
 
 // RunContentPlanOutput is returned to the model after a content plan runs.
 type RunContentPlanOutput struct {
-	PostCount    int `json:"postCount"`
-	WarningCount int `json:"warningCount"`
+	PostCount    int        `json:"postCount"`
+	WarningCount int        `json:"warningCount"`
+	UsedAssets   []AssetRef `json:"usedAssets,omitempty"` // CON-118: assets that informed the plan
 }
 
 // CampaignPostInfo is a single element of the listCampaignPosts output.
@@ -108,13 +117,14 @@ type GeneratePostsInput struct {
 
 // GeneratePostsOutput is returned to the model after targeted posts are created.
 type GeneratePostsOutput struct {
-	PostCount      int      `json:"postCount"`
-	RequestedCount int      `json:"requestedCount"`
-	Clamped        bool     `json:"clamped"` // true when requestedCount exceeded the per-call cap
-	PhaseID        string   `json:"phaseId"`
-	PhaseName      string   `json:"phaseName"`
-	Platforms      []string `json:"platforms"` // resolved platform names
-	Warnings       []string `json:"warnings,omitempty"`
+	PostCount      int        `json:"postCount"`
+	RequestedCount int        `json:"requestedCount"`
+	Clamped        bool       `json:"clamped"` // true when requestedCount exceeded the per-call cap
+	PhaseID        string     `json:"phaseId"`
+	PhaseName      string     `json:"phaseName"`
+	Platforms      []string   `json:"platforms"` // resolved platform names
+	Warnings       []string   `json:"warnings,omitempty"`
+	UsedAssets     []AssetRef `json:"usedAssets,omitempty"` // CON-118: assets that informed the posts
 }
 
 // SetCampaignDatesInput is the input for the setCampaignDates tool (CON-115).
@@ -142,6 +152,27 @@ type CheckPostsInput struct {
 	Max int `json:"max,omitempty" jsonschema:"description=Optional cap on how many posts to review; omit for the default."`
 }
 
+// AskCampaignAssetsInput is the input for the askCampaignAssets read tool (CON-118).
+type AskCampaignAssetsInput struct {
+	Query string `json:"query" jsonschema:"description=The question to answer from the campaign's attached assets, e.g. 'what does the pricing PDF say about enterprise tiers?'"`
+}
+
+// AskCampaignAssetsOutput returns the asset excerpts most relevant to the query
+// for the planner to answer from (CON-118).
+type AskCampaignAssetsOutput struct {
+	Excerpts  []AssetExcerpt `json:"excerpts"`
+	Available bool           `json:"available"`      // false when asset search could not run
+	Note      string         `json:"note,omitempty"` // status when excerpts is empty
+}
+
+// AssetExcerpt is one relevant chunk of an attached asset.
+type AssetExcerpt struct {
+	AssetID string `json:"assetId"`
+	Title   string `json:"title"`
+	Pages   string `json:"pages,omitempty"`
+	Text    string `json:"text"`
+}
+
 // ── Tool registration ────────────────────────────────────────────────────────
 
 type toolSet struct {
@@ -154,6 +185,7 @@ type toolSet struct {
 	redistributePosts     ai.ToolRef
 	checkBrief            ai.ToolRef
 	checkPostsConsistency ai.ToolRef
+	askCampaignAssets     ai.ToolRef
 }
 
 func defineTools(g *genkit.Genkit) *toolSet {
@@ -231,6 +263,16 @@ func defineTools(g *genkit.Genkit) *toolSet {
 		},
 	)
 
+	askCampaignAssets := genkit.DefineTool(g, "askCampaignAssets",
+		"Answers a question grounded in the campaign's attached assets (uploaded PDFs, markdown, etc.). "+
+			"Call this when the user asks what an attached asset says or wants facts pulled from the campaign's assets — e.g. \"what does the pricing PDF say about enterprise tiers?\". "+
+			"Read-only: it returns the most relevant excerpts (with asset title + page) for you to answer from; it does NOT generate posts. "+
+			"If the result reports available:false or no excerpts, tell the user you couldn't search / find matching asset content.",
+		func(ctx *ai.ToolContext, in AskCampaignAssetsInput) (*AskCampaignAssetsOutput, error) {
+			return toolAskCampaignAssets(ctx, in)
+		},
+	)
+
 	return &toolSet{
 		runContentPlan:        runContentPlan,
 		enrichBrief:           enrichBrief,
@@ -241,6 +283,7 @@ func defineTools(g *genkit.Genkit) *toolSet {
 		redistributePosts:     redistributePosts,
 		checkBrief:            checkBrief,
 		checkPostsConsistency: checkPostsConsistency,
+		askCampaignAssets:     askCampaignAssets,
 	}
 }
 
@@ -252,6 +295,9 @@ func toolRunContentPlan(ctx context.Context) (*RunContentPlanOutput, error) {
 		return nil, fmt.Errorf("content plan generation is not available")
 	}
 
+	// CON-118: generate from the campaign's attached assets when it has any.
+	ensureCampaignAssetUse(ctx, st)
+
 	emit(st.onEvent, SSEEventContentPlanStarted, ContentPlanStartedEventPayload{})
 
 	// Forward the sub-flow's native events, namespaced, so the client sees
@@ -273,9 +319,28 @@ func toolRunContentPlan(ctx context.Context) (*RunContentPlanOutput, error) {
 		return nil, err
 	}
 
-	res := &ContentPlanResult{PostCount: len(resp.Posts), Warnings: resp.Warnings}
+	// CON-118: report which attached assets informed the plan.
+	used := toAssetRefs(resp.UsedAssets)
+	if len(used) > 0 {
+		emit(st.onEvent, SSEEventAssetsUsed, AssetsUsedEventPayload{Assets: used})
+	}
+
+	res := &ContentPlanResult{PostCount: len(resp.Posts), Warnings: resp.Warnings, UsedAssets: used}
 	st.contentPlanResult = res
-	return &RunContentPlanOutput{PostCount: res.PostCount, WarningCount: len(res.Warnings)}, nil
+	return &RunContentPlanOutput{PostCount: res.PostCount, WarningCount: len(res.Warnings), UsedAssets: used}, nil
+}
+
+// toAssetRefs maps the content_plan provenance list into the assistant's local
+// AssetRef type for SSE events + tool output (CON-118).
+func toAssetRefs(in []content_plan.AssetRef) []AssetRef {
+	if len(in) == 0 {
+		return nil
+	}
+	out := make([]AssetRef, len(in))
+	for i, a := range in {
+		out[i] = AssetRef{ID: a.ID, Title: a.Title}
+	}
+	return out
 }
 
 func toolEnrichBrief(ctx context.Context, in EnrichBriefInput) (*EnrichBriefOutput, error) {
@@ -361,6 +426,8 @@ func toolGeneratePosts(ctx context.Context, in GeneratePostsInput) (*GeneratePos
 	if st.generatePosts == nil {
 		return nil, fmt.Errorf("targeted post generation is not available")
 	}
+	// CON-118: generate from the campaign's attached assets when it has any.
+	ensureCampaignAssetUse(ctx, st)
 	campaign := st.campaign
 	now := time.Now().UTC()
 
@@ -430,11 +497,18 @@ func toolGeneratePosts(ctx context.Context, in GeneratePostsInput) (*GeneratePos
 		return nil, err
 	}
 
+	// CON-118: report which attached assets informed the generated posts.
+	used := toAssetRefs(resp.UsedAssets)
+	if len(used) > 0 {
+		emit(st.onEvent, SSEEventAssetsUsed, AssetsUsedEventPayload{Assets: used})
+	}
+
 	st.generatedPostsResult = &GeneratedPostsResult{
 		PostCount:   len(resp.Posts),
 		PlatformIDs: platformIDs,
 		PhaseID:     phaseID,
 		Warnings:    resp.Warnings,
+		UsedAssets:  used,
 	}
 	return &GeneratePostsOutput{
 		PostCount:      len(resp.Posts),
@@ -444,9 +518,146 @@ func toolGeneratePosts(ctx context.Context, in GeneratePostsInput) (*GeneratePos
 		PhaseName:      phaseName,
 		Platforms:      platformNames,
 		Warnings:       resp.Warnings,
+		UsedAssets:     used,
 	}, nil
 }
 
+// minAskAssetsSimilarity is the cosine-similarity floor for asset Q&A. Lower
+// than content_plan's generation threshold: a specific question benefits from
+// recall, and the planner filters the returned excerpts (CON-118).
+const minAskAssetsSimilarity = 0.5
+
+// askAssetsChunkLimit caps how many chunks the Q&A tool returns to the planner.
+const askAssetsChunkLimit = 8
+
+// toolAskCampaignAssets answers a question grounded in the campaign's attached
+// assets by embedding the query and searching the assets' chunks (CON-118). It
+// is read-only and degrades cleanly (available:false / a note) rather than
+// failing the turn when the embedder is unavailable or nothing matches.
+func toolAskCampaignAssets(ctx context.Context, in AskCampaignAssetsInput) (*AskCampaignAssetsOutput, error) {
+	st := getRequestState(ctx)
+	query := strings.TrimSpace(in.Query)
+	if query == "" {
+		return nil, fmt.Errorf("ask a question about the campaign's assets")
+	}
+	if !embedopts.Available(st.embedder) || st.repos.Chunks == nil || st.repos.Assets == nil {
+		return &AskCampaignAssetsOutput{Available: false, Note: "asset search is unavailable right now"}, nil
+	}
+
+	ids, err := readyCampaignAssetIDs(ctx, st.campaign, st.repos.Assets)
+	if err != nil {
+		return nil, fmt.Errorf("resolve campaign assets: %w", err)
+	}
+	if len(ids) == 0 {
+		return &AskCampaignAssetsOutput{Available: true, Note: "this campaign has no ready assets to search"}, nil
+	}
+
+	qResp, err := st.embedder.Embed(ctx, &ai.EmbedRequest{
+		Input:   []*ai.Document{ai.DocumentFromText(query, nil)},
+		Options: embedopts.Query(),
+	})
+	if err != nil || len(qResp.Embeddings) == 0 {
+		return &AskCampaignAssetsOutput{Available: false, Note: "asset search is unavailable right now"}, nil
+	}
+
+	chunks, err := st.repos.Chunks.SearchSimilar(ctx, pgvector.NewHalfVector(qResp.Embeddings[0].Embedding), ids, minAskAssetsSimilarity, askAssetsChunkLimit)
+	if err != nil {
+		return nil, fmt.Errorf("search assets: %w", err)
+	}
+	if len(chunks) == 0 {
+		return &AskCampaignAssetsOutput{Available: true, Note: "no attached asset content matched the question"}, nil
+	}
+
+	titles := make(map[string]string)
+	excerpts := make([]AssetExcerpt, 0, len(chunks))
+	for _, c := range chunks {
+		title, ok := titles[c.AssetID]
+		if !ok {
+			if a, err := st.repos.Assets.GetByID(ctx, c.AssetID); err == nil {
+				title = a.Title
+			}
+			titles[c.AssetID] = title
+		}
+		excerpts = append(excerpts, AssetExcerpt{
+			AssetID: c.AssetID,
+			Title:   title,
+			Pages:   pageRef(c.PageStart, c.PageEnd),
+			Text:    c.Content,
+		})
+	}
+	return &AskCampaignAssetsOutput{Excerpts: excerpts, Available: true}, nil
+}
+
+// readyCampaignAssetIDs resolves the campaign's attached, ready asset IDs — the
+// explicit AssetIDs list when set, otherwise all tenant-ready assets — excluding
+// failed/partial. Mirrors content_plan's candidate resolution (CON-118). It does
+// NOT require campaign.UseAssets: that flag governs automatic inclusion during
+// generation, whereas Q&A is an explicit request to consult the assets.
+func readyCampaignAssetIDs(ctx context.Context, campaign *models.Campaign, assets repository.AssetRepository) ([]string, error) {
+	bad := func(status string) bool {
+		return status == models.AssetStatusFailed || status == models.AssetStatusPartial
+	}
+	if len(campaign.AssetIDs) > 0 {
+		out := make([]string, 0, len(campaign.AssetIDs))
+		for _, id := range campaign.AssetIDs {
+			a, err := assets.GetByID(ctx, id)
+			if err != nil || bad(a.Status) {
+				continue
+			}
+			out = append(out, a.ID)
+		}
+		return out, nil
+	}
+	all, err := assets.List(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]string, 0, len(all))
+	for _, a := range all {
+		if !bad(a.Status) {
+			out = append(out, a.ID)
+		}
+	}
+	return out, nil
+}
+
+// ensureCampaignAssetUse turns on asset-sourced content generation when the
+// campaign has ready attached assets but UseAssets is still off, and persists
+// the flag so it sticks for later turns and the UI (CON-118). content_plan
+// injects the attached assets into the generation prompt whenever UseAssets is
+// true, so flipping it here is all that's needed. Best-effort: if the flag
+// can't be persisted, generation just proceeds without assets, as before.
+func ensureCampaignAssetUse(ctx context.Context, st *requestState) {
+	if st.campaign.UseAssets || st.repos.Assets == nil || st.repos.Campaigns == nil {
+		return
+	}
+	if len(st.campaign.AssetIDs) == 0 {
+		return // no assets attached to this campaign
+	}
+	ids, err := readyCampaignAssetIDs(ctx, st.campaign, st.repos.Assets)
+	if err != nil || len(ids) == 0 {
+		return // nothing ready to use
+	}
+	st.campaign.UseAssets = true
+	if err := st.repos.Campaigns.Update(ctx, st.campaign); err != nil {
+		st.campaign.UseAssets = false // keep in-memory state consistent with the DB
+		slog.WarnContext(ctx, "could not enable asset use for generation",
+			logging.AttrComponent, "genkit.campaign_assistant",
+			"campaign_id", st.campaignID, logging.AttrError, err)
+	}
+}
+
+// pageRef renders an asset chunk's page span for citation (CON-118).
+func pageRef(start, end *int) string {
+	if start == nil {
+		return ""
+	}
+	if end == nil || *end == *start {
+		return fmt.Sprintf("p. %d", *start)
+	}
+	return fmt.Sprintf("pp. %d-%d", *start, *end)
+}
+
 // resolveTargetPlatforms maps requested platform names/ids to campaign-target
 // platform ids. It errors (naming the targets) when a requested platform isn't
 // one the campaign already targets — no silent scope expansion.