diff --git a/Dockerfile b/Dockerfile index ab78a0da92..bf0a66e401 100644 --- a/Dockerfile +++ b/Dockerfile @@ -84,7 +84,7 @@ RUN mix do db.certs, agent.chart, sentry.package_source_code, release FROM alpine:3.21.3 as tools ARG TARGETARCH=amd64 -ENV CLI_VERSION=v0.12.59 +ENV CLI_VERSION=v0.12.60 COPY AGENT_VERSION AGENT_VERSION diff --git a/go/deployment-operator/pkg/agentrun-harness/controller/analysis_gate_test.go b/go/deployment-operator/pkg/agentrun-harness/controller/analysis_gate_test.go index 3d4faf6045..a44bebe65c 100644 --- a/go/deployment-operator/pkg/agentrun-harness/controller/analysis_gate_test.go +++ b/go/deployment-operator/pkg/agentrun-harness/controller/analysis_gate_test.go @@ -405,6 +405,88 @@ func TestBuildBabysitContextIncludesDelayedNewComments(t *testing.T) { require.Contains(t, bCtx.Prompt, "review:2") } +func TestBuildBabysitContextSkipsStateChangesWithoutCommentsOrFailingCI(t *testing.T) { + t.Parallel() + + prURL := "https://github.com/pluralsh/console/pull/1" + in := &agentRunController{dir: t.TempDir()} + agentRun := &gqlclient.AgentRunFragment{ + PullRequests: []*gqlclient.PullRequestFragment{{URL: prURL}}, + } + client := &fakeBabysitGRPCClient{ + details: map[string]*scm.PRDetails{ + prURL: { + Title: "Fix babysit", + HeadRef: "fix/babysit", + State: scm.PRStateOpen, + CIChecks: []scm.CICheck{{ + Name: "test", + Status: scm.CICheckStatusCompleted, + Conclusion: scm.CICheckConclusionSuccess, + }}, + }, + }, + } + + require.Nil(t, in.buildBabysitContext(context.Background(), agentRun, client)) + baselineSHA := in.lastPRSHA + + client.details[prURL] = &scm.PRDetails{ + Title: "Fix babysit docs", + HeadRef: "fix/babysit", + State: scm.PRStateOpen, + CIChecks: []scm.CICheck{{ + Name: "test", + Status: scm.CICheckStatusInProgress, + }}, + } + + require.Nil(t, in.buildBabysitContext(context.Background(), agentRun, client)) + require.NotEqual(t, baselineSHA, in.lastPRSHA) +} + +func TestBuildBabysitContextIncludesFailingCIWithoutComments(t *testing.T) { + t.Parallel() + + prURL := "https://github.com/pluralsh/console/pull/1" + in := &agentRunController{dir: t.TempDir()} + agentRun := &gqlclient.AgentRunFragment{ + PullRequests: []*gqlclient.PullRequestFragment{{URL: prURL}}, + } + client := &fakeBabysitGRPCClient{ + details: map[string]*scm.PRDetails{ + prURL: { + Title: "Fix babysit", + HeadRef: "fix/babysit", + State: scm.PRStateOpen, + CIChecks: []scm.CICheck{{ + Name: "test", + Status: scm.CICheckStatusCompleted, + Conclusion: scm.CICheckConclusionSuccess, + }}, + }, + }, + } + + require.Nil(t, in.buildBabysitContext(context.Background(), agentRun, client)) + + client.details[prURL] = &scm.PRDetails{ + Title: "Fix babysit", + HeadRef: "fix/babysit", + State: scm.PRStateOpen, + CIChecks: []scm.CICheck{{ + Name: "test", + Status: scm.CICheckStatusCompleted, + Conclusion: scm.CICheckConclusionFailure, + }}, + } + + bCtx := in.buildBabysitContext(context.Background(), agentRun, client) + require.NotNil(t, bCtx) + require.Contains(t, bCtx.Prompt, "Failing CI checks") + require.Contains(t, bCtx.Prompt, "test") +} + func TestBuildBabysitContextIncludesReviewSummariesAsNonReactable(t *testing.T) { t.Parallel() diff --git a/go/deployment-operator/pkg/agentrun-harness/controller/controller.go b/go/deployment-operator/pkg/agentrun-harness/controller/controller.go index 005d43f3a8..cc838bf19e 100644 --- a/go/deployment-operator/pkg/agentrun-harness/controller/controller.go +++ b/go/deployment-operator/pkg/agentrun-harness/controller/controller.go @@ -384,6 +384,13 @@ func (in *agentRunController) buildBabysitContext(ctx context.Context, agentRun return nil } + if !hasBabysitActivity(enriched) { + in.lastPRSHA = sha + in.lastPRCheckAt = time.Now() + klog.V(log.LogLevelExtended).InfoS("PR state changed without new comments or failing CI, skipping reprompt") + return nil + } + // Determine working branch from the live PR head ref. // Fall back to a generic placeholder if none of the PRs carry a head ref. branch := "your working branch" @@ -479,6 +486,30 @@ func buildBabysitPrompt(branch, _ string, prs []toolv1.EnrichedPR, lastChecked t return sb.String() } +func hasBabysitActivity(prs []toolv1.EnrichedPR) bool { + for _, pr := range prs { + if len(pr.NewComments) > 0 { + return true + } + if pr.Details != nil && hasFailingCICheck(pr.Details.CIChecks) { + return true + } + } + + return false +} + +func hasFailingCICheck(checks []scm.CICheck) bool { + for _, ci := range checks { + switch ci.Conclusion { + case scm.CICheckConclusionFailure, "timed_out", scm.CICheckConclusionCancelled: + return true + } + } + + return false +} + func (in *agentRunController) newOrUpdatedPRComments(prURL string, comments []scm.PRComment) []scm.PRComment { in.ensureSeenPRCommentBodies() diff --git a/lib/console/ai/workbench/engine.ex b/lib/console/ai/workbench/engine.ex index afb93c8349..b1652f543f 100644 --- a/lib/console/ai/workbench/engine.ex +++ b/lib/console/ai/workbench/engine.ex @@ -87,7 +87,7 @@ defmodule Console.AI.Workbench.Engine do tools(job, environment, activities) |> MemoryEngine.new(50, engine_opts(job) ++ [system_prompt: &sysprompt(job, environment, &1), acc: %Acc{messages: msgs}, tool_fmt: &tool_fmt/1, callback: &callback(job, &1)]) - |> MemoryEngine.reduce(Enum.reverse([{:user, String.trim(continue_prompt(engine: engine))} | messages]), &reducer/2) + |> MemoryEngine.reduce(Enum.reverse(maybe_continue(messages, engine) ++ [{:user, job.prompt}]), &reducer/2) |> case do {:ok, %Complete{ conclusion: conclusion, @@ -370,6 +370,9 @@ defmodule Console.AI.Workbench.Engine do end defp verifiable(engine), do: engine + defp maybe_continue(messages, %__MODULE__{activities: [_ | _]} = engine), do: [{:user, String.trim(continue_prompt(engine: engine))} | messages] + defp maybe_continue(messages, _), do: messages + defp preload_job(%WorkbenchJob{type: :skill} = job), do: Repo.preload(job, @preloads ++ [referenced_job: [:result, workbench: [:workbench_skills, :repository], activities: :thoughts]]) defp preload_job(job), do: Repo.preload(job, @preloads) diff --git a/lib/console/ai/workbench/heartbeat.ex b/lib/console/ai/workbench/heartbeat.ex index c7576c03eb..d69d968164 100644 --- a/lib/console/ai/workbench/heartbeat.ex +++ b/lib/console/ai/workbench/heartbeat.ex @@ -70,7 +70,7 @@ defmodule Console.AI.Workbench.Heartbeat do do: {:stop, {:shutdown, {:budget, :cost, tc}}, %{s | usage: usage}} defp enforce_budget(usage, %State{} = state), do: {:noreply, %{state | usage: usage}} - defp preserve_usage(%WorkbenchJob{usage: %{} = usage}), do: AIUsage.sanitize(usage) + defp preserve_usage(%{} = usage), do: AIUsage.sanitize(usage) defp preserve_usage(_), do: %{} defp reprompt(%WorkbenchJob{usage: %{}}), do: true diff --git a/lib/console/chat/utils.ex b/lib/console/chat/utils.ex index c97d6fc73e..6c3b2f6226 100644 --- a/lib/console/chat/utils.ex +++ b/lib/console/chat/utils.ex @@ -31,11 +31,19 @@ defmodule Console.Chat.Utils do """ def handle_mention(%Reference{id: external_id} = msg, %Reference{text: channel} = chan_ref, %ChatConnection{id: id} = conn, %{} = extra) do bot = Workbenches.workbench_chatbot(id, channel) |> Repo.preload([user: [:groups]]) - with %WorkbenchChatbot{user: %User{} = user, prompt: prompt, message_behavior: behavior} = chatbot <- bot do - prompt = prompt(chat: conn, msg: msg, channel: chan_ref, custom: prompt, behavior: behavior) + with %WorkbenchChatbot{user: %User{} = user, prompt: custom, message_behavior: behavior} = chatbot <- bot do case parent_job(msg) do - %WorkbenchJob{} = job -> Workbenches.create_message(%{prompt: prompt}, job, user) + %WorkbenchJob{} = job -> + prompt = reply_prompt(chat: conn, msg: msg, channel: chan_ref, custom: custom, job: job) + + Workbenches.create_queued_prompt(%{ + prompt: prompt, + dequeable_at: DateTime.utc_now() + }, job, user) + _ -> + prompt = prompt(chat: conn, msg: msg, channel: chan_ref, custom: custom, behavior: behavior) + chatbot_message = Map.merge(%{ message: msg.text, @@ -61,7 +69,7 @@ defmodule Console.Chat.Utils do chatbot_msg(id) |> Repo.preload(:workbench_job, force: true) |> case do - %ChatbotMessage{workbench_job: %WorkbenchJob{status: s} = job} when s not in ~w(running pending)a -> job + %ChatbotMessage{workbench_job: %WorkbenchJob{} = job} -> job _ -> nil end end @@ -71,4 +79,5 @@ defmodule Console.Chat.Utils do defp cache_id(%ChatbotMessage{id: id}), do: id EEx.function_from_file(:defp, :prompt, Console.priv_filename(["prompts", "workbench", "chat.md.eex"]), [:assigns]) + EEx.function_from_file(:defp, :reply_prompt, Console.priv_filename(["prompts", "workbench", "chat_reply.md.eex"]), [:assigns]) end diff --git a/lib/console/deployments/events.ex b/lib/console/deployments/events.ex index 5fb85cdabe..f235685669 100644 --- a/lib/console/deployments/events.ex +++ b/lib/console/deployments/events.ex @@ -131,6 +131,7 @@ defmodule Console.PubSub.WorkbenchCronDeleted, do: use Piazza.PubSub.Event defmodule Console.PubSub.WorkbenchPromptCreated, do: use Piazza.PubSub.Event defmodule Console.PubSub.WorkbenchPromptUpdated, do: use Piazza.PubSub.Event defmodule Console.PubSub.WorkbenchPromptDeleted, do: use Piazza.PubSub.Event +defmodule Console.PubSub.WorkbenchQueuedPromptCreated, do: use Piazza.PubSub.Event defmodule Console.PubSub.WorkbenchSkillCreated, do: use Piazza.PubSub.Event defmodule Console.PubSub.WorkbenchSkillUpdated, do: use Piazza.PubSub.Event defmodule Console.PubSub.WorkbenchSkillDeleted, do: use Piazza.PubSub.Event diff --git a/lib/console/deployments/pubsub/recurse.ex b/lib/console/deployments/pubsub/recurse.ex index 2716180cf7..422666d05b 100644 --- a/lib/console/deployments/pubsub/recurse.ex +++ b/lib/console/deployments/pubsub/recurse.ex @@ -336,3 +336,7 @@ end defimpl Console.PubSub.Recurse, for: Console.PubSub.WorkbenchJobCreated do def process(%{item: job}), do: Console.Pipelines.AI.Workbench.Producer.kick(job) end + +defimpl Console.PubSub.Recurse, for: Console.PubSub.WorkbenchQueuedPromptCreated do + def process(%{item: prompt}), do: Console.Pipelines.AI.QueuedPrompt.Producer.kick(prompt) +end diff --git a/lib/console/deployments/workbenches.ex b/lib/console/deployments/workbenches.ex index d4e4e3b0f9..a2af362289 100644 --- a/lib/console/deployments/workbenches.ex +++ b/lib/console/deployments/workbenches.ex @@ -804,6 +804,7 @@ defmodule Console.Deployments.Workbenches do |> QueuedPrompt.changeset(attrs) |> allow(user, :read) |> when_ok(:insert) + |> notify(:create, user) end def create_queued_prompt(attrs, job_id, %User{} = user) when is_binary(job_id) do @@ -1176,6 +1177,8 @@ defmodule Console.Deployments.Workbenches do do: handle_notify(PubSub.WorkbenchPromptUpdated, prompt, actor: user) defp notify({:ok, %WorkbenchPrompt{} = prompt}, :delete, user), do: handle_notify(PubSub.WorkbenchPromptDeleted, prompt, actor: user) + defp notify({:ok, %QueuedPrompt{} = prompt}, :create, user), + do: handle_notify(PubSub.WorkbenchQueuedPromptCreated, prompt, actor: user) defp notify({:ok, %WorkbenchSkill{} = skill}, :create, user), do: handle_notify(PubSub.WorkbenchSkillCreated, skill, actor: user) defp notify({:ok, %WorkbenchSkill{} = skill}, :update, user), diff --git a/lib/console/services/users.ex b/lib/console/services/users.ex index b00b4a06db..51809818a9 100644 --- a/lib/console/services/users.ex +++ b/lib/console/services/users.ex @@ -258,7 +258,7 @@ defmodule Console.Services.Users do Determines if a user can refresh their jwt and returns the user back if so """ @spec authorize_refresh(binary) :: user_resp - def authorize_refresh(token) do + def authorize_refresh(token) when is_binary(token) do get_refresh_token(token) |> Repo.preload([:user]) |> case do @@ -266,6 +266,7 @@ defmodule Console.Services.Users do _ -> {:error, "could not fetch refresh token"} end end + def authorize_refresh(_), do: {:error, "no refresh token provided"} def create_service_account(attrs) do %User{service_account: true} diff --git a/priv/prompts/workbench/chat_reply.md.eex b/priv/prompts/workbench/chat_reply.md.eex new file mode 100644 index 0000000000..97c51044a1 --- /dev/null +++ b/priv/prompts/workbench/chat_reply.md.eex @@ -0,0 +1,7 @@ +A user posted this reply in <%= @chat.type %> you must follow up on: + +``` +<%= @msg.text %> +``` + +Before working, react to this in <%= @chat.type %>. Message details are id=<%= @msg.id %>, channel=<%= @channel.text %>,channel_id=id=<%= @channel.id %>. Then do what you can to address the prompt above. diff --git a/priv/prompts/workbench/job.md.eex b/priv/prompts/workbench/job.md.eex index 0d90444f46..3b6a74ed6f 100644 --- a/priv/prompts/workbench/job.md.eex +++ b/priv/prompts/workbench/job.md.eex @@ -35,11 +35,16 @@ a gitops change is unlikely to be the fix or the fix needs to be applied immedia **You will also be given a list of skills to guide your investigation. You should search them at least once to potentially gather useful context on how this should be done. Skills do not change between investigations though, so don't requery them if you already know them** -The overarching task is as follows: +There can be follow-up messages for this job sent by a user, which can redefine the task at hand. **These should also be considered fully valid tasks, and do not assume the initial task must be always completed if they change scope.** -<%= @prompt %> +Examples of this are: +* a user starting a job as an analysis task and then realizing they want to create a pr or make some change in response to findings. +* a user doing additional work in the course of a long running job. +* etc -The user might resteer this task, so if you see any user messages, be sure to use them to guide what work needs to be done as well. +Always respect a user's requirements throughout the entire journey of this investigation. + +Also if a user resteers an investigation, *always regenerate the TODOs and theory for the job to represent that new objective.* ## Tone of Voice Guidance diff --git a/test/console/ai/workbench/heartbeat_test.exs b/test/console/ai/workbench/heartbeat_test.exs new file mode 100644 index 0000000000..4c17e02be1 --- /dev/null +++ b/test/console/ai/workbench/heartbeat_test.exs @@ -0,0 +1,53 @@ +defmodule Console.AI.Workbench.HeartbeatTest do + use Console.DataCase, async: true + alias Console.AI.Workbench.Heartbeat + + @usage %{ + input_tokens: 100, + output_tokens: 25, + total_tokens: 125, + cached_tokens: 10, + reasoning_tokens: 5, + input_cost: 0.01, + output_cost: 0.02, + total_cost: 0.03 + } + + describe "usage_callback/2" do + test "adds new usage to usage already persisted on the job" do + job = insert(:workbench_job, status: :running, usage: @usage) + {:ok, pid} = Heartbeat.start_link(job) + Process.unlink(pid) + + on_exit(fn -> + if Process.alive?(pid), do: GenServer.stop(pid, :cancel) + end) + + Heartbeat.usage_callback(job, %{ + input_tokens: 5, + output_tokens: 4, + total_tokens: 9, + cached_tokens: 1, + reasoning_tokens: 2, + input_cost: 0.005, + output_cost: 0.006, + total_cost: 0.011 + }) + + %{usage: usage} = :sys.get_state(pid) + + assert usage.input_tokens == 105 + assert usage.output_tokens == 29 + assert usage.total_tokens == 134 + assert usage.cached_tokens == 11 + assert usage.reasoning_tokens == 7 + assert_in_delta usage.input_cost, 0.015, 0.000_001 + assert_in_delta usage.output_cost, 0.026, 0.000_001 + assert_in_delta usage.total_cost, 0.041, 0.000_001 + + ExUnit.CaptureLog.capture_log(fn -> + GenServer.stop(pid, :cancel) + end) + end + end +end diff --git a/test/console/chat/utils_test.exs b/test/console/chat/utils_test.exs index 52cd88233c..88811f1612 100644 --- a/test/console/chat/utils_test.exs +++ b/test/console/chat/utils_test.exs @@ -2,7 +2,7 @@ defmodule Console.Chat.UtilsTest do use Console.DataCase, async: true alias Console.Chat.{Reference, Utils} - alias Console.Schema.{ChatbotMessage, WorkbenchJob, WorkbenchJobActivity} + alias Console.Schema.{ChatbotMessage, QueuedPrompt, WorkbenchJob, WorkbenchJobActivity} describe "handle_mention/3" do test "creates a workbench job for a new chatbot mention" do @@ -40,7 +40,7 @@ defmodule Console.Chat.UtilsTest do refute chatbot_msg.external_parent_id end - test "creates a job message for a reply to a prior chatbot message" do + test "queues an immediate prompt for a reply to a prior chatbot message" do user = insert(:user) workbench = insert(:workbench, read_bindings: [%{user_id: user.id}]) conn = insert(:chat_connection) @@ -73,15 +73,69 @@ defmodule Console.Chat.UtilsTest do text: "<@bot> follow up on this" } - {:ok, %WorkbenchJobActivity{} = activity} = + {:ok, %QueuedPrompt{} = prompt} = Utils.handle_mention(msg, %Reference{id: "C123", text: channel}, conn) - assert activity.workbench_job_id == job.id - assert activity.user_id == user.id - assert activity.type == :user - assert activity.status == :successful - assert activity.prompt =~ msg.text - assert refetch(job).status == :pending + assert prompt.workbench_job_id == job.id + assert prompt.user_id == user.id + assert prompt.prompt =~ msg.text + assert prompt.prompt =~ "A user posted this reply in #{conn.type} you must follow up on" + assert prompt.prompt =~ "Before working, react to this in #{conn.type}" + assert prompt.prompt =~ "Message details are id=#{msg.id}" + assert prompt.prompt =~ "channel=#{channel}" + assert prompt.prompt =~ "channel_id=id=C123" + refute prompt.prompt =~ "You should reply to the original message" + assert DateTime.compare(prompt.dequeable_at, DateTime.utc_now()) in [:lt, :eq] + assert refetch(job).status == :successful + assert Repo.all(WorkbenchJobActivity) == [] + end + + test "queues replies even when the parent job is active" do + user = insert(:user) + workbench = insert(:workbench, read_bindings: [%{user_id: user.id}]) + conn = insert(:chat_connection) + channel = "chat-channel" + parent_ts = "1730000000.000001" + + insert(:workbench_chatbot, + channel: channel, + chat_connection: conn, + workbench: workbench, + user: user + ) + + job = + insert(:workbench_job, + status: :running, + user: user, + workbench: workbench + ) + + insert(:chatbot_message, + external_id: parent_ts, + chat_connection: conn, + workbench_job: job + ) + + msg = %Reference{ + id: "1730000001.000002", + parent_id: parent_ts, + text: "<@bot> follow up while running" + } + + {:ok, %QueuedPrompt{} = prompt} = + Utils.handle_mention(msg, %Reference{id: "C123", text: channel}, conn) + + assert prompt.workbench_job_id == job.id + assert prompt.prompt =~ msg.text + assert prompt.prompt =~ "A user posted this reply in #{conn.type} you must follow up on" + assert prompt.prompt =~ "Before working, react to this in #{conn.type}" + assert prompt.prompt =~ "Message details are id=#{msg.id}" + assert prompt.prompt =~ "channel=#{channel}" + assert prompt.prompt =~ "channel_id=id=C123" + refute prompt.prompt =~ "You should reply to the original message" + assert DateTime.compare(prompt.dequeable_at, DateTime.utc_now()) in [:lt, :eq] + assert refetch(job).status == :running end end end