diff --git a/src/config/config.go b/src/config/config.go index 2e03218..5801ec1 100644 --- a/src/config/config.go +++ b/src/config/config.go @@ -123,6 +123,12 @@ type Config struct { // tool (CON-114) so it stays an incremental add rather than a full re-plan. GeneratePostsMax int `envconfig:"GENERATE_POSTS_MAX" default:"10"` + // DraftPostMax caps the Campaign Assistant's draftPost tool (CON-207) — how + // many finished, content-first drafts one "create a post from this research" + // call may produce (total across platforms). Smaller than GeneratePostsMax: + // each draft is full-length copy, not a terse thesis. + DraftPostMax int `envconfig:"DRAFT_POST_MAX" default:"5"` + // ConsistencyPostsMax caps how many posts the Campaign Assistant's posts-vs- // brief consistency check (CON-116) analyzes in a single model call. ConsistencyPostsMax int `envconfig:"CONSISTENCY_POSTS_MAX" default:"20"` diff --git a/src/genkit/flows/campaign_assistant/draft_post_tool_test.go b/src/genkit/flows/campaign_assistant/draft_post_tool_test.go new file mode 100644 index 0000000..5ab1416 --- /dev/null +++ b/src/genkit/flows/campaign_assistant/draft_post_tool_test.go @@ -0,0 +1,229 @@ +package campaign_assistant + +import ( + "context" + "fmt" + "testing" + + "github.com/ogen-app/ogen/src/genkit/flows/draft_post" + "github.com/ogen-app/ogen/src/models" + "github.com/ogen-app/ogen/src/repository" +) + +// fakeMessagesRepo is a minimal CampaignAssistantMessageRepository that returns +// a preset conversation history (oldest-first, as the real repo does). +type fakeMessagesRepo struct { + repository.CampaignAssistantMessageRepository + msgs []models.CampaignAssistantMessage +} + +func (f *fakeMessagesRepo) ListRecentByCampaignID(context.Context, string, int) ([]models.CampaignAssistantMessage, error) { + return f.msgs, nil +} + +func modelMsg(content string) models.CampaignAssistantMessage { + return models.CampaignAssistantMessage{Role: "model", Content: content} +} +func userMsg(content string) models.CampaignAssistantMessage { + return models.CampaignAssistantMessage{Role: "user", Content: content} +} + +// CON-207 FR2: the source material is the latest *answer* in the chat, pulled +// from the stored JSON envelope's "explanation" field — an override wins, an +// action-confirmation turn (e.g. post_drafted) is skipped, and legacy plain-text +// model messages are used as-is. +func TestResolveDraftSource(t *testing.T) { + st := func(msgs []models.CampaignAssistantMessage) *requestState { + return &requestState{campaignID: "c1", repos: CampaignAssistantRepos{Messages: &fakeMessagesRepo{msgs: msgs}}} + } + + // Override always wins, no history consulted. + if got, _ := resolveDraftSource(context.Background(), st(nil), " pasted "); got != "pasted" { + t.Fatalf("override = %q, want pasted", got) + } + + // Latest answered explanation is used. + got, err := resolveDraftSource(context.Background(), st([]models.CampaignAssistantMessage{ + userMsg("what do the assets say?"), + modelMsg(`{"action":"answered","explanation":"the research"}`), + }), "") + if err != nil || got != "the research" { + t.Fatalf("answered = %q err=%v", got, err) + } + + // A post_drafted confirmation is skipped in favour of the earlier answer. + got, _ = resolveDraftSource(context.Background(), st([]models.CampaignAssistantMessage{ + modelMsg(`{"action":"answered","explanation":"research A"}`), + userMsg("draft it"), + modelMsg(`{"action":"post_drafted","explanation":"I drafted 1 post."}`), + }), "") + if got != "research A" { + t.Fatalf("skip-confirmation = %q, want research A", got) + } + + // No prior research → empty (the tool then declines). + if got, _ := resolveDraftSource(context.Background(), st([]models.CampaignAssistantMessage{userMsg("hi")}), ""); got != "" { + t.Fatalf("no-research = %q, want empty", got) + } + + // Legacy plain-text model message (not JSON-wrapped) is used directly. + if got, _ := resolveDraftSource(context.Background(), st([]models.CampaignAssistantMessage{modelMsg("plain answer")}), ""); got != "plain answer" { + t.Fatalf("legacy = %q, want plain answer", got) + } +} + +// stubDraftPost records each request and returns req.Count synthetic posts. +func stubDraftPost(calls *[]draft_post.DraftPostRequest) func(context.Context, draft_post.DraftPostRequest, draft_post.OnEventFunc) (*draft_post.DraftPostResponse, error) { + return func(_ context.Context, req draft_post.DraftPostRequest, _ draft_post.OnEventFunc) (*draft_post.DraftPostResponse, error) { + *calls = append(*calls, req) + posts := make([]draft_post.DraftedPost, req.Count) + for i := range posts { + posts[i] = draft_post.DraftedPost{PostID: fmt.Sprintf("%s-%d", req.PlatformID, i), PlatformID: req.PlatformID, PublishDate: req.WindowStart} + } + return &draft_post.DraftPostResponse{Posts: posts}, nil + } +} + +func newDraftState(msgs []models.CampaignAssistantMessage, calls *[]draft_post.DraftPostRequest) *requestState { + return &requestState{ + campaignID: "c1", + campaign: timelineCampaign(), + instruction: "make it punchy", + repos: CampaignAssistantRepos{Messages: &fakeMessagesRepo{msgs: msgs}}, + draftPost: stubDraftPost(calls), + maxDraftPosts: 5, + } +} + +// No research in the chat and no override → the tool declines without running +// the flow (CON-207 §9). +func TestToolDraftPost_DeclinesWithoutSource(t *testing.T) { + var calls []draft_post.DraftPostRequest + st := newDraftState([]models.CampaignAssistantMessage{userMsg("hi")}, &calls) + ctx := withRequestState(context.Background(), st) + + out, err := toolDraftPost(ctx, DraftPostInput{Platforms: []string{"LinkedIn"}, Count: 1}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if out.Note != noDraftSourceNote { + t.Fatalf("note = %q, want the no-source note", out.Note) + } + if len(calls) != 0 { + t.Fatalf("flow must not run without source, got %d calls", len(calls)) + } + if st.draftPostResult != nil { + t.Fatal("draftPostResult must stay nil on decline") + } + // The decline must NOT consume the turn's heavy-action slot (CON-213): a later + // valid heavy tool call must still be able to reserve and run. + if !st.reserveHeavyAction() { + t.Fatal("a no-source decline must leave the heavy-action slot available") + } +} + +// The per-call cap is a TOTAL budget across platforms: count=3 over two +// platforms with max 5 yields 3 + 2 and reports clamped (CON-207 §10). +func TestToolDraftPost_BudgetAcrossPlatforms(t *testing.T) { + var calls []draft_post.DraftPostRequest + st := newDraftState([]models.CampaignAssistantMessage{ + modelMsg(`{"action":"answered","explanation":"the research"}`), + }, &calls) + ctx := withRequestState(context.Background(), st) + + out, err := toolDraftPost(ctx, DraftPostInput{Platforms: []string{"LinkedIn", "Threads"}, Count: 3}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if out.PostCount != 5 || !out.Clamped { + t.Fatalf("out = {PostCount:%d Clamped:%v}, want {5 true}", out.PostCount, out.Clamped) + } + if len(calls) != 2 || calls[0].Count != 3 || calls[1].Count != 2 { + t.Fatalf("flow calls counts = [%d %d], want [3 2]", calls[0].Count, calls[1].Count) + } + // Source + steering are threaded to the flow. + if calls[0].SourceMaterial != "the research" || calls[0].Instruction != "make it punchy" { + t.Fatalf("call[0] source=%q instruction=%q", calls[0].SourceMaterial, calls[0].Instruction) + } + if st.draftPostResult == nil || st.draftPostResult.PostCount != 5 { + t.Fatalf("draftPostResult = %+v", st.draftPostResult) + } +} + +// CON-215 parity: a user-correctable input (past date, non-target platform) +// fails soft — zero posts + a warning, no flow call, draftPostResult unset, and +// the heavy-action slot left free — so the turn stays conversational instead of +// aborting with a raw "model call failed". +func TestToolDraftPost_SoftFailsUserInput(t *testing.T) { + research := []models.CampaignAssistantMessage{ + modelMsg(`{"action":"answered","explanation":"the research"}`), + } + + // A past publish date. + var calls []draft_post.DraftPostRequest + st := newDraftState(research, &calls) + ctx := withRequestState(context.Background(), st) + out, err := toolDraftPost(ctx, DraftPostInput{Platforms: []string{"LinkedIn"}, Count: 1, PublishDate: "2020-01-01"}) + if err != nil { + t.Fatalf("past date must not error: %v", err) + } + if out.PostCount != 0 || len(out.Warnings) == 0 { + t.Fatalf("past date: out = {PostCount:%d Warnings:%v}, want zero posts + a warning", out.PostCount, out.Warnings) + } + if len(calls) != 0 || st.draftPostResult != nil { + t.Fatal("past date must not run the flow or set draftPostResult") + } + if !st.reserveHeavyAction() { + t.Fatal("a soft failure must leave the heavy-action slot available") + } + + // A non-target platform. + calls = nil + st = newDraftState(research, &calls) + ctx = withRequestState(context.Background(), st) + out, err = toolDraftPost(ctx, DraftPostInput{Platforms: []string{"TikTok"}, Count: 1}) + if err != nil { + t.Fatalf("non-target platform must not error: %v", err) + } + if out.PostCount != 0 || len(out.Warnings) == 0 || len(calls) != 0 { + t.Fatalf("non-target platform: out = %+v, calls = %d", out, len(calls)) + } +} + +// An explicit sourceMaterial override bypasses history lookup. +func TestToolDraftPost_SourceOverride(t *testing.T) { + var calls []draft_post.DraftPostRequest + st := newDraftState(nil, &calls) // no history at all + ctx := withRequestState(context.Background(), st) + + if _, err := toolDraftPost(ctx, DraftPostInput{Platforms: []string{"LinkedIn"}, Count: 1, SourceMaterial: "pasted material"}); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(calls) != 1 || calls[0].SourceMaterial != "pasted material" { + t.Fatalf("override not used: %+v", calls) + } +} + +// CON-213: once a heavy action ran this turn, draftPost yields the skip note and +// never runs the flow. +func TestToolDraftPost_HeavyLatch(t *testing.T) { + var calls []draft_post.DraftPostRequest + st := newDraftState([]models.CampaignAssistantMessage{ + modelMsg(`{"action":"answered","explanation":"the research"}`), + }, &calls) + if !st.reserveHeavyAction() { + t.Fatal("first reservation should win") + } + ctx := withRequestState(context.Background(), st) + + out, err := toolDraftPost(ctx, DraftPostInput{Platforms: []string{"LinkedIn"}, Count: 1}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if out.Note != heavySkipNote { + t.Fatalf("note = %q, want heavy skip note", out.Note) + } + if len(calls) != 0 { + t.Fatal("flow must not run once a heavy action was reserved") + } +} diff --git a/src/genkit/flows/campaign_assistant/prewarm.go b/src/genkit/flows/campaign_assistant/prewarm.go index 56060e0..c64e284 100644 --- a/src/genkit/flows/campaign_assistant/prewarm.go +++ b/src/genkit/flows/campaign_assistant/prewarm.go @@ -34,7 +34,7 @@ func prewarmToolCache(g *genkit.Genkit, cfg CampaignAssistantFlowConfig, t *tool ai.WithPrompt("warmup"), ai.WithTools( t.runContentPlan, t.enrichBrief, t.listCampaignPosts, t.getCampaignOverview, - t.generatePosts, t.setCampaignDates, t.redistributePosts, t.checkBrief, t.checkPostsConsistency, + t.generatePosts, t.draftPost, t.setCampaignDates, t.redistributePosts, t.checkBrief, t.checkPostsConsistency, ), ai.WithMaxTurns(1), cfg.Provider.CallConfig(1), // max_tokens: 1 — grammar compiles during prep diff --git a/src/genkit/flows/campaign_assistant/prompts/campaign_assistant.tmpl b/src/genkit/flows/campaign_assistant/prompts/campaign_assistant.tmpl index aa56167..eccada3 100644 --- a/src/genkit/flows/campaign_assistant/prompts/campaign_assistant.tmpl +++ b/src/genkit/flows/campaign_assistant/prompts/campaign_assistant.tmpl @@ -9,6 +9,7 @@ You can help with: 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). +8. Draft a post from research — turn research just discussed in this chat into one or more finished, ready-to-publish post drafts (draftPost). ## 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. @@ -21,22 +22,24 @@ You can help with: - 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. +- draftPost: Call this when the user asks to turn the research just discussed into an actual post — e.g. "create a post with this info for tomorrow, for LinkedIn", "draft 3 LinkedIn posts from that research", "write a post based on what you just told me". It rewrites the research into finished, ready-to-publish copy (a full post body, NOT a bullet outline) and saves it as a draft. This is DIFFERENT from generatePosts, which produces terse bullet-point draft-theses for bulk planning: use draftPost when the user wants a real, finished post built from a specific piece of research or pasted material, and generatePosts when they want to bulk-add planning drafts for a platform/timeframe. Only platforms the campaign already targets are allowed; if the user names a different one, do NOT call the tool — tell them it isn't a target platform. Resolve any date into ISO YYYY-MM-DD against today's date below and pass it as publishDate (omit for the next two weeks); the date must be today or later. Set count to the number the user asks for. Do NOT paste the research into sourceMaterial — it is loaded automatically from this conversation; pass sourceMaterial only when the user gave brand-new material in this same message. If there's no prior research in the chat, the tool will tell you — then ask the user to research the assets first. If draftPost comes back with zero posts and a warning (e.g. the date is in the past or the platform isn't a target), do NOT report success — relay the warning and suggest a valid alternative (today or a later date, or one of the campaign's target platforms). 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. -Run at most ONE generation or review action per message — one of runContentPlan, generatePosts, enrichBrief, checkBrief, or checkPostsConsistency. After it returns, stop calling action tools and reply. If the user asked for several, do the most important one now and offer the rest as a follow-up. (Read tools like getCampaignOverview, listCampaignPosts, and askCampaignAssets may be used first to inform that one action.) If an action tool returns a `note` saying one already ran this turn, do not retry — just reply with what was done. +Run at most ONE generation or review action per message — one of runContentPlan, generatePosts, draftPost, enrichBrief, checkBrief, or checkPostsConsistency. After it returns, stop calling action tools and reply. If the user asked for several, do the most important one now and offer the rest as a follow-up. (Read tools like getCampaignOverview, listCampaignPosts, and askCampaignAssets may be used first to inform that one action.) If an action tool returns a `note` saying one already ran this turn, do not retry — just reply with what was done. ## Response format After any tool calls, reply with a SINGLE JSON object and nothing else — no markdown fences, no prose before or after: { "explanation": "", - "action": "" + "action": "" } Rules for `action`: - "content_plan_generated" — you called runContentPlan this turn. - "posts_generated" — you called generatePosts this turn. +- "post_drafted" — you called draftPost this turn. - "brief_enriched" — you called enrichBrief this turn. - "dates_updated" — you called setCampaignDates this turn. - "posts_redistributed" — you called redistributePosts this turn. diff --git a/src/genkit/flows/campaign_assistant/run.go b/src/genkit/flows/campaign_assistant/run.go index 7ef3289..9b4ac91 100644 --- a/src/genkit/flows/campaign_assistant/run.go +++ b/src/genkit/flows/campaign_assistant/run.go @@ -99,12 +99,15 @@ func runCampaignAssistant( campaign: campaign, repos: repos, onEvent: onEvent, + instruction: req.Instruction, embedder: cfg.Embedder, contentPlan: cfg.ContentPlan, enrichBrief: cfg.EnrichBrief, overview: cfg.Overview, generatePosts: cfg.GeneratePosts, maxGeneratePosts: cfg.MaxGeneratePosts, + draftPost: cfg.DraftPost, + maxDraftPosts: cfg.MaxDraftPosts, checkBrief: cfg.CheckBrief, checkPosts: cfg.CheckPosts, } @@ -194,7 +197,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, tools.askCampaignAssets), + ai.WithTools(tools.runContentPlan, tools.enrichBrief, tools.listCampaignPosts, tools.getCampaignOverview, tools.generatePosts, tools.draftPost, tools.setCampaignDates, tools.redistributePosts, tools.checkBrief, tools.checkPostsConsistency, tools.askCampaignAssets), ai.WithMaxTurns(maxTurns), ai.WithStreaming(streamCb), cfg.Provider.CallConfig(maxTokens), @@ -276,6 +279,17 @@ func runCampaignAssistant( } } } + if st.draftPostResult != nil { + result.Action = "post_drafted" + result.DraftedPosts = st.draftPostResult + if result.Explanation == "" { + if st.draftPostResult.PostCount == 0 { + result.Explanation = "I couldn't draft a post from that research." + } else { + result.Explanation = fmt.Sprintf("I drafted %d post(s) from your research, ready for review.", st.draftPostResult.PostCount) + } + } + } if st.datesResult != nil { result.Action = "dates_updated" result.Dates = st.datesResult @@ -323,7 +337,7 @@ func runCampaignAssistant( // Pure-prose recovery: the model ignored the JSON envelope and answered in // plain prose (common for informational questions) and no tool ran. Salvage // the raw text as an "answered" reply. - if result.Explanation == "" && st.contentPlanResult == nil && st.briefResult == nil && st.generatedPostsResult == nil && st.datesResult == nil && st.redistributeResult == nil && st.briefReviewResult == nil && st.postsReviewResult == nil { + if result.Explanation == "" && st.contentPlanResult == nil && st.briefResult == nil && st.generatedPostsResult == nil && st.draftPostResult == nil && st.datesResult == nil && st.redistributeResult == nil && st.briefReviewResult == nil && st.postsReviewResult == nil { raw := strings.TrimSpace(scanner.FullText()) if raw != "" && !strings.Contains(raw, "{") { slog.WarnContext(ctx, "model emitted prose-only response, treating as answered", logging.AttrComponent, "genkit.campaign_assistant", "campaign_id", req.CampaignID, "len", len(raw)) @@ -412,6 +426,12 @@ func runCampaignAssistant( Warnings: result.GeneratedPosts.Warnings, }) } + if result.DraftedPosts != nil { + emit(onEvent, SSEEventDraftPostComplete, DraftPostCompleteEventPayload{ + PostCount: result.DraftedPosts.PostCount, + Warnings: result.DraftedPosts.Warnings, + }) + } if result.Dates != nil { emit(onEvent, SSEEventDatesUpdated, DatesUpdatedEventPayload{ StartDate: result.Dates.StartDate, diff --git a/src/genkit/flows/campaign_assistant/tools.go b/src/genkit/flows/campaign_assistant/tools.go index b31ac28..5da0bef 100644 --- a/src/genkit/flows/campaign_assistant/tools.go +++ b/src/genkit/flows/campaign_assistant/tools.go @@ -2,6 +2,7 @@ package campaign_assistant import ( "context" + "encoding/json" "fmt" "log/slog" "sort" @@ -18,6 +19,7 @@ import ( "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/draft_post" "github.com/ogen-app/ogen/src/genkit/flows/enrich_brief" "github.com/ogen-app/ogen/src/logging" "github.com/ogen-app/ogen/src/models" @@ -38,6 +40,9 @@ type requestState struct { campaign *models.Campaign repos CampaignAssistantRepos onEvent OnEventFunc + // instruction is the current user turn, passed to draftPost as steering for + // the Sonnet drafter (CON-207 FR2). + instruction string // embedder backs the askCampaignAssets read tool (CON-118); nil / unavailable // degrades that tool to "search unavailable". embedder ai.Embedder @@ -51,6 +56,10 @@ type requestState struct { // maxGeneratePosts caps a single call. generatePosts func(ctx context.Context, req content_plan.GeneratePostsRequest, onEvent content_plan.OnEventFunc) (*content_plan.ContentPlanResponse, error) maxGeneratePosts int + // draftPost backs the draftPost tool (CON-207): rewrite chat research into + // extended, content-first post drafts; maxDraftPosts caps a single call. + draftPost func(ctx context.Context, req draft_post.DraftPostRequest, onEvent draft_post.OnEventFunc) (*draft_post.DraftPostResponse, error) + maxDraftPosts int // checkBrief / checkPosts back the read-only consistency review tools (CON-116). checkBrief func(ctx context.Context, campaignID string, onEvent consistency.OnEventFunc) (*consistency.BriefReview, error) checkPosts func(ctx context.Context, req consistency.PostsCheckRequest, onEvent consistency.OnEventFunc) (*consistency.PostsReview, error) @@ -69,6 +78,7 @@ type requestState struct { contentPlanResult *ContentPlanResult briefResult *BriefResult generatedPostsResult *GeneratedPostsResult + draftPostResult *DraftPostResult datesResult *DatesResult redistributeResult *RedistributeResult briefReviewResult *consistency.BriefReview @@ -173,6 +183,36 @@ type GeneratePostsOutput struct { Note string `json:"note,omitempty"` } +// DraftPostInput is the input for the draftPost tool (CON-207). The model +// resolves the platform, count, and publish date from the conversation before +// calling; the source material (the research to rewrite) is loaded server-side +// from the chat unless SourceMaterial overrides it. +type DraftPostInput struct { + Platforms []string `json:"platforms" jsonschema:"description=Platform names or ids to draft for, e.g. [\"LinkedIn\"]. Must be platforms the campaign already targets."` + Count int `json:"count,omitempty" jsonschema:"description=Number of posts to draft per platform: the exact number the user names (\"draft 1 post\"->1, \"3 LinkedIn posts\"->3). Omit for 1. Capped per call."` + PublishDate string `json:"publishDate,omitempty" jsonschema:"description=Publish date (ISO YYYY-MM-DD) resolved from the requested timeframe against today; omit to spread across the next two weeks. Must be today or later."` + PostType string `json:"postType,omitempty" jsonschema:"description=Optional post-type slug (e.g. text-post, article); omit for the platform default."` + SourceMaterial string `json:"sourceMaterial,omitempty" jsonschema:"description=Optional override source text to turn into posts. Omit to use the research already discussed earlier in this chat (the normal case)."` +} + +// DraftPostOutput is returned to the model after drafts are created. +type DraftPostOutput struct { + PostCount int `json:"postCount"` + RequestedCount int `json:"requestedCount"` + Clamped bool `json:"clamped"` // true when the request exceeded the per-call cap + PlatformIDs []string `json:"platformIds"` + Platforms []string `json:"platforms"` // resolved platform names + PhaseID string `json:"phaseId"` + PhaseName string `json:"phaseName"` + Dates []string `json:"dates,omitempty"` // actual publish dates of the created posts + Warnings []string `json:"warnings,omitempty"` + UsedAssets []AssetRef `json:"usedAssets,omitempty"` + // Note carries a status when the tool short-circuited without running — a + // heavy action already ran this turn (CON-213), or there's no research to + // draft from (CON-207). + Note string `json:"note,omitempty"` +} + // SetCampaignDatesInput is the input for the setCampaignDates tool (CON-115). // The model resolves relative phrasing ("beginning of July") against today. type SetCampaignDatesInput struct { @@ -227,6 +267,7 @@ type toolSet struct { listCampaignPosts ai.ToolRef getCampaignOverview ai.ToolRef generatePosts ai.ToolRef + draftPost ai.ToolRef setCampaignDates ai.ToolRef redistributePosts ai.ToolRef checkBrief ai.ToolRef @@ -275,6 +316,17 @@ func defineTools(g *genkit.Genkit) *toolSet { }, ) + draftPost := genkit.DefineTool(g, "draftPost", + "Turns research already discussed in THIS chat (typically an askCampaignAssets answer) into one or more finished, ready-to-publish post drafts — full copy, not a bullet outline — for a specific platform. "+ + "Call this when the user asks to 'create/write/draft a post with this info / from this research / based on that' after you've answered a question about the campaign's assets. "+ + "Different from generatePosts (which produces terse draft-thesis outlines for bulk planning) and runContentPlan (which regenerates the whole plan). Only platforms the campaign already targets are allowed. "+ + "Resolve the publish date to ISO YYYY-MM-DD against today's date shown in the context; omit it to spread across the next two weeks. Set count to the number the user asks for. "+ + "The research is loaded automatically from the conversation — do NOT paste it back; pass sourceMaterial only when the user supplied fresh material in this very message. Returns how many posts were drafted.", + func(ctx *ai.ToolContext, in DraftPostInput) (*DraftPostOutput, error) { + return toolDraftPost(ctx, in) + }, + ) + setCampaignDates := genkit.DefineTool(g, "setCampaignDates", "Changes the campaign's start and/or end date. Call this when the user asks to move, shift, extend, or shorten the campaign's dates (e.g. 'move the campaign end to the beginning of July', 'push the start to next Monday'). "+ "Resolve any relative phrasing into ISO YYYY-MM-DD dates using today's date shown in the context; pass only the field(s) that change. The change is saved automatically. "+ @@ -325,6 +377,7 @@ func defineTools(g *genkit.Genkit) *toolSet { listCampaignPosts: listCampaignPosts, getCampaignOverview: getCampaignOverview, generatePosts: generatePosts, + draftPost: draftPost, setCampaignDates: setCampaignDates, redistributePosts: redistributePosts, checkBrief: checkBrief, @@ -576,6 +629,198 @@ func toolGeneratePosts(ctx context.Context, in GeneratePostsInput) (*GeneratePos }, nil } +// noDraftSourceNote steers the planner to ask the user to research first when +// there's no source material to draft from (CON-207). +const noDraftSourceNote = "There's no research in this chat to turn into a post yet. Ask me a question about the campaign's assets first (or paste the material into your message), then I'll draft a post from it." + +// toolDraftPost turns research already discussed in the chat into finished, +// content-first post drafts (CON-207). It loads the source material server-side +// (the latest assistant answer, or a paste-in override), resolves the platform / +// phase / count / date with the same helpers as generatePosts, then runs the +// Sonnet draft_post flow once per platform and persists each draft content-first. +func toolDraftPost(ctx context.Context, in DraftPostInput) (*DraftPostOutput, error) { + st := getRequestState(ctx) + if st.draftPost == nil { + return nil, fmt.Errorf("post drafting is not available") + } + campaign := st.campaign + now := time.Now().UTC() + + // Source material: an explicit override, else the latest research answer from + // earlier in this chat. Resolved BEFORE the heavy-action reservation so a + // no-source decline (a graceful non-error return, unlike the other heavy + // tools) leaves the turn's single heavy slot free for another heavy tool. + source, err := resolveDraftSource(ctx, st, in.SourceMaterial) + if err != nil { + return nil, err + } + if strings.TrimSpace(source) == "" { + return &DraftPostOutput{Note: noDraftSourceNote}, nil + } + + // Reuse the generatePosts resolvers verbatim (CON-114). A non-target platform, + // unknown phase, or past/invalid date is user-correctable, so it fails soft + // (CON-215) — a zero-post warning the planner relays — rather than aborting the + // turn; these run before the heavy-action reservation, so they never burn the slot. + platformIDs, platformNames, err := resolveTargetPlatforms(campaign, in.Platforms) + if err != nil { + return softDraftFailure(err) + } + // draftPost always targets the current phase — it drafts "now", from what was + // just discussed. (A future revision can accept an explicit phase.) + phaseID, phaseName, err := resolvePhase(campaign, "", now) + if err != nil { + return softDraftFailure(err) + } + // A single publish date: pin both window bounds to it so N posts land on that + // day; omitted → the next two weeks, across which the flow spreads them. + windowStart, windowEnd, err := resolveWindow(in.PublishDate, in.PublishDate, now) + if err != nil { + return softDraftFailure(err) + } + + // Reserve the turn's single heavy-action slot only now — after source and + // input validation succeed — so a no-source or invalid-input decline above + // never burns the slot for a later heavy tool (CON-213). The reservation still + // precedes every st.draftPost flow call, so two heavy tools dispatched in + // parallel can never both generate. + if !st.reserveHeavyAction() { + return &DraftPostOutput{Note: heavySkipNote}, nil + } + + maxN := st.maxDraftPosts + if maxN <= 0 { + maxN = 5 + } + perPlatform, requested, clamped := resolveGenerateCount(in.Count, maxN) + + emit(st.onEvent, SSEEventDraftPostStarted, DraftPostStartedEventPayload{ + PlatformIDs: platformIDs, + Count: perPlatform, + }) + + // Forward the flow's nested events, namespaced, so drafts stream in live. + nested := draft_post.OnEventFunc(func(name draft_post.SSEEventKind, data any) { + switch name { + case draft_post.SSEEventStep: + emit(st.onEvent, SSEEventDraftPostStep, data) + case draft_post.SSEEventPost: + emit(st.onEvent, SSEEventDraftPostPost, data) + case draft_post.SSEEventWarning: + emit(st.onEvent, SSEEventDraftPostWarning, data) + } + }) + + // One flow call per platform, with a shared budget so the total across all + // platforms never exceeds the per-call cap (CON-207 §10). + var ( + total int + allDates []string + allWarn []string + ) + budget := maxN + for _, pid := range platformIDs { + if budget <= 0 { + clamped = true + break + } + n := perPlatform + if n > budget { + n = budget + clamped = true + } + resp, err := st.draftPost(ctx, draft_post.DraftPostRequest{ + CampaignID: st.campaignID, + PlatformID: pid, + PostType: in.PostType, + Count: n, + SourceMaterial: source, + Instruction: st.instruction, + WindowStart: windowStart, + WindowEnd: windowEnd, + PhaseID: phaseID, + // UsedAssetIDs is intentionally left empty in v1: which assets the prior + // research cited isn't tracked, and over-stamping would pollute + // asset-usage provenance (cf. CON-214). The "Source research" note on + // each post carries the exact source instead. + }, nested) + if err != nil { + return nil, err + } + total += len(resp.Posts) + for _, p := range resp.Posts { + if p.PublishDate != "" { + allDates = append(allDates, p.PublishDate) + } + } + allWarn = append(allWarn, resp.Warnings...) + budget -= len(resp.Posts) + } + + st.draftPostResult = &DraftPostResult{ + PostCount: total, + PlatformIDs: platformIDs, + PhaseID: phaseID, + Dates: allDates, + Warnings: allWarn, + } + return &DraftPostOutput{ + PostCount: total, + RequestedCount: requested, + Clamped: clamped, + PlatformIDs: platformIDs, + Platforms: platformNames, + PhaseID: phaseID, + PhaseName: phaseName, + Dates: allDates, + Warnings: allWarn, + }, nil +} + +// resolveDraftSource returns the source text to draft from: an explicit override +// when given, else the most recent research answer in the conversation (CON-207 +// FR2). The stored model message is the compact JSON envelope persistTurn writes +// (action + explanation), so the research lives in the "explanation" field — +// parse it rather than using the raw content. Only answers (action "answered" or +// empty) qualify, so an action-confirmation turn (e.g. a prior post_drafted) is +// never mistaken for research. +func resolveDraftSource(ctx context.Context, st *requestState, override string) (string, error) { + if s := strings.TrimSpace(override); s != "" { + return s, nil + } + if st.repos.Messages == nil { + return "", nil + } + msgs, err := st.repos.Messages.ListRecentByCampaignID(ctx, st.campaignID, 10) + if err != nil { + return "", fmt.Errorf("load conversation history: %w", err) + } + // Messages come back oldest-first; scan newest-first for the latest answer. + for i := len(msgs) - 1; i >= 0; i-- { + if msgs[i].Role != "model" { + continue + } + var env struct { + Action string `json:"action"` + Explanation string `json:"explanation"` + } + if err := json.Unmarshal([]byte(msgs[i].Content), &env); err != nil { + // Legacy/plain-text model message (not JSON-wrapped) — use it directly. + if s := strings.TrimSpace(msgs[i].Content); s != "" { + return s, nil + } + continue + } + if env.Action != "" && env.Action != "answered" { + continue // an action confirmation, not research + } + if s := strings.TrimSpace(env.Explanation); s != "" { + return s, nil + } + } + return "", nil +} + // softGenerateFailure turns a user-correctable generatePosts problem — a // non-target platform, an unknown phase, or a past/invalid publish window — into // a zero-post result the planner relays to the user, rather than a Go error. @@ -588,6 +833,14 @@ func softGenerateFailure(err error) (*GeneratePostsOutput, error) { return &GeneratePostsOutput{PostCount: 0, Warnings: []string{err.Error()}}, nil } +// softDraftFailure is the draftPost analogue of softGenerateFailure (CON-215): +// a user-correctable problem (non-target platform, unknown phase, past/invalid +// date) becomes a zero-post warning the planner relays, not a turn-aborting Go +// error. It leaves draftPostResult unset, so the turn stays conversational. +func softDraftFailure(err error) (*DraftPostOutput, error) { + return &DraftPostOutput{PostCount: 0, Warnings: []string{err.Error()}}, nil +} + // resolveGenerateCount maps the model-supplied count to the number of posts the // generatePosts tool will actually create. An explicit positive count is honored // exactly. A missing or non-positive count defaults to 1 — the safe minimum, so diff --git a/src/genkit/flows/campaign_assistant/types.go b/src/genkit/flows/campaign_assistant/types.go index 6c3e7bb..901e5a7 100644 --- a/src/genkit/flows/campaign_assistant/types.go +++ b/src/genkit/flows/campaign_assistant/types.go @@ -9,6 +9,7 @@ import ( "github.com/ogen-app/ogen/src/eventhub" "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/draft_post" "github.com/ogen-app/ogen/src/genkit/flows/enrich_brief" "github.com/ogen-app/ogen/src/repository" "github.com/ogen-app/ogen/src/usage" @@ -28,7 +29,7 @@ type CampaignAssistantRequest struct { // ran this turn. type CampaignAssistantResponse struct { Explanation string `json:"explanation" jsonschema:"description=Conversational reply to the user"` - Action string `json:"action" jsonschema:"description=answered for a grounded reply; content_plan_generated when runContentPlan ran; brief_enriched when enrichBrief ran; posts_generated when generatePosts ran; dates_updated when setCampaignDates ran; posts_redistributed when redistributePosts ran; brief_reviewed when checkBrief ran; posts_reviewed when checkPostsConsistency ran; declined when the request is out of scope,enum=answered,enum=content_plan_generated,enum=brief_enriched,enum=posts_generated,enum=dates_updated,enum=posts_redistributed,enum=brief_reviewed,enum=posts_reviewed,enum=declined"` + Action string `json:"action" jsonschema:"description=answered for a grounded reply; content_plan_generated when runContentPlan ran; brief_enriched when enrichBrief ran; posts_generated when generatePosts ran; post_drafted when draftPost ran; dates_updated when setCampaignDates ran; posts_redistributed when redistributePosts ran; brief_reviewed when checkBrief ran; posts_reviewed when checkPostsConsistency ran; declined when the request is out of scope,enum=answered,enum=content_plan_generated,enum=brief_enriched,enum=posts_generated,enum=post_drafted,enum=dates_updated,enum=posts_redistributed,enum=brief_reviewed,enum=posts_reviewed,enum=declined"` // ContentPlan is set by the server when the runContentPlan tool created // draft posts this turn. Action is then "content_plan_generated". ContentPlan *ContentPlanResult `json:"contentPlan,omitempty" jsonschema:"-"` @@ -38,6 +39,10 @@ type CampaignAssistantResponse struct { // GeneratedPosts is set by the server when the generatePosts tool added // targeted posts this turn. Action is then "posts_generated". GeneratedPosts *GeneratedPostsResult `json:"generatedPosts,omitempty" jsonschema:"-"` + // DraftedPosts is set by the server when the draftPost tool created + // content-first drafts from chat research this turn (CON-207). Action is then + // "post_drafted". + DraftedPosts *DraftPostResult `json:"draftedPosts,omitempty" jsonschema:"-"` // Dates is set by the server when setCampaignDates changed the campaign's // dates this turn. Action is then "dates_updated". Dates *DatesResult `json:"dates,omitempty" jsonschema:"-"` @@ -76,6 +81,18 @@ type GeneratedPostsResult struct { UsedAssets []AssetRef `json:"usedAssets,omitempty"` } +// DraftPostResult summarises a draftPost tool invocation (CON-207). +type DraftPostResult struct { + PostCount int `json:"postCount"` + PlatformIDs []string `json:"platformIds"` + PhaseID string `json:"phaseId"` + Dates []string `json:"dates,omitempty"` // actual publish dates of the created posts + Warnings []string `json:"warnings,omitempty"` + // UsedAssets lists the campaign assets that informed the drafts (CON-118); + // empty in v1 (provenance is carried by each post's Source research note). + UsedAssets []AssetRef `json:"usedAssets,omitempty"` +} + // ContentPlanResult summarises a runContentPlan tool invocation. type ContentPlanResult struct { PostCount int `json:"postCount"` @@ -150,6 +167,12 @@ type CampaignAssistantFlowConfig struct { // MaxGeneratePosts caps how many posts one generatePosts call may create // (CON-114). 0 falls back to 10. MaxGeneratePosts int + // DraftPost backs the draftPost tool (CON-207): rewrite chat research into + // extended content-first drafts. nil disables the tool. + DraftPost func(ctx context.Context, req draft_post.DraftPostRequest, onEvent draft_post.OnEventFunc) (*draft_post.DraftPostResponse, error) + // MaxDraftPosts caps how many posts one draftPost call may create (CON-207). + // 0 falls back to 5. + MaxDraftPosts int // CheckBrief / CheckPosts back the read-only consistency review tools // (CON-116). nil disables the corresponding tool. CheckBrief func(ctx context.Context, campaignID string, onEvent consistency.OnEventFunc) (*consistency.BriefReview, error) @@ -198,6 +221,12 @@ const ( SSEEventGeneratePostsWarning SSEEventKind = "generate_posts_warning" SSEEventGeneratePostsComplete SSEEventKind = "generate_posts_complete" + SSEEventDraftPostStarted SSEEventKind = "draft_post_started" + SSEEventDraftPostStep SSEEventKind = "draft_post_step" + SSEEventDraftPostPost SSEEventKind = "draft_post_post" + SSEEventDraftPostWarning SSEEventKind = "draft_post_warning" + SSEEventDraftPostComplete SSEEventKind = "draft_post_complete" + // SSEEventAssetsUsed reports which attached assets informed the generated // posts (CON-118); emitted by runContentPlan/generatePosts when non-empty. SSEEventAssetsUsed SSEEventKind = "assets_used" @@ -265,6 +294,18 @@ type GeneratePostsCompleteEventPayload struct { Warnings []string `json:"warnings,omitempty"` } +// DraftPostStartedEventPayload is emitted when the draftPost tool begins (CON-207). +type DraftPostStartedEventPayload struct { + PlatformIDs []string `json:"platformIds"` + Count int `json:"count"` +} + +// DraftPostCompleteEventPayload is emitted once content-first drafts are persisted. +type DraftPostCompleteEventPayload struct { + PostCount int `json:"postCount"` + Warnings []string `json:"warnings,omitempty"` +} + // AssetsUsedEventPayload lists the attached assets that informed a generation // (CON-118). type AssetsUsedEventPayload struct { diff --git a/src/genkit/flows/draft_post/flow.go b/src/genkit/flows/draft_post/flow.go new file mode 100644 index 0000000..b53c1fd --- /dev/null +++ b/src/genkit/flows/draft_post/flow.go @@ -0,0 +1,70 @@ +package draft_post + +import ( + "context" + "embed" + "fmt" + "text/template" + + "github.com/firebase/genkit/go/core" + "github.com/firebase/genkit/go/genkit" +) + +//go:embed prompts/draft_post.tmpl +var promptFS embed.FS + +// DraftPostFlow is the singleton Genkit flow. Set by InitDraftPost. It is +// registered for Dev-UI discovery; the SSE path uses the runner closure below +// instead so it can stream per-post events. +var DraftPostFlow *core.Flow[DraftPostRequest, *DraftPostResponse, struct{}] + +// draftPostRunner is the direct closure that threads an OnEventFunc for SSE +// streaming. Set by InitDraftPost. +var draftPostRunner func(ctx context.Context, req DraftPostRequest, onEvent OnEventFunc) (*DraftPostResponse, error) + +// InitDraftPost parses the prompt template and registers the draftPost Genkit +// flow. Must be called after the Genkit instance has been initialised with the +// Anthropic plugin. +func InitDraftPost(g *genkit.Genkit, cfg DraftPostFlowConfig, repos DraftPostRepos) error { + raw, err := promptFS.ReadFile("prompts/draft_post.tmpl") + if err != nil { + return fmt.Errorf("load draft_post.tmpl: %w", err) + } + tmpl, err := template.New("draft_post").Parse(string(raw)) + if err != nil { + return fmt.Errorf("parse draft_post.tmpl: %w", err) + } + cfg.systemTmpl = tmpl.Lookup("system") + cfg.contextTmpl = tmpl.Lookup("context") + if cfg.systemTmpl == nil || cfg.contextTmpl == nil { + return fmt.Errorf("draft_post.tmpl must define both {{define \"system\"}} and {{define \"context\"}} blocks") + } + + DraftPostFlow = genkit.DefineFlow(g, "draftPost", + func(ctx context.Context, req DraftPostRequest) (*DraftPostResponse, error) { + return runDraftPost(ctx, g, req, cfg, repos, nil) + }, + ) + + draftPostRunner = func(ctx context.Context, req DraftPostRequest, onEvent OnEventFunc) (*DraftPostResponse, error) { + return runDraftPost(ctx, g, req, cfg, repos, onEvent) + } + + return nil +} + +// NewDraftPostCallback returns a callback suitable for the campaign assistant +// tool. onEvent is forwarded to the flow for SSE streaming; pass nil for a +// silent, non-streaming call. +func NewDraftPostCallback() func(ctx context.Context, req DraftPostRequest, onEvent OnEventFunc) (*DraftPostResponse, error) { + return func(ctx context.Context, req DraftPostRequest, onEvent OnEventFunc) (*DraftPostResponse, error) { + return draftPostRunner(ctx, req, onEvent) + } +} + +// emit calls onEvent when it is non-nil. It is a safe no-op otherwise. +func emit(onEvent OnEventFunc, name SSEEventKind, data any) { + if onEvent != nil { + onEvent(name, data) + } +} diff --git a/src/genkit/flows/draft_post/prompts/draft_post.tmpl b/src/genkit/flows/draft_post/prompts/draft_post.tmpl new file mode 100644 index 0000000..1f3d7e3 --- /dev/null +++ b/src/genkit/flows/draft_post/prompts/draft_post.tmpl @@ -0,0 +1,51 @@ +{{define "system"}}You are an expert social-media copywriter. Rewrite the RESEARCH provided in the context into finished, ready-to-publish posts for a single platform. + +## Your task +Produce exactly {{.Count}} distinct {{.PlatformName}} post{{if gt .Count 1}}s{{end}}. Each must be polished, platform-native copy that a person could publish as-is — NOT an outline, NOT a bullet-point thesis, NOT a summary of the research. + +## Rules +- Write in the campaign's output language: {{if .Language}}{{.Language}}{{else}}English{{end}}. Do not translate the JSON field names. +- Ground every post in the RESEARCH shown in the context. Do not invent facts, figures, statistics, or claims the research and brief do not support. +- Match the campaign's tone guidelines and speak to its target persona. +- Write native {{.PlatformName}} copy — respect the platform's conventions and stay within its character limit.{{if .Constraints}} Platform guidance: {{.Constraints}}{{end}} +{{- if gt .Count 1}} +- Make the {{.Count}} posts genuinely distinct: different angles, hooks, or facets of the research — never restatements of one another. +{{- end}} +{{- if .Instruction}} +- Additional steering from the user: {{.Instruction}} +{{- end}} + +## Response format +Respond with a raw JSON array only — no markdown fences, no prose before or after: +[ + { + "title": "short internal title for the post", + "content": "the finished, ready-to-publish post copy" + } +] +The array must contain exactly {{.Count}} object{{if gt .Count 1}}s{{end}}.{{end}} + +{{define "context"}}## Campaign +Name: {{.CampaignName}} +{{if .CampaignTypeLabel}}Type: {{.CampaignTypeLabel}} +{{end}}{{if .PhaseName}}Current phase: {{.PhaseName}} +{{end}} +## Brief +Description: {{.Description}} +Target persona: {{.TargetPersona}} +Key messages: {{.KeyMessages}} +Tone guidelines: {{.ToneGuidelines}} + +## Target platform +{{.PlatformName}}{{if .PostType}} — post type: {{.PostType}}{{end}} +{{- if .Constraints}} +Guidance: {{.Constraints}} +{{- end}} + +## Research to turn into {{.Count}} post{{if gt .Count 1}}s{{end}} +{{.SourceMaterial}} +{{- if .Instruction}} + +## Additional instruction +{{.Instruction}} +{{- end}}{{end}} diff --git a/src/genkit/flows/draft_post/run.go b/src/genkit/flows/draft_post/run.go new file mode 100644 index 0000000..9b55984 --- /dev/null +++ b/src/genkit/flows/draft_post/run.go @@ -0,0 +1,488 @@ +package draft_post + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log/slog" + "sort" + "strings" + "text/template" + "time" + + "github.com/firebase/genkit/go/ai" + "github.com/firebase/genkit/go/genkit" + + "github.com/ogen-app/ogen/src/logging" + "github.com/ogen-app/ogen/src/models" + "github.com/ogen-app/ogen/src/notes" + "github.com/ogen-app/ogen/src/repository" + "github.com/ogen-app/ogen/src/scheduling" + "github.com/ogen-app/ogen/src/settings" + "github.com/ogen-app/ogen/src/vendors/llm" +) + +// defaultMaxOutputTokens caps a single draft generation when the config leaves +// it at 0. Enough for a handful of full-length posts. +const defaultMaxOutputTokens int64 = 8192 + +// contextTemplateData is the view model passed to both prompt blocks. +type contextTemplateData struct { + CampaignName string + CampaignTypeLabel string + PhaseName string + Description string + TargetPersona string + KeyMessages string + ToneGuidelines string + Language string + PlatformName string + PostType string + Constraints string + Count int + SourceMaterial string + Instruction string +} + +// resolvedPlatform is the platform metadata the flow needs: the persisted +// post-type slug and the CON-91 character-limit / format guidance fed to the +// prompt. +type resolvedPlatform struct { + ID string + Name string + PostType string // resolved slug to persist (may be "") + Constraints string // CON-91 character limits, format notes +} + +func runDraftPost( + ctx context.Context, + g *genkit.Genkit, + req DraftPostRequest, + cfg DraftPostFlowConfig, + repos DraftPostRepos, + onEvent OnEventFunc, +) (*DraftPostResponse, error) { + start := time.Now() + slog.InfoContext(ctx, "starting", logging.AttrComponent, "genkit.draft_post", + "campaign_id", req.CampaignID, "platform_id", req.PlatformID, "count", req.Count, + "source_len", len(req.SourceMaterial), "instruction_len", len(req.Instruction)) + + // Enforcement gate (CON-86 FR9): block before any provider call when the + // tenant is already over a cap. Nil checker = no gate. + if err := cfg.Checker.Enforce(ctx); err != nil { + return nil, err + } + + // ── Validate ───────────────────────────────────────────────────────────── + if req.CampaignID == "" { + return nil, &ValidationError{Msg: "campaign id is required"} + } + if req.PlatformID == "" { + return nil, &ValidationError{Msg: "platform id is required"} + } + source := strings.TrimSpace(req.SourceMaterial) + if source == "" { + return nil, &ValidationError{Msg: "source material is required to draft a post"} + } + count := req.Count + if count <= 0 { + count = 1 + } + + // ── Load campaign (tenant-scoped) ──────────────────────────────────────── + campaign, err := repos.Campaigns.GetByID(ctx, req.CampaignID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, &ValidationError{Msg: "campaign not found"} + } + return nil, fmt.Errorf("load campaign: %w", err) + } + + // ── Resolve platform (+ post type + CON-91 constraints) ────────────────── + platform, err := resolvePlatform(ctx, campaign, req.PlatformID, req.PostType, repos.Platforms) + if err != nil { + return nil, err + } + + // ── Resolve window + per-post dates ────────────────────────────────────── + winStart, err := time.Parse("2006-01-02", req.WindowStart) + if err != nil { + return nil, &ValidationError{Msg: "windowStart must be an ISO date (YYYY-MM-DD)"} + } + winEnd, err := time.Parse("2006-01-02", req.WindowEnd) + if err != nil { + return nil, &ValidationError{Msg: "windowEnd must be an ISO date (YYYY-MM-DD)"} + } + if winEnd.Before(winStart) { + return nil, &ValidationError{Msg: "windowEnd must be on or after windowStart"} + } + dates := spreadDates(winStart, winEnd, count) + + // ── Build prompts ──────────────────────────────────────────────────────── + typeLabel := "" + if campaign.CampaignType != nil { + typeLabel = campaign.CampaignType.Label + } + data := contextTemplateData{ + CampaignName: campaign.Name, + CampaignTypeLabel: typeLabel, + PhaseName: phaseNameByID(campaign, req.PhaseID), + Description: campaign.Description, + TargetPersona: campaign.TargetPersona, + KeyMessages: campaign.KeyMessages, + ToneGuidelines: campaign.ToneGuidelines, + Language: campaign.Language, + PlatformName: platform.Name, + PostType: platform.PostType, + Constraints: platform.Constraints, + Count: count, + SourceMaterial: source, + Instruction: strings.TrimSpace(req.Instruction), + } + systemPrompt, err := renderTemplate(cfg.systemTmpl, data) + if err != nil { + return nil, fmt.Errorf("render system prompt: %w", err) + } + contextBlock, err := renderTemplate(cfg.contextTmpl, data) + if err != nil { + return nil, fmt.Errorf("render context block: %w", err) + } + emit(onEvent, SSEEventStep, StepEventPayload{Step: "buildContext", Status: "done"}) + + maxTokens := cfg.MaxOutputTokens + if maxTokens == 0 { + maxTokens = defaultMaxOutputTokens + } + modelName := cfg.Provider.Ref(llm.RoleGeneration) + modelCfg := cfg.Provider.CallConfig(maxTokens) + usageVendor := cfg.Provider.Vendor() + usageModel := cfg.Provider.Model(llm.RoleGeneration) + + loc, _ := settings.ResolveTimezone(campaign.Timezone) + + var out []DraftedPost + var warnings []string + + // appendDraft persists one finished draft content-first (CON-207) and emits + // its post event, capped at the requested count so an over-producing model + // can't inflate the result. + appendDraft := func(d modelDraft) { + if len(out) >= count { + return + } + title := strings.TrimSpace(d.Title) + content := strings.TrimSpace(d.Content) + if content == "" { + warnings = append(warnings, "dropped a draft with empty content") + emit(onEvent, SSEEventWarning, WarningPayload{Message: "dropped a draft with empty content"}) + return + } + publishDate := dates[len(out)] // len(out) < count == len(dates) + dp, id, perr := persistDraft(ctx, campaign, platform, req.PhaseID, publishDate, loc, &winStart, &winEnd, title, content, req.UsedAssetIDs, source, repos) + if perr != nil { + slog.ErrorContext(ctx, "persist draft failed", logging.AttrComponent, "genkit.draft_post", "title", title, logging.AttrError, perr) + msg := fmt.Sprintf("draft %q could not be saved: %v", title, perr) + warnings = append(warnings, msg) + emit(onEvent, SSEEventWarning, WarningPayload{Message: msg}) + return + } + emit(onEvent, SSEEventPost, PostEventPayload{Post: dp, Index: len(out), ID: id}) + out = append(out, dp) + } + + // ── Stream generation ──────────────────────────────────────────────────── + // No ai.WithOutputType: genkit's strict validator drops the whole response on + // common Claude JSON drift. We scan complete objects out of the stream and + // parse each tolerantly instead (mirrors content_plan). + scanner := newJSONObjScanner() + parsedCount := 0 + var streamErr error + for result, err := range genkit.GenerateStream(ctx, g, + ai.WithModelName(modelName), + ai.WithSystem(systemPrompt), + ai.WithPrompt(contextBlock), + modelCfg, + ) { + if err != nil { + streamErr = err + break + } + if result.Done { + if result.Response != nil { + if result.Response.Usage != nil { + u := result.Response.Usage + slog.InfoContext(ctx, "tokens", logging.AttrComponent, "genkit.draft_post", "input", u.InputTokens, "output", u.OutputTokens, "total", u.InputTokens+u.OutputTokens) + } + cfg.Recorder.RecordResp(ctx, usageVendor, usageModel, "draft_post", result.Response) + } + break + } + for _, raw := range scanner.push(result.Chunk.Text()) { + parsedCount++ + var d modelDraft + if uerr := json.Unmarshal([]byte(raw), &d); uerr != nil { + slog.WarnContext(ctx, "malformed draft chunk", logging.AttrComponent, "genkit.draft_post", "raw_preview", logging.Preview(raw, 120)) + continue + } + appendDraft(d) + } + } + + // ── Blocking fallback on stream failure ────────────────────────────────── + // Recover the rest of the batch with a single blocking call, skipping the + // objects already seen during streaming so we never double-insert. + if streamErr != nil { + slog.WarnContext(ctx, "stream error, falling back to blocking Generate", logging.AttrComponent, "genkit.draft_post", "persisted", len(out), "parsed", parsedCount, logging.AttrError, streamErr) + resp, gerr := genkit.Generate(ctx, g, + ai.WithModelName(modelName), + ai.WithSystem(systemPrompt), + ai.WithPrompt(contextBlock), + modelCfg, + ) + if gerr != nil { + if len(out) == 0 { + return nil, &AIError{Msg: fmt.Sprintf("model call failed (stream+fallback): %v", gerr)} + } + warnings = append(warnings, fmt.Sprintf("generation was cut short: %v", gerr)) + } else { + if resp.Usage != nil { + slog.InfoContext(ctx, "tokens (fallback)", logging.AttrComponent, "genkit.draft_post", "input", resp.Usage.InputTokens, "output", resp.Usage.OutputTokens, "total", resp.Usage.InputTokens+resp.Usage.OutputTokens) + } + cfg.Recorder.RecordResp(ctx, usageVendor, usageModel, "draft_post", resp) + var arr []modelDraft + if uerr := json.Unmarshal([]byte(stripFences(resp.Text())), &arr); uerr != nil { + if len(out) == 0 { + return nil, &AIError{Msg: fmt.Sprintf("model response not valid JSON: %v", uerr)} + } + } else { + for i, d := range arr { + if i < parsedCount { + continue // already handled during streaming + } + appendDraft(d) + } + } + } + } + emit(onEvent, SSEEventStep, StepEventPayload{Step: "generate", Status: "done"}) + + slog.InfoContext(ctx, "done", logging.AttrComponent, "genkit.draft_post", + "campaign_id", req.CampaignID, "duration_ms", time.Since(start).Milliseconds(), + "posts", len(out), "warnings", len(warnings)) + + // An empty result is a soft failure (0 posts + warnings), not an error — the + // assistant runner surfaces a friendly reply rather than a 502 (CON-207 §9). + return &DraftPostResponse{Posts: out, Warnings: warnings}, nil +} + +// persistDraft inserts one finished draft as a content-first Post row (CON-207): +// the generated copy goes into Post.Content, status=draft, scheduled per CON-181, +// and the source research is kept as a "Source research" reference note. Returns +// the DraftedPost view + the new row id. +func persistDraft( + ctx context.Context, + campaign *models.Campaign, + platform resolvedPlatform, + phaseID, publishDate string, + loc *time.Location, + windowStart, windowEnd *time.Time, + title, content string, + usedAssetIDs []string, + source string, + repos DraftPostRepos, +) (DraftedPost, string, error) { + id, err := models.NewID() + if err != nil { + return DraftedPost{}, "", err + } + + // CON-181: snap the publish date to an enabled publishing day, place it at + // the campaign's publishing time in its timezone, ± deterministic spread — + // bounded by the generation window. + scheduledAt, effDate, noEnabledDay := scheduling.ComposeScheduledAt( + publishDate, id, loc, campaign.PublishingTime, campaign.PublishingDays, + campaign.SpreadMinutes, windowStart, windowEnd, + ) + if noEnabledDay { + slog.WarnContext(ctx, "no enabled publishing day in window; kept model date", + logging.AttrComponent, "genkit.draft_post", "post_id", id, "date", publishDate) + } + + var phaseIDPtr *string + if phaseID != "" { + phaseIDPtr = &phaseID + } + + row := &models.Post{ + ID: id, + CampaignID: campaign.ID, + PlatformID: platform.ID, + PlatformPostType: platform.PostType, + Title: title, + Content: content, // CON-207: content-first — the finished copy, not "" + MediaURLs: models.StringSlice{}, + Status: models.PostStatusDraft, + CTAType: models.CTATypeNone, + CTAUrl: "", + UsedAssetIDs: models.StringSlice(usedAssetIDs), + CampaignTypePhaseID: phaseIDPtr, + ScheduledAt: scheduledAt, + CreatedBy: campaign.CreatedBy, + } + if err := repos.Posts.Create(ctx, row); err != nil { + return DraftedPost{}, "", err + } + + // CON-207: keep the source research as a reference note. Best-effort — a + // note-write failure must never discard the persisted post (CON-66/CON-188). + if repos.Notes != nil { + if body := strings.TrimSpace(source); body != "" { + if err := createSourceNote(ctx, repos.Notes, id, campaign.CreatedBy, body); err != nil { + slog.ErrorContext(ctx, "source research note create failed", logging.AttrComponent, "genkit.draft_post", "post_id", id, logging.AttrError, err) + } + } + } + + return DraftedPost{ + Title: title, + Content: content, + PlatformID: platform.ID, + PostType: platform.PostType, + PublishDate: effDate, + PostID: id, + }, id, nil +} + +// createSourceNote persists the source research as a free-form reference note +// (type note, origin assistant), title "Source research". The body is trimmed +// to the note service's max length. +func createSourceNote(ctx context.Context, noteRepo repository.PostNoteRepository, postID, createdBy, body string) error { + if r := []rune(body); len(r) > notes.MaxBodyLen { + body = string(r[:notes.MaxBodyLen]) + } + noteID, err := models.NewID() + if err != nil { + return err + } + now := time.Now().UTC() + return noteRepo.Create(ctx, &models.PostNote{ + ID: noteID, + PostID: postID, + Type: models.PostNoteTypeNote, + Title: "Source research", + Body: body, + Origin: models.PostNoteOriginAssistant, + CreatedBy: createdBy, + CreatedAt: now, + UpdatedAt: now, + }) +} + +// resolvePlatform looks up the platform's metadata (name + CON-91 constraints) +// and resolves the post-type slug to persist: an explicit slug when given, else +// the campaign's first selected post type for this platform, else the platform's +// first available slug. +func resolvePlatform(ctx context.Context, campaign *models.Campaign, platformID, postType string, platformRepo repository.PlatformRepository) (resolvedPlatform, error) { + all, err := platformRepo.List(ctx) + if err != nil { + return resolvedPlatform{}, fmt.Errorf("list platforms: %w", err) + } + var p *models.Platform + for i := range all { + if all[i].ID == platformID { + p = &all[i] + break + } + } + if p == nil { + return resolvedPlatform{}, &ValidationError{Msg: fmt.Sprintf("platform %q is not a known platform", platformID)} + } + + // Prefer the campaign's selected post types for this platform (deterministic + // order), then fall back to the platform's own available slugs. + var campaignSlugs []string + for _, tp := range campaign.TargetPlatforms { + if tp.ID == platformID { + campaignSlugs = append(campaignSlugs, tp.PostTypes...) + } + } + resolvedType := strings.TrimSpace(postType) + if resolvedType == "" { + if len(campaignSlugs) > 0 { + resolvedType = campaignSlugs[0] + } else if len(p.PostTypes) > 0 { + // Map iteration order is randomised, so pick the lexicographically first + // available slug for a deterministic default across identical requests. + slugs := make([]string, 0, len(p.PostTypes)) + for slug := range p.PostTypes { + slugs = append(slugs, slug) + } + sort.Strings(slugs) + resolvedType = slugs[0] + } + } + return resolvedPlatform{ID: p.ID, Name: p.Name, PostType: resolvedType, Constraints: p.Constraints}, nil +} + +// phaseNameByID returns the campaign phase's name for the prompt, or "" when the +// id is unknown (the tool already validated it; the prompt just tolerates a miss). +func phaseNameByID(campaign *models.Campaign, phaseID string) string { + if phaseID == "" || campaign.CampaignType == nil { + return "" + } + for _, ph := range campaign.CampaignType.Phases { + if ph.ID == phaseID { + return ph.Name + } + } + return "" +} + +// spreadDates returns n ISO dates evenly distributed across [start, end] +// inclusive. n==1 pins the single date to start; otherwise the first lands on +// start and the last on end. ComposeScheduledAt later snaps each to an enabled +// publishing weekday within the same window. +func spreadDates(start, end time.Time, n int) []string { + const iso = "2006-01-02" + out := make([]string, 0, n) + if n <= 1 { + return append(out, start.Format(iso)) + } + totalDays := int(end.Sub(start).Hours() / 24) + if totalDays < 0 { + totalDays = 0 + } + for i := 0; i < n; i++ { + off := 0 + if totalDays > 0 { + off = int(int64(i) * int64(totalDays) / int64(n-1)) + } + out = append(out, start.AddDate(0, 0, off).Format(iso)) + } + return out +} + +// stripFences removes a leading ```lang fence and trailing ``` from a blocking +// model response, so a fenced JSON array still parses (mirrors content_plan). +func stripFences(s string) string { + text := strings.TrimSpace(s) + if strings.HasPrefix(text, "```") { + if i := strings.Index(text, "\n"); i >= 0 { + text = text[i+1:] + } + text = strings.TrimSuffix(strings.TrimSpace(text), "```") + text = strings.TrimSpace(text) + } + return text +} + +func renderTemplate(tmpl *template.Template, data any) (string, error) { + var buf bytes.Buffer + if err := tmpl.Execute(&buf, data); err != nil { + return "", err + } + return buf.String(), nil +} diff --git a/src/genkit/flows/draft_post/run_test.go b/src/genkit/flows/draft_post/run_test.go new file mode 100644 index 0000000..fd2c6db --- /dev/null +++ b/src/genkit/flows/draft_post/run_test.go @@ -0,0 +1,140 @@ +package draft_post + +import ( + "context" + "reflect" + "testing" + "time" + + "github.com/ogen-app/ogen/src/models" + "github.com/ogen-app/ogen/src/repository" +) + +func day(s string) time.Time { + t, err := time.Parse("2006-01-02", s) + if err != nil { + panic(err) + } + return t +} + +// spreadDates distributes N posts across [start, end]: n==1 pins to start; +// otherwise the first lands on start and the last on end, evenly spaced. +func TestSpreadDates(t *testing.T) { + cases := []struct { + name string + start, end string + n int + want []string + }{ + {"single pins to start", "2026-08-15", "2026-08-15", 1, []string{"2026-08-15"}}, + {"single ignores window end", "2026-08-15", "2026-08-29", 1, []string{"2026-08-15"}}, + {"three across two weeks", "2026-08-01", "2026-08-15", 3, []string{"2026-08-01", "2026-08-08", "2026-08-15"}}, + {"two endpoints", "2026-08-01", "2026-08-11", 2, []string{"2026-08-01", "2026-08-11"}}, + {"same-day multi collapses", "2026-08-20", "2026-08-20", 3, []string{"2026-08-20", "2026-08-20", "2026-08-20"}}, + } + for _, c := range cases { + got := spreadDates(day(c.start), day(c.end), c.n) + if !reflect.DeepEqual(got, c.want) { + t.Errorf("%s: spreadDates(%s, %s, %d) = %v, want %v", c.name, c.start, c.end, c.n, got, c.want) + } + } +} + +// The streaming scanner must yield each top-level object exactly once, tolerate +// split chunks, and not be fooled by braces or brackets inside strings. +func TestJSONObjScanner(t *testing.T) { + s := newJSONObjScanner() + var got []string + // Feed the array in awkward chunk boundaries. + feed := []string{ + `[{"title":"a","content":"hello `, + `{world}"},`, + `{"title":"b",`, + `"content":"line1\nline2"}]`, + } + for _, f := range feed { + got = append(got, s.push(f)...) + } + want := []string{ + `{"title":"a","content":"hello {world}"}`, + `{"title":"b","content":"line1\nline2"}`, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("scanner objects = %v, want %v", got, want) + } +} + +func TestStripFences(t *testing.T) { + cases := []struct{ in, want string }{ + {"[{\"a\":1}]", `[{"a":1}]`}, + {"```json\n[{\"a\":1}]\n```", `[{"a":1}]`}, + {"```\n[1,2]\n```", `[1,2]`}, + {" [1] ", `[1]`}, + } + for _, c := range cases { + if got := stripFences(c.in); got != c.want { + t.Errorf("stripFences(%q) = %q, want %q", c.in, got, c.want) + } + } +} + +// fakePlatformRepo is a minimal PlatformRepository returning a fixed list. +type fakePlatformRepo struct { + repository.PlatformRepository + platforms []models.Platform +} + +func (f *fakePlatformRepo) List(context.Context) ([]models.Platform, error) { + return f.platforms, nil +} + +// resolvePlatform resolves the post-type slug: explicit wins, else the +// campaign's first selected type, else a platform slug; and it carries the +// platform's free-text constraints for the prompt. +func TestResolvePlatform(t *testing.T) { + repo := &fakePlatformRepo{platforms: []models.Platform{ + {ID: "li", Name: "LinkedIn", PostTypes: models.PostTypeMap{"article": "Article"}, Constraints: "3000 chars max"}, + }} + campaign := &models.Campaign{ + TargetPlatforms: models.CampaignPlatforms{ + {ID: "li", PostTypes: []string{"text-post", "article"}}, + }, + } + + // Explicit post type wins. + got, err := resolvePlatform(context.Background(), campaign, "li", "article", repo) + if err != nil { + t.Fatalf("explicit: %v", err) + } + if got.Name != "LinkedIn" || got.PostType != "article" || got.Constraints != "3000 chars max" { + t.Fatalf("explicit = %+v", got) + } + + // No explicit type → the campaign's first selected slug for this platform. + got, err = resolvePlatform(context.Background(), campaign, "li", "", repo) + if err != nil { + t.Fatalf("default: %v", err) + } + if got.PostType != "text-post" { + t.Fatalf("default post type = %q, want text-post", got.PostType) + } + + // No explicit type and no campaign-selected slugs → the lexicographically + // first platform slug (deterministic, not map-iteration order). + multi := &fakePlatformRepo{platforms: []models.Platform{ + {ID: "x", Name: "X", PostTypes: models.PostTypeMap{"zeta": "Z", "alpha": "A", "mid": "M"}}, + }} + got, err = resolvePlatform(context.Background(), &models.Campaign{}, "x", "", multi) + if err != nil { + t.Fatalf("map fallback: %v", err) + } + if got.PostType != "alpha" { + t.Fatalf("map fallback post type = %q, want alpha (lexicographically first)", got.PostType) + } + + // Unknown platform id → error. + if _, err := resolvePlatform(context.Background(), campaign, "nope", "", repo); err == nil { + t.Fatal("expected error for unknown platform id") + } +} diff --git a/src/genkit/flows/draft_post/scanner.go b/src/genkit/flows/draft_post/scanner.go new file mode 100644 index 0000000..1e170b4 --- /dev/null +++ b/src/genkit/flows/draft_post/scanner.go @@ -0,0 +1,58 @@ +package draft_post + +// jsonObjScanner incrementally scans a stream of JSON text and yields complete +// top-level JSON objects (each element of the outer array) as they arrive. It is +// a copy of content_plan's private post scanner: the draft flow streams the same +// array-of-objects shape but stays decoupled from content_plan's internals. +type jsonObjScanner struct { + buf []byte + depth int + inStr bool + escaped bool + objStart int // index of the opening '{' of the current object; -1 when not inside one +} + +func newJSONObjScanner() *jsonObjScanner { + return &jsonObjScanner{objStart: -1} +} + +// push appends chunk to the internal buffer and returns all newly-complete JSON +// object strings found since the last call. +func (s *jsonObjScanner) push(chunk string) []string { + var complete []string + for i := 0; i < len(chunk); i++ { + c := chunk[i] + s.buf = append(s.buf, c) + pos := len(s.buf) - 1 + + if s.escaped { + s.escaped = false + continue + } + if s.inStr { + switch c { + case '\\': + s.escaped = true + case '"': + s.inStr = false + } + continue + } + switch c { + case '"': + s.inStr = true + case '{': + if s.depth == 0 { + s.objStart = pos + } + s.depth++ + case '}': + s.depth-- + if s.depth == 0 && s.objStart >= 0 { + complete = append(complete, string(s.buf[s.objStart:pos+1])) + s.objStart = -1 + } + } + } + return complete +} diff --git a/src/genkit/flows/draft_post/types.go b/src/genkit/flows/draft_post/types.go new file mode 100644 index 0000000..4b3ba1b --- /dev/null +++ b/src/genkit/flows/draft_post/types.go @@ -0,0 +1,141 @@ +package draft_post + +import ( + "text/template" + + "github.com/ogen-app/ogen/src/repository" + "github.com/ogen-app/ogen/src/usage" + "github.com/ogen-app/ogen/src/vendors/llm" +) + +// DraftPostRequest is the input to the draftPost generation flow (CON-207): +// rewrite SourceMaterial into Count finished, platform-ready post drafts for a +// single platform, phase, and publish window. All fields are concrete — the +// campaign_assistant draftPost tool resolves any natural language (platform, +// phase, count, date) before calling, so the flow stays deterministic. +type DraftPostRequest struct { + CampaignID string `json:"campaignId"` + PlatformID string `json:"platformId"` + PostType string `json:"postType"` // optional post-type slug; empty = platform/campaign default + Count int `json:"count"` // number of drafts to produce (>=1) + SourceMaterial string `json:"sourceMaterial"` // research/source text to rewrite into posts + Instruction string `json:"instruction"` // optional user steering ("punchier", "for execs") + WindowStart string `json:"windowStart"` // ISO YYYY-MM-DD (inclusive) + WindowEnd string `json:"windowEnd"` // ISO YYYY-MM-DD (inclusive) + PhaseID string `json:"phaseId"` // a single campaign phase id + UsedAssetIDs []string `json:"usedAssetIds"` // asset ids stamped on each created post for provenance (may be empty) +} + +// DraftedPost is one finished post draft produced + persisted by the flow. +type DraftedPost struct { + Title string `json:"title"` + Content string `json:"content"` + PlatformID string `json:"platformId"` + PostType string `json:"postType"` + PublishDate string `json:"publishDate"` + PostID string `json:"postId"` +} + +// DraftPostResponse is returned by the flow. Posts is the persisted set — every +// element already has a real Post row (CON-66). An empty Posts with warnings is +// a soft failure (the model produced nothing usable), not an error. +type DraftPostResponse struct { + Posts []DraftedPost `json:"posts"` + Warnings []string `json:"warnings,omitempty"` +} + +// modelDraft is the per-post schema the model emits: just a title and the +// finished copy. The platform, post type, and publish date are decided +// server-side (already resolved by the tool), so the model focuses only on +// writing native, publish-ready copy from the research. +type modelDraft struct { + Title string `json:"title" jsonschema:"description=Short descriptive title for the post"` + Content string `json:"content" jsonschema:"description=The finished, platform-ready post copy — publish-ready, not an outline or bullet list"` +} + +// DraftPostRepos bundles the repository dependencies for the flow. +type DraftPostRepos struct { + Campaigns repository.CampaignRepository + Platforms repository.PlatformRepository + Posts repository.PostRepository + // Notes stores the source research as a reference note on each created post + // (CON-207). nil skips note creation (the post is still created). + Notes repository.PostNoteRepository +} + +// DraftPostFlowConfig holds static settings for the flow. Unlike content_plan it +// has no Hub: draftPost is only reached through the campaign assistant, whose +// runner already publishes the coarse assistant_completed finalisation event +// (CON-112), so a second per-flow finalisation would be redundant. +type DraftPostFlowConfig struct { + // Provider resolves the model reference + call config by role (CON-86 FR12). + // Draft copywriting runs on RoleGeneration (Sonnet-tier), like content_plan. + Provider *llm.Provider + // Recorder captures usage under flow name "draft_post"; nil disables it. + Recorder *usage.Recorder + // Checker gates the flow against the tenant's spend caps; nil = no gate. + Checker *usage.Checker + ModelID string + // MaxOutputTokens caps a single generation call. 0 falls back to 8192 — + // enough for a handful of full-length drafts. + MaxOutputTokens int64 + + systemTmpl *template.Template + contextTmpl *template.Template +} + +// ValidationError is returned when preconditions are not met (HTTP 400). +type ValidationError struct{ Msg string } + +func (e *ValidationError) Error() string { return e.Msg } + +// AIError is returned when the model call fails or returns nothing usable +// (HTTP 502). +type AIError struct{ Msg string } + +func (e *AIError) Error() string { return e.Msg } + +// ── SSE types ───────────────────────────────────────────────────────────────── + +// SSEEventKind identifies the SSE event types emitted by the flow. The +// campaign_assistant tool re-emits these namespaced (draft_post_*). +type SSEEventKind string + +const ( + SSEEventStep SSEEventKind = "step" + SSEEventPost SSEEventKind = "post" + SSEEventWarning SSEEventKind = "warning" + SSEEventComplete SSEEventKind = "complete" + SSEEventError SSEEventKind = "error" +) + +// StepEventPayload marks a flow stage as finished. +type StepEventPayload struct { + Step string `json:"step"` + Status string `json:"status"` // always "done" +} + +// PostEventPayload carries one finished draft with its persisted row id, so the +// client can render it immediately (CON-66). Index is the draft's slot in the +// returned order. +type PostEventPayload struct { + Post DraftedPost `json:"post"` + Index int `json:"index"` + ID string `json:"id"` +} + +// WarningPayload is emitted when a draft is dropped (empty copy) or fails to +// persist; the run continues. +type WarningPayload struct { + Message string `json:"message"` +} + +// ErrorEventPayload is emitted when the flow fails mid-stream. +type ErrorEventPayload struct { + Message string `json:"message"` + Code int `json:"code"` // HTTP semantic: 400, 402, 502, 500 +} + +// OnEventFunc is an optional callback invoked as SSE events are produced. A nil +// OnEventFunc is valid — the flow runs silently. +type OnEventFunc func(name SSEEventKind, data any) diff --git a/src/integration/campaign_assistant_test.go b/src/integration/campaign_assistant_test.go index ce22edc..84eeb43 100644 --- a/src/integration/campaign_assistant_test.go +++ b/src/integration/campaign_assistant_test.go @@ -17,6 +17,7 @@ import ( "github.com/ogen-app/ogen/src/campaign_actions/overview" "github.com/ogen-app/ogen/src/genkit/flows/campaign_assistant" "github.com/ogen-app/ogen/src/genkit/flows/content_plan" + "github.com/ogen-app/ogen/src/genkit/flows/draft_post" "github.com/ogen-app/ogen/src/genkit/flows/enrich_brief" "github.com/ogen-app/ogen/src/models" "github.com/ogen-app/ogen/src/repository" @@ -34,6 +35,7 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { campaignID string campaignRepo repository.CampaignRepository postRepo repository.PostRepository + postNoteRepo repository.PostNoteRepository messageRepo repository.CampaignAssistantMessageRepository assetRepo repository.AssetRepository callback func(ctx context.Context, req campaign_assistant.CampaignAssistantRequest, onEvent campaign_assistant.OnEventFunc) (*campaign_assistant.CampaignAssistantResponse, error) @@ -56,6 +58,7 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { campaignTypeRepo := repository.NewCampaignTypeRepository(db) campaignRepo = repository.NewCampaignRepository(db, tagRepo, platformRepo, campaignTypeRepo) postRepo = repository.NewPostRepository(db) + postNoteRepo = repository.NewPostNoteRepository(db) messageRepo = repository.NewCampaignAssistantMessageRepository(db) // Seed user. @@ -128,6 +131,15 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { enrichBriefCb := enrich_brief.NewEnrichBriefCallback() generatePostsCb := content_plan.NewGeneratePostsCallback() + // CON-207: register the draftPost generation flow as an assistant tool. + Expect(draft_post.InitDraftPost(g, draft_post.DraftPostFlowConfig{Provider: provider}, draft_post.DraftPostRepos{ + Campaigns: campaignRepo, + Platforms: platformRepo, + Posts: postRepo, + Notes: postNoteRepo, + })).To(Succeed()) + draftPostCb := draft_post.NewDraftPostCallback() + overviewSvc := overview.New(campaignRepo, postRepo, platformRepo) Expect(campaign_assistant.InitCampaignAssistant(g, campaign_assistant.CampaignAssistantFlowConfig{ @@ -137,6 +149,8 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { Overview: overviewSvc, GeneratePosts: generatePostsCb, MaxGeneratePosts: 10, + DraftPost: draftPostCb, + MaxDraftPosts: 5, }, campaign_assistant.CampaignAssistantRepos{ Messages: messageRepo, Campaigns: campaignRepo, @@ -148,8 +162,10 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { }) AfterEach(func() { - // Reset conversation + posts between specs; the campaign persists. + // Reset conversation + posts (and their notes) between specs; the campaign + // persists. Notes are deleted first — they FK the posts (CON-188/CON-207). _, _ = db.NewDelete().TableExpr("campaign_assistant_messages").Where("1 = 1").Exec(ctx) + _, _ = db.NewDelete().TableExpr("post_notes").Where("post_id IN (SELECT id FROM posts WHERE campaign_id = ?)", campaignID).Exec(ctx) _, _ = db.NewDelete().TableExpr("posts").Where("campaign_id = ?", campaignID).Exec(ctx) }) @@ -158,6 +174,7 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { return } _, _ = db.NewDelete().TableExpr("campaign_assistant_messages").Where("1 = 1").Exec(ctx) + _, _ = db.NewDelete().TableExpr("post_notes").Where("post_id IN (SELECT id FROM posts WHERE campaign_id = ?)", campaignID).Exec(ctx) _, _ = db.NewDelete().TableExpr("posts").Where("campaign_id = ?", campaignID).Exec(ctx) _, _ = db.NewDelete().TableExpr("campaigns").Where("id = ?", campaignID).Exec(ctx) _, _ = db.NewDelete().TableExpr("users").Where("id = ?", userID).Exec(ctx) @@ -313,6 +330,66 @@ var _ = Describe("Campaign assistant flow", Ordered, func() { }) }) + Describe("draft post from research", func() { + It("turns a prior research answer into a content-first draft with a Source research note", func() { + // Turn 1: a research-style answer that persists as the latest model turn. + _, err := callback(ctx, campaign_assistant.CampaignAssistantRequest{ + CampaignID: campaignID, + Instruction: "In two or three sentences, summarise why Go suits production AI features for this campaign.", + }, nil) + Expect(err).NotTo(HaveOccurred()) + + // Turn 2: turn that research into a real LinkedIn post. + var gotDraftComplete, gotComplete bool + onEvent := campaign_assistant.OnEventFunc(func(name campaign_assistant.SSEEventKind, _ any) { + switch name { + case campaign_assistant.SSEEventDraftPostComplete: + gotDraftComplete = true + case campaign_assistant.SSEEventComplete: + gotComplete = true + } + }) + resp, err := callback(ctx, campaign_assistant.CampaignAssistantRequest{ + CampaignID: campaignID, + Instruction: "Great — create a LinkedIn post with this info for next week.", + }, onEvent) + Expect(err).NotTo(HaveOccurred()) + Expect(resp).NotTo(BeNil()) + + Expect(resp.Action).To(Equal("post_drafted")) + Expect(resp.DraftedPosts).NotTo(BeNil()) + Expect(resp.DraftedPosts.PostCount).To(BeNumerically(">", 0)) + Expect(gotDraftComplete).To(BeTrue(), "the draft_post_complete event should be forwarded") + Expect(gotComplete).To(BeTrue(), "complete signals the canonical final response") + + posts, err := postRepo.ListByCampaign(ctx, campaignID) + Expect(err).NotTo(HaveOccurred()) + Expect(len(posts)).To(Equal(resp.DraftedPosts.PostCount)) + + p := posts[0] + // Content-first: the finished copy lives in the post body (not the empty + // content_plan body), status draft, on the requested platform. + Expect(strings.TrimSpace(p.Content)).NotTo(BeEmpty(), "draft should carry full copy in Content") + Expect(p.Status).To(Equal(models.PostStatusDraft)) + Expect(p.PlatformID).To(Equal(platformID)) + + // A "Source research" note (type note, origin assistant) — and NO + // draft_thesis note (that's the content_plan path, not this one). + notes, err := postNoteRepo.ListByPostID(ctx, p.ID) + Expect(err).NotTo(HaveOccurred()) + var sawSource bool + for _, n := range notes { + Expect(n.Type).NotTo(Equal(models.PostNoteTypeDraftThesis), "draftPost must not create a draft_thesis note") + if n.Type == models.PostNoteTypeNote && n.Title == "Source research" { + sawSource = true + Expect(n.Origin).To(Equal(models.PostNoteOriginAssistant)) + Expect(strings.TrimSpace(n.Body)).NotTo(BeEmpty()) + } + } + Expect(sawSource).To(BeTrue(), "the source research should be kept as a reference note") + }) + }) + Describe("change dates", func() { It("moves the campaign end date and saves it", func() { resp, err := callback(ctx, campaign_assistant.CampaignAssistantRequest{ diff --git a/src/server/campaign_assistant.go b/src/server/campaign_assistant.go index 5a7fb3b..7f69679 100644 --- a/src/server/campaign_assistant.go +++ b/src/server/campaign_assistant.go @@ -13,6 +13,7 @@ import ( "github.com/ogen-app/ogen/src/genkit/flows/campaign_assistant" "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/draft_post" "github.com/ogen-app/ogen/src/genkit/flows/enrich_brief" "github.com/ogen-app/ogen/src/usage" "github.com/ogen-app/ogen/src/vendors/llm" @@ -35,6 +36,7 @@ func initCampaignAssistant( enrichBriefFn func(ctx context.Context, req enrich_brief.EnrichBriefRequest, onEvent enrich_brief.OnEventFunc) (*enrich_brief.EnrichBriefResponse, error), overviewSvc *overview.Service, generatePostsFn func(ctx context.Context, req content_plan.GeneratePostsRequest, onEvent content_plan.OnEventFunc) (*content_plan.ContentPlanResponse, error), + draftPostFn func(ctx context.Context, req draft_post.DraftPostRequest, onEvent draft_post.OnEventFunc) (*draft_post.DraftPostResponse, error), checkBriefFn func(ctx context.Context, campaignID string, onEvent consistency.OnEventFunc) (*consistency.BriefReview, error), checkPostsFn func(ctx context.Context, req consistency.PostsCheckRequest, onEvent consistency.OnEventFunc) (*consistency.PostsReview, error), ) (func(ctx context.Context, req campaign_assistant.CampaignAssistantRequest, onEvent campaign_assistant.OnEventFunc) (*campaign_assistant.CampaignAssistantResponse, error), error) { @@ -67,6 +69,8 @@ func initCampaignAssistant( Overview: overviewSvc, GeneratePosts: generatePostsFn, MaxGeneratePosts: cfg.GeneratePostsMax, + DraftPost: draftPostFn, + MaxDraftPosts: cfg.DraftPostMax, CheckBrief: checkBriefFn, CheckPosts: checkPostsFn, } diff --git a/src/server/draft_post.go b/src/server/draft_post.go new file mode 100644 index 0000000..40e3ece --- /dev/null +++ b/src/server/draft_post.go @@ -0,0 +1,37 @@ +package server + +import ( + "context" + "fmt" + + "github.com/firebase/genkit/go/genkit" + + "github.com/ogen-app/ogen/src/config" + "github.com/ogen-app/ogen/src/genkit/flows/draft_post" + "github.com/ogen-app/ogen/src/usage" + "github.com/ogen-app/ogen/src/vendors/llm" +) + +// initDraftPost registers the draftPost generation flow (CON-207) on the shared +// Genkit instance and returns an SSE-capable callback for the campaign assistant +// tool. Copywriting runs on the generation role (Sonnet-tier, cfg.ModelID), like +// content_plan; MaxOutputTokens is left at 0 so the flow uses its own default. +func initDraftPost( + g *genkit.Genkit, + cfg *config.Config, + provider *llm.Provider, + recorder *usage.Recorder, + checker *usage.Checker, + repos draft_post.DraftPostRepos, +) (func(ctx context.Context, req draft_post.DraftPostRequest, onEvent draft_post.OnEventFunc) (*draft_post.DraftPostResponse, error), error) { + flowCfg := draft_post.DraftPostFlowConfig{ + Provider: provider, + Recorder: recorder, + Checker: checker, + ModelID: cfg.ModelID, + } + if err := draft_post.InitDraftPost(g, flowCfg, repos); err != nil { + return nil, fmt.Errorf("init draft post flow: %w", err) + } + return draft_post.NewDraftPostCallback(), nil +} diff --git a/src/server/genkit_runtime.go b/src/server/genkit_runtime.go index 85d842f..4519bab 100644 --- a/src/server/genkit_runtime.go +++ b/src/server/genkit_runtime.go @@ -17,6 +17,7 @@ import ( "github.com/ogen-app/ogen/src/genkit/flows/campaign_assistant" "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/draft_post" "github.com/ogen-app/ogen/src/genkit/flows/enrich_brief" "github.com/ogen-app/ogen/src/genkit/flows/post_assistant" "github.com/ogen-app/ogen/src/genkit/flows/post_quality" @@ -69,6 +70,7 @@ type genkitRuntime struct { postQualityRepos post_quality.PostQualityRepos enrichBriefRepos enrich_brief.EnrichBriefRepos campaignAssistRepos campaign_assistant.CampaignAssistantRepos + draftPostRepos draft_post.DraftPostRepos campaignOverviewSvc *overview.Service cloneSvc *clone.Service restoreSvc *restore.Service @@ -89,6 +91,7 @@ type genkitDeps struct { postQualityRepos post_quality.PostQualityRepos enrichBriefRepos enrich_brief.EnrichBriefRepos campaignAssistRepos campaign_assistant.CampaignAssistantRepos + draftPostRepos draft_post.DraftPostRepos campaignOverviewSvc *overview.Service cloneSvc *clone.Service restoreSvc *restore.Service @@ -114,6 +117,7 @@ func newGenkitRuntime(ctx context.Context, deps genkitDeps, store secrets.Store) postQualityRepos: deps.postQualityRepos, enrichBriefRepos: deps.enrichBriefRepos, campaignAssistRepos: deps.campaignAssistRepos, + draftPostRepos: deps.draftPostRepos, campaignOverviewSvc: deps.campaignOverviewSvc, cloneSvc: deps.cloneSvc, restoreSvc: deps.restoreSvc, @@ -314,15 +318,20 @@ func (r *genkitRuntime) rebuild(ctx context.Context, store secrets.Store) error // CON-114: the targeted generation callback shares the content_plan flow // (already registered by initContentPlan above). generatePostsFn := content_plan.NewGeneratePostsCallback() + // CON-207: the draftPost generation flow (asset research → content-first drafts). + draftPostFn, err := initDraftPost(g, r.cfg, provider, r.recorder, r.checker, r.draftPostRepos) + if err != nil { + return fmt.Errorf("init draft post: %w", err) + } // CON-116: read-only brief + posts consistency review. checkBriefFn, checkPostsFn, err := initConsistency(g, r.cfg, provider, r.recorder, r.checker, r.hub, r.campaignAssistRepos.Campaigns, r.campaignAssistRepos.Posts) if err != nil { return fmt.Errorf("init consistency: %w", err) } // CON-112: the campaign assistant reuses the content_plan + enrich_brief - // callbacks (plus the CON-114 generatePosts and CON-116 consistency - // callbacks) as tools, so it is initialised after them. - campaignAssistantFn, err := initCampaignAssistant(g, r.cfg, provider, r.recorder, r.checker, r.embedder, r.hub, r.campaignAssistRepos, contentPlanFn, enrichBriefFn, r.campaignOverviewSvc, generatePostsFn, checkBriefFn, checkPostsFn) + // callbacks (plus the CON-114 generatePosts, CON-207 draftPost, and CON-116 + // consistency callbacks) as tools, so it is initialised after them. + campaignAssistantFn, err := initCampaignAssistant(g, r.cfg, provider, r.recorder, r.checker, r.embedder, r.hub, r.campaignAssistRepos, contentPlanFn, enrichBriefFn, r.campaignOverviewSvc, generatePostsFn, draftPostFn, checkBriefFn, checkPostsFn) if err != nil { return fmt.Errorf("init campaign assistant: %w", err) } diff --git a/src/server/server.go b/src/server/server.go index 731476f..e9e12db 100644 --- a/src/server/server.go +++ b/src/server/server.go @@ -24,6 +24,7 @@ import ( "github.com/ogen-app/ogen/src/eventhub" "github.com/ogen-app/ogen/src/genkit/flows/campaign_assistant" "github.com/ogen-app/ogen/src/genkit/flows/content_plan" + "github.com/ogen-app/ogen/src/genkit/flows/draft_post" "github.com/ogen-app/ogen/src/genkit/flows/enrich_brief" "github.com/ogen-app/ogen/src/genkit/flows/post_assistant" "github.com/ogen-app/ogen/src/genkit/flows/post_quality" @@ -529,6 +530,12 @@ func New(ctx context.Context, db, analyticsDB *bun.DB, cfg *config.Config, secre Assets: pieceRepo, Chunks: chunksRepo, }, + draftPostRepos: draft_post.DraftPostRepos{ + Campaigns: campaignRepo, + Platforms: platformRepo, + Posts: postRepo, + Notes: postNoteRepo, + }, campaignOverviewSvc: campaignOverviewSvc, cloneSvc: cloneSvc, restoreSvc: restoreSvc,