diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f30bafa6..51b81eee5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,8 +16,27 @@ upgrade, is in ## [Unreleased] +### Added + +- **`!rotate` from a Buzz channel opens a new conversation.** The channel-bound + resume (#774) meant a rotated harness's next `session/new` landed straight + back on the same conversation, so rotation did nothing on a hosted agent. + The harness now sends `_meta.freshSession: true` on that one `session/new` + (block/buzz#6103); `fountain acp` forwards it as `fresh: true` on + `POST /api/conversations`, which unbinds the current conversation from the + channel (it keeps running and is retired like any other idle one) and opens + a new one as the binding. `fresh` is documented in the OpenAPI schema and + ignored without `channel_id`. + ### Fixed +- **`!shutdown` no longer restart-loops a hosted harness.** The supervisor + restarts `buzz-acp` on any exit and the fresh process replayed the same + `!shutdown` from its subscription backlog — five exits per command before + the message aged out, ending *online*. The harness now ignores owner + control commands created before it started (block/buzz#6104). The pin moves + to `buzz-acp-v0.5.14-fountain.3` for both changes. + - **Owner control commands (`!rotate`, `!cancel`, `!shutdown`) now work from the Buzz Desktop composer.** The hosted `buzz-acp` required the message body to be *exactly* the command, but Desktop renders the `@Name` mention into the diff --git a/Dockerfile b/Dockerfile index e9514238e..16602992d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -90,7 +90,7 @@ RUN CGO_ENABLED=0 GOOS=linux GOARCH="${TARGETARCH:-amd64}" \ FROM debian:trixie-slim AS buzzacp ARG TARGETARCH -ARG BUZZ_ACP_VERSION=0.5.14-fountain.2 +ARG BUZZ_ACP_VERSION=0.5.14-fountain.3 # buzz-acp: the hosted harness (gate 2). buzz: the CLI the server-side MCP tools # shell out to for signed publishes (gate 3, #737). Both from our own release. RUN apt-get update -y \ diff --git a/apps/fountain/lib/fountain/conversations.ex b/apps/fountain/lib/fountain/conversations.ex index fb9fa3190..f5b55f0a3 100644 --- a/apps/fountain/lib/fountain/conversations.ex +++ b/apps/fountain/lib/fountain/conversations.ex @@ -970,6 +970,15 @@ defmodule Fountain.Conversations do resuming, so a new one is opened and becomes the binding. Returns `{:ok, conv, :resumed}` or `{:ok, conv, :created}`; without a `channel_id` it always creates. + `attrs["fresh"]` (`true`) skips the resume this once: the conversation + currently bound to the channel is unbound (its `channel_id` cleared — it + keeps running, and the sandbox reaper retires it like any other idle one) + and a new one is opened as the binding. It is how a chat harness relays its + owner's `!rotate` — ACP `session/new` `_meta.freshSession` — through a + binding that would otherwise hand the old conversation straight back. + Unbinding, rather than relying on "newest wins", keeps the outcome + independent of `inserted_at`'s one-second precision. + Two concurrent first calls for one channel can both create; the next call resumes whichever is newer. Nothing is audited on the resume path — nothing changed. @@ -986,7 +995,13 @@ defmodule Fountain.Conversations do {:ok, env_id} <- resolve_environment_id(attrs["environment_id"], user_id, agent) do case find_channel_conversation(user_id, agent.id, vault_id, env_id, channel_id) do %Conversation{} = conv -> - {:ok, conv, :resumed} + if fresh_requested?(attrs) do + with {:ok, _} <- unbind_channel(conv), + {:ok, fresh} <- start_conversation(attrs, opts), + do: {:ok, fresh, :created} + else + {:ok, conv, :resumed} + end nil -> with {:ok, conv} <- start_conversation(attrs, opts), do: {:ok, conv, :created} @@ -998,6 +1013,20 @@ defmodule Fountain.Conversations do with {:ok, conv} <- start_conversation(attrs, opts), do: {:ok, conv, :created} end + # `true` or `"true"` — the ACP adapter sends a JSON boolean, a hand-built + # request may send a string. Anything else is not a request. + defp fresh_requested?(%{"fresh" => fresh}), do: fresh in [true, "true"] + defp fresh_requested?(_attrs), do: false + + # The rotated-away conversation stops being the channel's binding. Nothing + # else about it changes: if it is mid-turn it finishes, and it stays in the + # user's list under its own id. + defp unbind_channel(%Conversation{} = conv) do + conv + |> Ecto.Changeset.change(channel_id: nil) + |> Repo.update() + end + # The newest conversation still worth resuming for this binding. `vault_id` # is part of the key: two entries on one agent with different vaults are # different identities (#727) and must not share a conversation. So is the diff --git a/apps/fountain/lib/fountain_web/schemas.ex b/apps/fountain/lib/fountain_web/schemas.ex index 197cf5e39..155cdd477 100644 --- a/apps/fountain/lib/fountain_web/schemas.ex +++ b/apps/fountain/lib/fountain_web/schemas.ex @@ -205,6 +205,14 @@ defmodule FountainWeb.Schemas do "Opaque key for the external channel this conversation is bound to (for example a " <> "Buzz channel id). When set, the latest live conversation for the same agent, vault " <> "and channel is resumed (200) instead of a new one being opened (201)." + }, + fresh: %Schema{ + type: :boolean, + nullable: true, + description: + "With channel_id: skip the resume and open a new conversation (201), which then " <> + "becomes the channel's binding. Sent by a chat harness relaying its owner's " <> + "rotate command. Ignored without channel_id." } }, required: [:agent_id] diff --git a/apps/fountain/test/fountain_web/controllers/conversation_controller_test.exs b/apps/fountain/test/fountain_web/controllers/conversation_controller_test.exs index 73a81693c..ed7518e5a 100644 --- a/apps/fountain/test/fountain_web/controllers/conversation_controller_test.exs +++ b/apps/fountain/test/fountain_web/controllers/conversation_controller_test.exs @@ -1037,6 +1037,40 @@ defmodule FountainWeb.ConversationControllerTest do second["data"]["id"] end + test "fresh: true opens a new conversation despite the binding, and it takes over", + %{conn: conn, raw_key: raw_key, agent: agent} do + body = %{"agent_id" => agent.id, "channel_id" => "chan-r"} + first = create(conn, raw_key, body) |> json_response(201) + + assert create(conn, raw_key, body) |> json_response(200) |> get_in(["data", "id"]) == + first["data"]["id"] + + # The harness's owner rotated the channel (ACP _meta.freshSession). + rotated = create(conn, raw_key, Map.put(body, "fresh", true)) |> json_response(201) + refute rotated["data"]["id"] == first["data"]["id"] + assert rotated["data"]["channel_id"] == "chan-r" + assert rotated["meta"]["resumed"] == false + + # From now on the new one is what an ordinary session/new resumes, and + # the old one is unbound (not terminated — it may be mid-turn). + assert create(conn, raw_key, body) |> json_response(200) |> get_in(["data", "id"]) == + rotated["data"]["id"] + + old = Fountain.Conversations._unsafe_get_conversation!(first["data"]["id"]) + assert old.channel_id == nil + refute old.status in ["terminated", "failed"] + + # fresh: false / absent / a non-boolean does not rotate. + assert create(conn, raw_key, Map.put(body, "fresh", false)) + |> json_response(200) + |> get_in(["data", "id"]) == rotated["data"]["id"] + + # Without a channel key fresh is meaningless — creates like any other. + assert %{"data" => %{"channel_id" => nil}} = + create(conn, raw_key, %{"agent_id" => agent.id, "fresh" => true}) + |> json_response(201) + end + test "another user's binding is invisible", %{conn: conn, agent: agent} do other = insert_verified_user() {_k, other_key} = insert_api_key(other) diff --git a/buzz-acp.source b/buzz-acp.source index 3a7a4d23a..1d13ecbbf 100644 --- a/buzz-acp.source +++ b/buzz-acp.source @@ -1,2 +1,2 @@ # Fork pin — see #776 for when and how to remove. Format: owner/repo@ref -jhgaylor/buzz@0bb9e5af7713ca6946b6eb5a332d817897024a31 +jhgaylor/buzz@f98c2a7f2d0882d632d0f03a67e2cb687101ccd4 diff --git a/buzz-acp.version b/buzz-acp.version index 6f4fa49b4..07915f601 100644 --- a/buzz-acp.version +++ b/buzz-acp.version @@ -1 +1 @@ -0.5.14-fountain.2 +0.5.14-fountain.3 diff --git a/cli/internal/acp/session.go b/cli/internal/acp/session.go index 6b3700f20..7d74db565 100644 --- a/cli/internal/acp/session.go +++ b/cli/internal/acp/session.go @@ -16,8 +16,10 @@ type API interface { // CreateConversation starts a conversation for an agent and returns its id. // With a non-empty channelID the server resumes the latest live conversation // already bound to that channel for the same agent and vault instead of - // opening a new one (#774); resumed reports which happened. - CreateConversation(ctx context.Context, agentID, channelID string) (id string, resumed bool, err error) + // opening a new one (#774); resumed reports which happened. fresh asks the + // server to skip that resume and open a new conversation that becomes the + // channel's binding — the client's owner rotated the channel. + CreateConversation(ctx context.Context, agentID, channelID string, fresh bool) (id string, resumed bool, err error) // StreamHead returns the conversation's current last event id, so a follow // can skip the history. Called BEFORE a prompt is sent — see prompt.go. StreamHead(ctx context.Context, convID string) (string, error) @@ -115,10 +117,14 @@ type newSessionParams struct { // Meta is the out-of-band bag ACP lets a client attach. `channelId` is // what a chat harness (buzz-acp) sends to name the channel this session // serves — the one thing the server can key a resume on, since the same - // harness forgets its sessions on every restart. Anything else in _meta - // is ignored here. + // harness forgets its sessions on every restart. `freshSession` rides + // with it on the first session/new after the harness's owner rotated the + // channel (`!rotate`): the resume must be skipped this once, or rotation + // is a no-op behind a channel-keyed server. Anything else in _meta is + // ignored here. Meta struct { - ChannelID string `json:"channelId"` + ChannelID string `json:"channelId"` + FreshSession bool `json:"freshSession"` } `json:"_meta"` } @@ -174,8 +180,10 @@ func (a *Agent) newSession(ctx context.Context, raw json.RawMessage) (any, error // session with a fresh id (it did not know the old one — that is the // point); from Fountain's side it is the same conversation, sandbox and // runtime session, which is what makes a chat harness's restart invisible - // to the people in the channel (#774). - convID, resumed, err := a.api.CreateConversation(ctx, ref.ID, params.Meta.ChannelID) + // to the people in the channel (#774). Unless the harness says the owner + // rotated the channel — then a new conversation is opened and becomes + // the binding. + convID, resumed, err := a.api.CreateConversation(ctx, ref.ID, params.Meta.ChannelID, params.Meta.FreshSession) if err != nil { return nil, Errorf(CodeInternalError, "could not start a conversation for %q: %s", ref.Name, err) } @@ -184,7 +192,7 @@ func (a *Agent) newSession(ctx context.Context, raw json.RawMessage) (any, error a.sessions.put(sess) a.log.Info("session opened", "sessionId", sess.ID, "agent", ref.Name, "runtime", ref.Runtime, "model", ref.Model, - "channelId", params.Meta.ChannelID, "resumed", resumed) + "channelId", params.Meta.ChannelID, "fresh", params.Meta.FreshSession, "resumed", resumed) return map[string]any{ "sessionId": sess.ID, diff --git a/cli/internal/acp/session_test.go b/cli/internal/acp/session_test.go index 1cd98ee06..ed82e2483 100644 --- a/cli/internal/acp/session_test.go +++ b/cli/internal/acp/session_test.go @@ -17,6 +17,7 @@ type fakeAPI struct { resolved []string created []string channels []string + fresh []bool convID string resumed bool @@ -61,9 +62,10 @@ func (f *fakeAPI) Agent(_ context.Context, target string) (AgentRef, error) { return f.ref, nil } -func (f *fakeAPI) CreateConversation(_ context.Context, agentID, channelID string) (string, bool, error) { +func (f *fakeAPI) CreateConversation(_ context.Context, agentID, channelID string, fresh bool) (string, bool, error) { f.created = append(f.created, agentID) f.channels = append(f.channels, channelID) + f.fresh = append(f.fresh, fresh) if f.createErr != nil { return "", false, f.createErr } @@ -344,6 +346,34 @@ func TestNewSessionForwardsTheChannelKeyAndAcceptsAResumedConversation(t *testin } } +// The first session/new after the harness's owner rotated the channel +// (`!rotate`) carries `_meta.freshSession: true`. That must reach the server as +// a request to skip the resume — otherwise the channel-bound resume hands the +// same conversation straight back and rotation is a no-op. +func TestNewSessionForwardsFreshSessionAfterARotate(t *testing.T) { + api := &fakeAPI{ref: acpAgentRef(), convID: "conv-new"} + a := sessionAgent(t, api, "researcher") + + if _, rpcErr := request(t, a, "session/new", map[string]any{ + "_meta": map[string]any{"channelId": "chan-002b49f3", "freshSession": true}, + }); rpcErr != nil { + t.Fatalf("session/new failed: %v", rpcErr) + } + if len(api.fresh) != 1 || !api.fresh[0] { + t.Errorf("fresh forwarded = %v, want [true]", api.fresh) + } + + // Absent → false: an ordinary (or restarted) session/new still resumes. + if _, rpcErr := request(t, a, "session/new", map[string]any{ + "_meta": map[string]any{"channelId": "chan-002b49f3"}, + }); rpcErr != nil { + t.Fatalf("session/new failed: %v", rpcErr) + } + if len(api.fresh) != 2 || api.fresh[1] { + t.Errorf("fresh forwarded = %v, want [true false]", api.fresh) + } +} + // No _meta.channelId → no channel key: an editor's session/new must not // accidentally bind to "" and start resuming across unrelated projects. func TestNewSessionWithoutAChannelKeySendsNone(t *testing.T) { diff --git a/cli/internal/cmd/acp.go b/cli/internal/cmd/acp.go index 52a2bb9e4..83f7335c5 100644 --- a/cli/internal/cmd/acp.go +++ b/cli/internal/cmd/acp.go @@ -204,7 +204,7 @@ func agentRef(data map[string]any) acp.AgentRef { } } -func (f fountainAPI) CreateConversation(_ context.Context, agentID, channelID string) (string, bool, error) { +func (f fountainAPI) CreateConversation(_ context.Context, agentID, channelID string, fresh bool) (string, bool, error) { var resp struct { Data map[string]any `json:"data"` Meta map[string]any `json:"meta"` @@ -217,6 +217,11 @@ func (f fountainAPI) CreateConversation(_ context.Context, agentID, channelID st // client sent none — an empty key would be a binding to "". if channelID != "" { body["channel_id"] = channelID + // The harness's owner rotated the channel: open a new conversation + // even though one is bound. Only meaningful with a channel key. + if fresh { + body["fresh"] = true + } } // A vault carries the secrets that belong to this entry rather than to the diff --git a/cli/internal/cmd/acp_test.go b/cli/internal/cmd/acp_test.go index d2950516e..b505056ce 100644 --- a/cli/internal/cmd/acp_test.go +++ b/cli/internal/cmd/acp_test.go @@ -189,7 +189,7 @@ func TestCreateConversationAttachesTheVault(t *testing.T) { api := acpTestAPI(t, srv.URL) api.vault = "buzz-philo" - id, _, err := api.CreateConversation(context.Background(), "agent-1", "") + id, _, err := api.CreateConversation(context.Background(), "agent-1", "", false) if err != nil { t.Fatalf("CreateConversation: %v", err) } @@ -215,7 +215,7 @@ func TestCreateConversationOmitsTheVaultWhenUnset(t *testing.T) { api := acpTestAPI(t, srv.URL) - if _, _, err := api.CreateConversation(context.Background(), "agent-1", ""); err != nil { + if _, _, err := api.CreateConversation(context.Background(), "agent-1", "", false); err != nil { t.Fatalf("CreateConversation: %v", err) } if _, present := body["vault_id"]; present { @@ -245,7 +245,7 @@ func TestCreateConversationAttachesTheEnvironment(t *testing.T) { api := acpTestAPI(t, srv.URL) api.environment = "buzz-env" - if _, _, err := api.CreateConversation(context.Background(), "agent-1", ""); err != nil { + if _, _, err := api.CreateConversation(context.Background(), "agent-1", "", false); err != nil { t.Fatalf("CreateConversation: %v", err) } if body["environment_id"] != "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" { @@ -266,7 +266,7 @@ func TestCreateConversationOmitsTheEnvironmentWhenUnset(t *testing.T) { api := acpTestAPI(t, srv.URL) - if _, _, err := api.CreateConversation(context.Background(), "agent-1", ""); err != nil { + if _, _, err := api.CreateConversation(context.Background(), "agent-1", "", false); err != nil { t.Fatalf("CreateConversation: %v", err) } if _, present := body["environment_id"]; present { @@ -288,7 +288,7 @@ func TestCreateConversationSendsTheChannelKeyAndReadsResumed(t *testing.T) { api := acpTestAPI(t, srv.URL) - id, resumed, err := api.CreateConversation(context.Background(), "agent-1", "chan-1") + id, resumed, err := api.CreateConversation(context.Background(), "agent-1", "chan-1", false) if err != nil { t.Fatalf("CreateConversation: %v", err) } @@ -298,6 +298,9 @@ func TestCreateConversationSendsTheChannelKeyAndReadsResumed(t *testing.T) { if body["channel_id"] != "chan-1" { t.Errorf("channel_id = %v, want chan-1", body["channel_id"]) } + if _, ok := body["fresh"]; ok { + t.Errorf("fresh sent = %v, want absent when not rotated", body["fresh"]) + } } func TestCreateConversationOmitsTheChannelKeyWhenEmpty(t *testing.T) { @@ -311,7 +314,7 @@ func TestCreateConversationOmitsTheChannelKeyWhenEmpty(t *testing.T) { defer srv.Close() api := acpTestAPI(t, srv.URL) - _, resumed, err := api.CreateConversation(context.Background(), "agent-1", "") + _, resumed, err := api.CreateConversation(context.Background(), "agent-1", "", false) if err != nil { t.Fatalf("CreateConversation: %v", err) } @@ -323,6 +326,41 @@ func TestCreateConversationOmitsTheChannelKeyWhenEmpty(t *testing.T) { } } +// A rotated channel (`_meta.freshSession`) rides as `fresh: true` next to the +// channel key, so the server opens a new conversation instead of resuming. +func TestCreateConversationSendsFreshWithTheChannelKey(t *testing.T) { + var body map[string]any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewDecoder(r.Body).Decode(&body) + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"data":{"id":"conv-new","channel_id":"chan-1"},"meta":{"resumed":false}}`)) + })) + defer srv.Close() + + api := acpTestAPI(t, srv.URL) + + id, resumed, err := api.CreateConversation(context.Background(), "agent-1", "chan-1", true) + if err != nil { + t.Fatalf("CreateConversation: %v", err) + } + if id != "conv-new" || resumed { + t.Errorf("got (%q, %v), want (conv-new, false)", id, resumed) + } + if body["channel_id"] != "chan-1" || body["fresh"] != true { + t.Errorf("body = %v, want channel_id chan-1 and fresh true", body) + } + + // fresh without a channel key is meaningless and must not be sent. + body = nil + if _, _, err := api.CreateConversation(context.Background(), "agent-1", "", true); err != nil { + t.Fatalf("CreateConversation: %v", err) + } + if _, ok := body["fresh"]; ok { + t.Errorf("fresh sent = %v without a channel key, want absent", body["fresh"]) + } +} + func TestUnknownEnvironmentNameIsReported(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte(`{"data":[]}`)) @@ -332,7 +370,7 @@ func TestUnknownEnvironmentNameIsReported(t *testing.T) { api := acpTestAPI(t, srv.URL) api.environment = "not-an-env" - _, _, err := api.CreateConversation(context.Background(), "agent-1", "") + _, _, err := api.CreateConversation(context.Background(), "agent-1", "", false) if err == nil || !strings.Contains(err.Error(), "not-an-env") { t.Fatalf("want an error naming the environment, got %v", err) } @@ -347,7 +385,7 @@ func TestUnknownVaultNameIsReported(t *testing.T) { api := acpTestAPI(t, srv.URL) api.vault = "not-a-vault" - _, _, err := api.CreateConversation(context.Background(), "agent-1", "") + _, _, err := api.CreateConversation(context.Background(), "agent-1", "", false) if err == nil || !strings.Contains(err.Error(), "not-a-vault") { t.Fatalf("want an error naming the vault, got %v", err) }