From d7ff114a918132a849c574c195c61a1a4dc7a177 Mon Sep 17 00:00:00 2001 From: Qiyuanqiii <267806965+Qiyuanqiii@users.noreply.github.com> Date: Mon, 19 Oct 2026 13:18:49 +0800 Subject: [PATCH 1/2] Add conversation branches from durable assistant replies --- docs/convergence/protocol-registry.md | 5 +- docs/openapi.json | 181 ++++++++++++ internal/application/thread_branch.go | 120 ++++++++ internal/application/thread_branch_test.go | 162 +++++++++++ internal/domain/thread_branch.go | 25 ++ internal/httpapi/openapi.go | 7 + internal/httpapi/openapi_test.go | 21 ++ internal/httpapi/thread_branch.go | 45 +++ internal/httpapi/thread_branch_test.go | 91 ++++++ internal/httpapi/thread_control.go | 6 +- internal/store/thread_branch.go | 319 +++++++++++++++++++++ internal/store/thread_branch_test.go | 259 +++++++++++++++++ protocols/registry.json | 1 + web/src/api/schema.d.ts | 82 ++++++ 14 files changed, 1321 insertions(+), 3 deletions(-) create mode 100644 internal/application/thread_branch.go create mode 100644 internal/application/thread_branch_test.go create mode 100644 internal/domain/thread_branch.go create mode 100644 internal/httpapi/thread_branch.go create mode 100644 internal/httpapi/thread_branch_test.go create mode 100644 internal/store/thread_branch.go create mode 100644 internal/store/thread_branch_test.go diff --git a/docs/convergence/protocol-registry.md b/docs/convergence/protocol-registry.md index eda56804d..45c7aa56d 100644 --- a/docs/convergence/protocol-registry.md +++ b/docs/convergence/protocol-registry.md @@ -26,7 +26,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi | [exported-evidence-and-handoff](#exported-evidence-and-handoff) | `external-durable` | Evidence, verification, report, and handoff maintainers | 35 | true | | [extension-package-contracts](#extension-package-contracts) | `external-durable` | Skill, Plugin, Hook, and extension maintainers | 48 | true | | [github-pr-followup-observation](#github-pr-followup-observation) | `projection` | GitHub review and task follow-up maintainers | 1 | false | -| [http-openapi-contract](#http-openapi-contract) | `external-durable` | HTTP/OpenAPI and generated-client maintainers | 136 | true | +| [http-openapi-contract](#http-openapi-contract) | `external-durable` | HTTP/OpenAPI and generated-client maintainers | 137 | true | | [in-memory-token-session](#in-memory-token-session) | `ephemeral` | Credential and bootstrap maintainers | 1 | false | | [lsp-process-session](#lsp-process-session) | `ephemeral` | Code intelligence maintainers | 1 | false | | [mcp-interchange](#mcp-interchange) | `external-durable` | MCP client/server maintainers | 9 | true | @@ -986,7 +986,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi - `pr-followup-http-request-reader` (`v1`, active) at `internal/application/pr_followup.go` - `pr-followup-http-response-reader` (`v1`, active) at `web/src/api/pr-followup.ts` -
136 active identifiers +
137 active identifiers - `agent-code-tools.v1` - `agent_browser_close.v1` @@ -1100,6 +1100,7 @@ This document is generated from [`protocols/registry.json`](../../protocols/regi - `thread_activity_detail.v2` - `thread_activity_summary.v1` - `thread_application_services.v1` +- `thread_branch.v1` - `thread_command_service_policy.v1` - `thread_context_compaction.v1` - `thread_context_compaction_operations.v1` diff --git a/docs/openapi.json b/docs/openapi.json index 77b6e430a..7aba6c58c 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -28172,6 +28172,129 @@ "x-cyberagent-read-only": false } }, + "/api/v1/threads/{thread_id}/branches": { + "post": { + "operationId": "branchThread", + "summary": "Branch a conversation at a durable assistant reply", + "description": "Atomically creates a fresh Thread, Run and Session containing conversation history through source_event_id, including earlier Runs. Historical assistant output is non-authorizing evidence. Requires Thread creation and Session message control. No model or tool is called; approvals and runtime authority are reset. Reusing a key with another reply conflicts.", + "tags": [ + "Control" + ], + "parameters": [ + { + "name": "thread_id", + "in": "path", + "description": "Thread identity", + "required": true, + "schema": { + "maxLength": 256, + "minLength": 1, + "pattern": "^[^/\\\\\\x00-\\x1f\\x7f]+$", + "type": "string" + } + }, + { + "name": "Idempotency-Key", + "in": "header", + "description": "Opaque retry key; only a domain-separated digest is persisted", + "required": true, + "schema": { + "maxLength": 256, + "minLength": 16, + "pattern": "^\\S+$", + "type": "string" + } + } + ], + "responses": { + "201": { + "description": "Resource created or idempotently replayed", + "content": { + "application/json": { + "schema": { + "additionalProperties": false, + "properties": { + "data": { + "$ref": "#/components/schemas/ThreadBranchView" + }, + "request_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "const": "api.v1", + "type": "string" + } + }, + "required": [ + "version", + "request_id", + "data" + ], + "type": "object" + } + } + } + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "403": { + "$ref": "#/components/responses/Forbidden" + }, + "404": { + "$ref": "#/components/responses/NotFound" + }, + "409": { + "$ref": "#/components/responses/Conflict" + }, + "412": { + "$ref": "#/components/responses/FailedPrecondition" + }, + "413": { + "$ref": "#/components/responses/RequestEntityTooLarge" + }, + "414": { + "$ref": "#/components/responses/RequestTooLarge" + }, + "415": { + "$ref": "#/components/responses/UnsupportedMediaType" + }, + "429": { + "$ref": "#/components/responses/ResourceExhausted" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "$ref": "#/components/responses/Unavailable" + }, + "504": { + "$ref": "#/components/responses/GatewayTimeout" + } + }, + "security": [ + { + "ControlBearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ThreadBranchRequestView" + } + } + } + }, + "x-cyberagent-read-only": false + } + }, "/api/v1/threads/{thread_id}/command-service-policy": { "get": { "operationId": "getThreadCommandServicePolicy", @@ -67988,6 +68111,64 @@ ], "type": "object" }, + "ThreadBranchRequestView": { + "additionalProperties": false, + "properties": { + "source_event_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "version": { + "enum": [ + "thread_branch.v1" + ], + "type": "string" + } + }, + "required": [ + "version", + "source_event_id" + ], + "type": "object" + }, + "ThreadBranchView": { + "additionalProperties": false, + "properties": { + "replayed": { + "type": "boolean" + }, + "run": { + "$ref": "#/components/schemas/RunView" + }, + "session": { + "$ref": "#/components/schemas/SessionView" + }, + "source_event_id": { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + "thread": { + "$ref": "#/components/schemas/ThreadView" + }, + "version": { + "enum": [ + "thread_branch.v1" + ], + "type": "string" + } + }, + "required": [ + "version", + "thread", + "run", + "session", + "source_event_id", + "replayed" + ], + "type": "object" + }, "ThreadCommandServicePolicy": { "additionalProperties": false, "properties": { diff --git a/internal/application/thread_branch.go b/internal/application/thread_branch.go new file mode 100644 index 000000000..6703c56cc --- /dev/null +++ b/internal/application/thread_branch.go @@ -0,0 +1,120 @@ +package application + +import ( + "context" + "strings" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/hooks" + "cyberagent-workbench/internal/session" +) + +type ThreadBranchStore interface { + GetThread(context.Context, string) (domain.Thread, error) + GetMission(context.Context, string) (domain.Mission, error) + GetRun(context.Context, string) (domain.Run, error) + GetRunMode(context.Context, string) (domain.RunModeSnapshot, error) + GetSession(context.Context, string) (session.Session, error) + GetThreadBranchSource(context.Context, domain.ThreadBranchRequest) (domain.ThreadBranchSource, error) + GetThreadBranchOperation(context.Context, domain.ThreadBranchRequest) (domain.ThreadBranchRecord, bool, error) + CreateThreadBranch(context.Context, domain.ThreadBranchRequest, domain.Mission, domain.Run, + domain.RunModeSnapshot, session.Session, []events.Event) (domain.ThreadBranchRecord, error) +} + +type ThreadBranchResult struct { + Thread domain.Thread + Run domain.Run + Session session.Session + SourceEventID string + Replayed bool +} + +type ThreadBranchService struct { + store ThreadBranchStore + lifecycleHooks *hooks.Engine +} + +func NewThreadBranchService(store ThreadBranchStore) *ThreadBranchService { + return &ThreadBranchService{store: store} +} + +func (s *ThreadBranchService) WithLifecycleHooks(engine *hooks.Engine) *ThreadBranchService { + s.lifecycleHooks = engine + return s +} + +func (s *ThreadBranchService) Branch(ctx context.Context, request domain.ThreadBranchRequest) (ThreadBranchResult, error) { + if s == nil || s.store == nil { + return ThreadBranchResult{}, apperror.New(apperror.CodeFailedPrecondition, "Thread branch store is unavailable") + } + key, err := domain.NormalizeAgentOperationKey(request.OperationKey) + if err != nil || key != request.OperationKey || strings.ContainsAny(key, " \t\r\n") || + request.Version != domain.ThreadBranchProtocolVersion || !domain.ValidAgentID(request.ThreadID) || + !domain.ValidAgentID(request.SourceEventID) || !domain.ValidAgentID(request.RequestedBy) { + return ThreadBranchResult{}, apperror.New(apperror.CodeInvalidArgument, "Thread branch identity or version is invalid") + } + if stored, found, err := s.store.GetThreadBranchOperation(ctx, request); err != nil { + return ThreadBranchResult{}, err + } else if found { + return s.load(ctx, stored) + } + source, err := s.store.GetThreadBranchSource(ctx, request) + if err != nil { + return ThreadBranchResult{}, err + } + run, err := s.store.GetRun(ctx, source.RunID) + if err != nil { + return ThreadBranchResult{}, err + } + mission, err := s.store.GetMission(ctx, run.MissionID) + if err != nil { + return ThreadBranchResult{}, err + } + mode, err := s.store.GetRunMode(ctx, run.ID) + if err != nil { + return ThreadBranchResult{}, err + } + prepared, err := prepareRun(ctx, CreateRunRequest{ + Goal: mission.Goal, Profile: string(mission.Profile), WorkspaceID: mission.WorkspaceID, + Surface: string(mode.Surface), Phase: string(mode.Phase), ModelRoute: run.Config.ModelRoute, + Interactive: true, Budget: run.Budget, RequestedBudget: run.Config.RequestedBudget, + RequestedBy: request.RequestedBy, WebAccessMode: string(mode.Scope.WebAccessMode), + WebAllowedTargets: append([]string(nil), mode.Scope.WebAllowedTargets...), + }, nil) + if err != nil { + return ThreadBranchResult{}, apperror.Normalize(err) + } + // Retain the pinned project guidance; execution leases, approvals, continuity + // snapshots and provider-private runtime state belong to their original Run. + prepared.Run.Config.ProjectConfig = append([]byte(nil), run.Config.ProjectConfig...) + prepared.Run.Config.ProjectConfigFingerprint = run.Config.ProjectConfigFingerprint + prepared.Run.Config.ProjectInstructions = append([]byte(nil), run.Config.ProjectInstructions...) + prepared.Run.Config.ProjectInstructionsFingerprint = run.Config.ProjectInstructionsFingerprint + if err := executeLifecycleBoundary(ctx, s.lifecycleHooks, hooks.SessionOpened, prepared.Run.ID, + mission.WorkspaceID, map[string]any{"session_id": prepared.Session.ID, "source": "thread_branch"}); err != nil { + return ThreadBranchResult{}, err + } + stored, err := s.store.CreateThreadBranch(ctx, request, prepared.Mission, prepared.Run, prepared.Mode, prepared.Session, prepared.InitialEvents) + if err != nil { + return ThreadBranchResult{}, apperror.Normalize(err) + } + return s.load(ctx, stored) +} + +func (s *ThreadBranchService) load(ctx context.Context, record domain.ThreadBranchRecord) (ThreadBranchResult, error) { + thread, err := s.store.GetThread(ctx, record.ThreadID) + if err != nil { + return ThreadBranchResult{}, err + } + run, err := s.store.GetRun(ctx, record.RunID) + if err != nil { + return ThreadBranchResult{}, err + } + sess, err := s.store.GetSession(ctx, record.SessionID) + if err != nil { + return ThreadBranchResult{}, err + } + return ThreadBranchResult{Thread: thread, Run: run, Session: sess, SourceEventID: record.SourceEventID, Replayed: record.Replayed}, nil +} diff --git a/internal/application/thread_branch_test.go b/internal/application/thread_branch_test.go new file mode 100644 index 000000000..2e74ba358 --- /dev/null +++ b/internal/application/thread_branch_test.go @@ -0,0 +1,162 @@ +package application_test + +import ( + "encoding/json" + "fmt" + "net/http" + "path/filepath" + "strings" + "testing" + + "cyberagent-workbench/internal/application" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/llm" + "cyberagent-workbench/internal/policy" + "cyberagent-workbench/internal/store" +) + +func TestThreadBranchHistoricalCommentaryKeepsDeliveredInputBeforeEveryUpdate(t *testing.T) { + first := toolResponse("branch-note-1", "note_create", `{"title":"First observation","content":"first fact"}`) + first.Text = "first public update" + second := toolResponse("branch-note-2", "note_create", `{"title":"Second observation","content":"second fact"}`) + second.Text = "selected public update" + provider := &scriptedToolProvider{responses: []*llm.ChatResponse{first, second, textResponse(rootActionResponse(domain.RootActionFinish, "FUTURE_FINAL_ANSWER", "done", ""))}} + st, turns, input := threadControlFixture(t, provider) + input.Content = "CURRENT_ATTEMPT_INPUT" + finished, err := turns.Execute(t.Context(), input) + if err != nil { + t.Fatal(err) + } + eventList, err := st.ListRunEvents(t.Context(), finished.Submission.Run.ID) + if err != nil { + t.Fatal(err) + } + anchor := "" + for _, event := range eventList { + var p struct{ Text string } + _ = json.Unmarshal([]byte(event.PayloadJSON), &p) + if event.Type == events.ModelPublicCommentaryEvent && p.Text == second.Text { + anchor = event.EventID + } + } + if anchor == "" { + t.Fatal("tool loop did not produce the selected durable update") + } + branch, err := application.NewThreadBranchService(st).Branch(t.Context(), domain.ThreadBranchRequest{Version: domain.ThreadBranchProtocolVersion, ThreadID: input.ThreadID, SourceEventID: anchor, OperationKey: "historical-commentary-branch", RequestedBy: "test_operator"}) + if err != nil { + t.Fatal(err) + } + messages, err := st.ListSessionMessages(t.Context(), branch.Session.ID, false) + if err != nil { + t.Fatal(err) + } + if len(messages) != 3 || messages[0].Content != input.Content || messages[1].Content != first.Text || messages[2].Content != second.Text { + t.Fatalf("input was lost, duplicated or ordered after an update: %+v", messages) + } + if len(provider.Requests()) != 3 { + t.Fatal("branch repeated original tools or provider work") + } +} + +func TestThreadBranchKimiContinuesWithHistoricalEvidenceAndFreshNativeReplay(t *testing.T) { + calls := 0 + router := kimiSupervisorFixtureRouter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body struct { + Messages []map[string]json.RawMessage `json:"messages"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + w.WriteHeader(400) + return + } + calls++ + encoded, _ := json.Marshal(body.Messages) + if calls >= 3 { + if !strings.Contains(string(encoded), "BRANCH_ORIGINAL_INTENT") || !strings.Contains(string(encoded), "selected answer") || strings.Contains(string(encoded), "FUTURE_ANSWER_CANARY") || strings.Contains(string(encoded), "native-private-1") || strings.Contains(string(encoded), "native-private-2") { + t.Error("branch model input lost selected history or inherited future/private source") + } + assistants := 0 + for _, message := range body.Messages { + var role, reason string + _ = json.Unmarshal(message["role"], &role) + _ = json.Unmarshal(message["reasoning_content"], &reason) + if role == "assistant" { + assistants++ + if calls == 3 || reason != "native-private-3" { + t.Error("imported assistant was sent without fresh native replay") + } + } + } + if calls == 4 && assistants != 1 { + t.Error("second branch turn did not use its own native history") + } + } + text := "selected answer" + if calls == 2 { + text = "FUTURE_ANSWER_CANARY" + } + if calls >= 3 { + text = "continued branch answer" + } + w.Header().Set("Content-Type", "text/event-stream") + emit := func(delta any, finish string) { + event := map[string]any{"id": fmt.Sprintf("branch-kimi-%d", calls), "model": "kimi-k3", "choices": []any{map[string]any{"index": 0, "delta": delta, "finish_reason": finish}}} + if finish != "" { + event["usage"] = map[string]int{"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5} + } + raw, _ := json.Marshal(event) + _, _ = fmt.Fprintf(w, "data: %s\n\n", raw) + } + emit(map[string]any{"role": "assistant", "reasoning_content": fmt.Sprintf("native-private-%d", calls)}, "") + emit(map[string]any{"content": rootActionResponse(domain.RootActionContinue, text, "", "")}, "stop") + _, _ = fmt.Fprint(w, "data: [DONE]\n\n") + })) + st, err := store.Open(filepath.Join(t.TempDir(), "branch-kimi.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + first := newStartedRunForProvider(t, st, "kimi-root", domain.Budget{MaxTurns: 8}) + agent := application.NewAgentRunner(st, router, policy.NewDefaultChecker()).WithModelRetryPolicy(application.ModelRetryPolicy{MaxAttempts: 1}) + if _, err := agent.StepWithInput(t.Context(), first.ID, "BRANCH_ORIGINAL_INTENT"); err != nil { + t.Fatal(err) + } + sourceEvents, err := st.ListRunEvents(t.Context(), first.ID) + if err != nil { + t.Fatal(err) + } + anchor := "" + for _, event := range sourceEvents { + var p struct{ Role, Content string } + _ = json.Unmarshal([]byte(event.PayloadJSON), &p) + if event.Type == events.SessionMessageEvent && p.Role == "assistant" && p.Content == "selected answer" { + anchor = event.EventID + } + } + if anchor == "" { + t.Fatal("source assistant event missing") + } + if _, err := agent.StepWithInput(t.Context(), first.ID, "later source input"); err != nil { + t.Fatal(err) + } + branch, err := application.NewThreadBranchService(st).Branch(t.Context(), domain.ThreadBranchRequest{Version: domain.ThreadBranchProtocolVersion, ThreadID: domain.InitialThreadID(first.ID), SourceEventID: anchor, OperationKey: "kimi-selected-branch", RequestedBy: "test_operator"}) + if err != nil { + t.Fatal(err) + } + if calls != 2 { + t.Fatal("branch creation called the provider") + } + if _, err := application.NewRunService(st).Start(t.Context(), branch.Run.ID); err != nil { + t.Fatal(err) + } + if _, err := agent.StepWithInput(t.Context(), branch.Run.ID, "Continue the selected answer"); err != nil { + t.Fatal("Kimi rejected imported historical context", err) + } + if _, err := agent.StepWithInput(t.Context(), branch.Run.ID, "Continue again"); err != nil { + t.Fatal("Kimi rejected its fresh branch native history", err) + } + if calls != 4 { + t.Fatalf("model calls=%d", calls) + } +} diff --git a/internal/domain/thread_branch.go b/internal/domain/thread_branch.go new file mode 100644 index 000000000..f54adae16 --- /dev/null +++ b/internal/domain/thread_branch.go @@ -0,0 +1,25 @@ +package domain + +const ThreadBranchProtocolVersion = "thread_branch.v1" + +// ThreadBranchRequest names a durable assistant reply, rather than a mutable +// latest-message position or a browser-generated transcript projection. +type ThreadBranchRequest struct { + Version string + ThreadID string + SourceEventID string + OperationKey string + RequestedBy string +} + +type ThreadBranchSource struct { + RunID string +} + +type ThreadBranchRecord struct { + ThreadID string `json:"thread_id"` + RunID string `json:"run_id"` + SessionID string `json:"session_id"` + SourceEventID string `json:"source_event_id"` + Replayed bool `json:"replayed"` +} diff --git a/internal/httpapi/openapi.go b/internal/httpapi/openapi.go index e3021d493..95118a5ca 100644 --- a/internal/httpapi/openapi.go +++ b/internal/httpapi/openapi.go @@ -974,6 +974,11 @@ func openAPIOperationSpecs() []openAPIOperationSpec { DataType: reflect.TypeOf(ThreadCreationControlView{}), RequestType: reflect.TypeOf(ThreadCreationControlRequestView{}), Control: true, Parameters: []openAPIParameter{dockerSandboxIdempotencyKey}}, + {Path: "/api/v1/threads/{thread_id}/branches", Method: http.MethodPost, + OperationID: "branchThread", Summary: "Branch a conversation at a durable assistant reply", Tag: "Control", + Description: "Atomically creates a fresh Thread, Run and Session containing conversation history through source_event_id, including earlier Runs. Historical assistant output is non-authorizing evidence. Requires Thread creation and Session message control. No model or tool is called; approvals and runtime authority are reset. Reusing a key with another reply conflicts.", + DataType: reflect.TypeOf(ThreadBranchView{}), RequestType: reflect.TypeOf(ThreadBranchRequestView{}), Control: true, + SuccessStatus: http.StatusCreated, NotFound: true, Parameters: []openAPIParameter{threadID, dockerSandboxIdempotencyKey}}, {Path: ThreadCreationRequestPath, OperationID: "inspectThreadCreationRequest", Summary: "Observe an original Thread creation request", Tag: "Threads", Description: "Pure read using the original Idempotency-Key and workspace identity. Does not create, replay, admit models, or execute. not_received means absent at this read; an in-flight original request may still arrive. completed confirms only the creation transaction, including after its Run has changed state.", @@ -3769,6 +3774,8 @@ var openAPIFieldEnums = map[string][]string{ "ConfigurationSource.source": {"default", "operator", "project", "snapshot"}, "ConfigurationSource.field": {"budget.max_turns", "budget.max_tokens", "budget.max_tool_calls", "budget.max_cost_usd", "budget.timeout_seconds", "read_only", "allowed_profiles", "exclude_paths", "skill_suggestions", "test_command_id", "format_command_id"}, "ThreadCreationControlRequestView.version": {domain.ThreadCreationProtocolVersion}, + "ThreadBranchRequestView.version": {domain.ThreadBranchProtocolVersion}, + "ThreadBranchView.version": {domain.ThreadBranchProtocolVersion}, "ThreadCreationControlRequestView.profile": {string(domain.ProfileCode), string(domain.ProfileReview), string(domain.ProfileLearn), string(domain.ProfileScript)}, "ThreadCreationControlRequestView.surface": {string(domain.ExecutionSurfaceCode), string(domain.ExecutionSurfaceCyber)}, "ThreadCreationControlRequestView.phase": {string(domain.ExecutionPhasePlan), string(domain.ExecutionPhaseDeliver)}, diff --git a/internal/httpapi/openapi_test.go b/internal/httpapi/openapi_test.go index 37c04bf9f..ddc456597 100644 --- a/internal/httpapi/openapi_test.go +++ b/internal/httpapi/openapi_test.go @@ -1220,6 +1220,23 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { "{candidate_id}": openAPIConfirmCandidate.ID, "{node_id}": openAPICheckpoint.ID, } + _, branchSourceRun, err := application.NewRunService(fixture.store).Create(t.Context(), application.CreateRunRequest{ + Goal: "OpenAPI branch source", Profile: "review", WorkspaceID: fixture.workspace.ID, Interactive: true, + }) + if err != nil { + t.Fatal(err) + } + if _, err := fixture.store.SaveSessionMessage(t.Context(), session.NewMessage(branchSourceRun.SessionID, "user", "Retain the selected conversation")); err != nil { + t.Fatal(err) + } + if _, err := fixture.store.SaveSessionMessage(t.Context(), session.NewMessage(branchSourceRun.SessionID, "assistant", "Selected OpenAPI reply")); err != nil { + t.Fatal(err) + } + branchEvents, err := fixture.store.ListRunEvents(t.Context(), branchSourceRun.ID) + if err != nil { + t.Fatal(err) + } + branchSourceEventID := branchEvents[len(branchEvents)-1].EventID for _, spec := range openAPIOperationSpecs() { requestPath := spec.Path for placeholder, value := range replacements { @@ -1287,6 +1304,8 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { } else if spec.Path == FileEditProposalRecoveryPathTemplate { requestPath = strings.ReplaceAll(spec.Path, "{run_id}", fixture.run.ID) requestPath = strings.ReplaceAll(requestPath, "{edit_id}", recoveryProposal.Edit.ID) + } else if spec.OperationID == "branchThread" { + requestPath = strings.ReplaceAll(spec.Path, "{thread_id}", domain.InitialThreadID(branchSourceRun.ID)) } else if spec.OperationID == "deleteThread" { requestPath = strings.ReplaceAll(spec.Path, "{thread_id}", deleteThread.ID) } else if spec.OperationID == "rejectContextMemoryCandidate" { @@ -1648,6 +1667,8 @@ func TestOpenAPIRoutesMatchAuthenticatedLiveHandlers(t *testing.T) { body = `{"version":"thread_context_compaction.v1","expected_run_id":"` + openAPIThreadRun.ID + `"}` } else if spec.OperationID == "recoverThreadContextCompaction" { body = `{"version":"thread_context_compaction_recovery.v1","expected_run_id":"` + contextRecoveryRun.ID + `","action":"discard"}` + } else if spec.OperationID == "branchThread" { + body = `{"version":"thread_branch.v1","source_event_id":"` + branchSourceEventID + `"}` } else if spec.OperationID == "createThread" { body = `{"version":"thread_creation.v1","goal":"OpenAPI live Thread",` + `"workspace_id":"` + fixture.workspace.ID + `"}` diff --git a/internal/httpapi/thread_branch.go b/internal/httpapi/thread_branch.go new file mode 100644 index 000000000..b6215bee3 --- /dev/null +++ b/internal/httpapi/thread_branch.go @@ -0,0 +1,45 @@ +package httpapi + +import ( + "net/http" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/application" + "cyberagent-workbench/internal/domain" +) + +type ThreadBranchRequestView struct { + Version string `json:"version"` + SourceEventID string `json:"source_event_id"` +} + +type ThreadBranchView struct { + Version string `json:"version"` + Thread ThreadView `json:"thread"` + Run RunView `json:"run"` + Session SessionView `json:"session"` + SourceEventID string `json:"source_event_id"` + Replayed bool `json:"replayed"` +} + +func (a *API) serveThreadBranch(writer http.ResponseWriter, request *http.Request, requestID, threadID, operationKey string) { + var view ThreadBranchRequestView + if !a.decodeThreadControlBody(writer, request, requestID, "Thread branch", &view) { + return + } + store, ok := any(a.store).(application.ThreadBranchStore) + if !ok { + a.writeError(writer, requestID, apperror.New(apperror.CodeFailedPrecondition, "Thread branch storage is unavailable"), 0) + return + } + result, err := application.NewThreadBranchService(store).WithLifecycleHooks(a.lifecycleHooks).Branch(request.Context(), domain.ThreadBranchRequest{ + Version: view.Version, ThreadID: threadID, SourceEventID: view.SourceEventID, OperationKey: operationKey, RequestedBy: "http_thread_operator", + }) + if err != nil { + a.writeError(writer, requestID, err, 0) + return + } + a.writeSuccessStatus(writer, requestID, ThreadBranchView{Version: domain.ThreadBranchProtocolVersion, + Thread: threadView(result.Thread, result.Run.Status), Run: runView(result.Run), Session: sessionView(result.Session), + SourceEventID: result.SourceEventID, Replayed: result.Replayed}, nil, http.StatusCreated) +} diff --git a/internal/httpapi/thread_branch_test.go b/internal/httpapi/thread_branch_test.go new file mode 100644 index 000000000..a2340a13b --- /dev/null +++ b/internal/httpapi/thread_branch_test.go @@ -0,0 +1,91 @@ +package httpapi + +import ( + "bytes" + "encoding/json" + "net/http" + "testing" + + "cyberagent-workbench/internal/application" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/session" +) + +func TestThreadBranchHTTPAuthorizationContractAndIndependentContinuation(t *testing.T) { + f := newAPIFixture(t) + _, run, err := application.NewRunService(f.store).Create(t.Context(), application.CreateRunRequest{Goal: "branch HTTP", WorkspaceID: f.workspace.ID, Profile: "review", Interactive: true}) + if err != nil { + t.Fatal(err) + } + if _, err := f.store.SaveSessionMessage(t.Context(), session.NewMessage(run.SessionID, "user", "preserve this requirement")); err != nil { + t.Fatal(err) + } + if _, err := f.store.SaveSessionMessage(t.Context(), session.NewMessage(run.SessionID, "assistant", "selected reply")); err != nil { + t.Fatal(err) + } + list, err := f.store.ListRunEvents(t.Context(), run.ID) + if err != nil { + t.Fatal(err) + } + var anchor string + for _, e := range list { + if e.Type == events.SessionMessageEvent { + var p struct{ Role string } + _ = json.Unmarshal([]byte(e.PayloadJSON), &p) + if p.Role == "assistant" { + anchor = e.EventID + } + } + } + path := ThreadCollectionPath + "/" + domain.InitialThreadID(run.ID) + "/branches" + body, _ := json.Marshal(ThreadBranchRequestView{Version: domain.ThreadBranchProtocolVersion, SourceEventID: anchor}) + for _, test := range []struct { + name, token, key string + creation, messages bool + status int + }{ + {"without control", testAccessToken, "branch-http-control", true, true, http.StatusUnauthorized}, + {"without creation", testControlToken, "branch-http-creation", false, true, http.StatusNotFound}, + {"without messages", testControlToken, "branch-http-messages", true, false, http.StatusNotFound}, + {"without key", testControlToken, "", true, true, http.StatusBadRequest}, + } { + t.Run(test.name, func(t *testing.T) { + api, err := New(f.store, Config{AccessToken: testAccessToken, ControlToken: testControlToken, RunCreationEnabled: test.creation, SessionMessageEnabled: test.messages}) + if err != nil { + t.Fatal(err) + } + response := performSessionMessageRequest(t, api, http.MethodPost, path, test.token, test.key, "application/json", bytes.NewReader(body)) + if response.Code != test.status { + t.Fatalf("status=%d body=%s", response.Code, response.Body) + } + }) + } + api, err := New(f.store, Config{AccessToken: testAccessToken, ControlToken: testControlToken, RunCreationEnabled: true, SessionMessageEnabled: true}) + if err != nil { + t.Fatal(err) + } + response := performSessionMessageRequest(t, api, http.MethodPost, path, testControlToken, "branch-http-original", "application/json", bytes.NewReader(body)) + var result ThreadBranchView + decodeDataStatus(t, response, http.StatusCreated, &result) + if result.Version != domain.ThreadBranchProtocolVersion || result.Thread.ID == domain.InitialThreadID(run.ID) || result.SourceEventID != anchor || result.Replayed || result.Run.Status != "created" { + t.Fatalf("branch response=%+v", result) + } + response = performSessionMessageRequest(t, api, http.MethodPost, path, testControlToken, "branch-http-original", "application/json", bytes.NewReader(body)) + var replay ThreadBranchView + decodeDataStatus(t, response, http.StatusCreated, &replay) + if !replay.Replayed || replay.Thread.ID != result.Thread.ID { + t.Fatal("HTTP retry created another branch") + } + message := []byte(`{"version":"thread_message_submission.v1","content":"continue this new conversation"}`) + response = performSessionMessageRequest(t, api, http.MethodPost, ThreadCollectionPath+"/"+result.Thread.ID+"/messages", testControlToken, "branch-http-continue", "application/json", bytes.NewReader(message)) + var continued ThreadMessageControlView + decodeDataStatus(t, response, http.StatusAccepted, &continued) + if continued.Thread.ID != result.Thread.ID || continued.RunID != result.Run.ID { + t.Fatal("continuation did not target the branch") + } + response = performSessionMessageRequest(t, api, http.MethodPost, path, testControlToken, "branch-http-unknown-field", "application/json", bytes.NewBufferString(`{"version":"thread_branch.v1","source_event_id":"`+anchor+`","execution_permission":"full_access"}`)) + if response.Code != http.StatusBadRequest { + t.Fatal("renderer supplied authority fields were accepted") + } +} diff --git a/internal/httpapi/thread_control.go b/internal/httpapi/thread_control.go index 58baeefe4..60ea8c121 100644 --- a/internal/httpapi/thread_control.go +++ b/internal/httpapi/thread_control.go @@ -570,7 +570,7 @@ func matchThreadMutationPath(requestPath string) (string, string, bool) { return "", "", false } switch parts[1] { - case "messages", "turns", "interrupt", "recovery", "archive", "restore", "delete": + case "messages", "turns", "branches", "interrupt", "recovery", "archive", "restore", "delete": return parts[0], parts[1], true default: return "", "", false @@ -745,6 +745,10 @@ func (a *API) serveThreadMutationControl(writer http.ResponseWriter, request *ht a.writeError(writer, requestID, err, 0) return } + if action == "branches" { + a.serveThreadBranch(writer, request, requestID, threadID, operationKey) + return + } if action == "interrupt" { controller, ok := a.threadTurnController.(ThreadExecutionController) if !ok { diff --git a/internal/store/thread_branch.go b/internal/store/thread_branch.go new file mode 100644 index 000000000..45512048a --- /dev/null +++ b/internal/store/thread_branch.go @@ -0,0 +1,319 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strconv" + "strings" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/events" + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/runmutation" + "cyberagent-workbench/internal/session" +) + +type branchAnchor struct { + runID string + ordinal, sequence int64 + eventType, payload string +} + +func readBranchAnchor(ctx context.Context, q historyQueryer, request domain.ThreadBranchRequest) (branchAnchor, error) { + var anchor branchAnchor + err := q.QueryRowContext(ctx, `SELECT e.run_id,b.ordinal,e.sequence,e.type,e.payload_json + FROM run_events e JOIN thread_runs b ON b.run_id=e.run_id JOIN threads t ON t.id=b.thread_id + WHERE t.id=? AND t.status!='deleted' AND e.event_id=?`, request.ThreadID, request.SourceEventID). + Scan(&anchor.runID, &anchor.ordinal, &anchor.sequence, &anchor.eventType, &anchor.payload) + if errors.Is(err, sql.ErrNoRows) { + return anchor, apperror.New(apperror.CodeNotFound, "The selected reply is unavailable in this Thread") + } + if err != nil { + return anchor, err + } + var payload struct { + Role, Content, Text string + SourceKind string `json:"source_kind"` + } + if json.Unmarshal([]byte(anchor.payload), &payload) != nil { + return anchor, apperror.New(apperror.CodeFailedPrecondition, "The selected reply has invalid stored content") + } + if (anchor.eventType == events.SessionMessageEvent && payload.Role == "assistant" && payload.SourceKind == session.SourceModelResponse && strings.TrimSpace(payload.Content) != "") || + (anchor.eventType == events.ModelPublicCommentaryEvent && strings.TrimSpace(payload.Text) != "") { + return anchor, nil + } + return anchor, apperror.New(apperror.CodeInvalidArgument, "Branching requires a durable assistant reply") +} + +func (s *SQLiteStore) GetThreadBranchSource(ctx context.Context, request domain.ThreadBranchRequest) (domain.ThreadBranchSource, error) { + a, err := readBranchAnchor(ctx, s.db, request) + return domain.ThreadBranchSource{RunID: a.runID}, err +} + +func branchOperationIdentity(r domain.ThreadBranchRequest) (string, string) { + return runmutation.Fingerprint("thread_branch_operation", r.ThreadID, r.OperationKey), + runmutation.Fingerprint(domain.ThreadBranchProtocolVersion, r.ThreadID, r.SourceEventID, r.RequestedBy) +} + +func readBranchOperation(ctx context.Context, q historyQueryer, r domain.ThreadBranchRequest) (domain.ThreadBranchRecord, bool, error) { + key, fingerprint := branchOperationIdentity(r) + var raw string + err := q.QueryRowContext(ctx, `SELECT payload_json FROM thread_events WHERE thread_id=? + AND type='thread.branched' AND source='thread_branch' + AND json_extract(payload_json,'$.operation_key_digest')=?`, r.ThreadID, key).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return domain.ThreadBranchRecord{}, false, nil + } + if err != nil { + return domain.ThreadBranchRecord{}, false, err + } + var stored struct { + Record domain.ThreadBranchRecord `json:"record"` + Fingerprint string `json:"request_fingerprint"` + } + if json.Unmarshal([]byte(raw), &stored) != nil || stored.Record.ThreadID == "" || stored.Fingerprint != fingerprint { + return domain.ThreadBranchRecord{}, false, apperror.New(apperror.CodeConflict, "This branch key was already used for a different reply") + } + stored.Record.Replayed = true + return stored.Record, true, nil +} + +func (s *SQLiteStore) GetThreadBranchOperation(ctx context.Context, r domain.ThreadBranchRequest) (domain.ThreadBranchRecord, bool, error) { + return readBranchOperation(ctx, s.db, r) +} + +// The existing write transaction serializes replay lookup, creation and the +// immutable branch receipt, including projectless conversations. +func (s *SQLiteStore) CreateThreadBranch(ctx context.Context, request domain.ThreadBranchRequest, + mission domain.Mission, run domain.Run, mode domain.RunModeSnapshot, sess session.Session, initial []events.Event, +) (domain.ThreadBranchRecord, error) { + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{}) + if err != nil { + return domain.ThreadBranchRecord{}, err + } + defer func() { _ = tx.Rollback() }() + if stored, found, err := readBranchOperation(ctx, tx, request); err != nil { + return domain.ThreadBranchRecord{}, err + } else if found { + return stored, tx.Commit() + } + anchor, err := readBranchAnchor(ctx, tx, request) + if err != nil { + return domain.ThreadBranchRecord{}, err + } + var workspace string + if err := tx.QueryRowContext(ctx, `SELECT COALESCE(workspace_id,'') FROM threads WHERE id=?`, request.ThreadID).Scan(&workspace); err != nil { + return domain.ThreadBranchRecord{}, err + } + if mission.WorkspaceID != workspace || sess.WorkspaceID != workspace || run.Config.ContinuityContextFingerprint != "" || len(run.Config.ContinuityContext) > 0 { + return domain.ThreadBranchRecord{}, apperror.New(apperror.CodeInvalidArgument, "Branch creation must retain its Workspace and use fresh context") + } + if err := createMissionRunTx(ctx, tx, mission, run, mode, sess, true, initial); err != nil { + return domain.ThreadBranchRecord{}, err + } + if err := copyBranchMessagesTx(ctx, tx, request, anchor, run); err != nil { + return domain.ThreadBranchRecord{}, err + } + record := domain.ThreadBranchRecord{ThreadID: domain.InitialThreadID(run.ID), RunID: run.ID, SessionID: sess.ID, SourceEventID: request.SourceEventID} + key, fingerprint := branchOperationIdentity(request) + payload, _ := json.Marshal(map[string]any{"version": domain.ThreadBranchProtocolVersion, "operation_key_digest": key, + "request_fingerprint": fingerprint, "source_event_id": request.SourceEventID, "record": record}) + if _, err := tx.ExecContext(ctx, `INSERT INTO thread_events(thread_id,run_id,type,source,payload_json,created_at) + VALUES(?,?,'thread.branched','thread_branch',?,?)`, request.ThreadID, anchor.runID, string(payload), ts(time.Now().UTC())); err != nil { + return domain.ThreadBranchRecord{}, err + } + return record, tx.Commit() +} + +type branchMessage struct { + message session.Message + steeringID string + eventID string + attemptID string +} + +func copyBranchMessagesTx(ctx context.Context, tx *sql.Tx, request domain.ThreadBranchRequest, anchor branchAnchor, run domain.Run) error { + // Order committed operator input at its queue event, as the transcript does. + // A live commentary may precede Session commit; include only its delivered + // input, never unrelated queued messages waiting for a subsequent turn. + rows, err := tx.QueryContext(ctx, `SELECT m.id,m.session_id,m.role,m.content,m.provenance_version,m.source_kind,m.source_ref, + m.content_sha256,m.instruction_authorized,m.token_estimate,m.compacted,m.created_at,COALESCE(o.id,''),b.ordinal, + COALESCE(queued.sequence,e.sequence) AS position,e.event_id,'' + FROM thread_runs b JOIN session_messages m ON m.session_id=b.session_id + JOIN run_events e ON e.run_id=b.run_id AND e.type='session.message_created' AND e.subject_id=CAST(m.id AS TEXT) + LEFT JOIN operator_steering_messages o ON o.session_message_id=m.id AND o.run_id=b.run_id + LEFT JOIN run_events queued ON queued.run_id=b.run_id AND queued.type='operator.steering_queued' AND queued.subject_id=o.id + WHERE b.thread_id=? AND m.role!='system' + AND NOT EXISTS(SELECT 1 FROM operator_message_attachment_evidence attachment WHERE attachment.session_message_id=m.id) + AND (b.ordinal Date: Mon, 19 Oct 2026 13:29:14 +0800 Subject: [PATCH 2/2] Add compact copy and branch actions beneath assistant replies --- web/src/v2/app.test.tsx | 38 ++++++++++++- web/src/v2/app.tsx | 4 ++ web/src/v2/components/conversation.test.tsx | 56 +++++++++++++++++- web/src/v2/components/conversation.tsx | 18 +++++- web/src/v2/components/narrative.test.tsx | 48 +++++++++++++++- web/src/v2/components/narrative.tsx | 63 +++++++++++++++++---- web/src/v2/projection/narrative.test.ts | 9 ++- web/src/v2/projection/narrative.ts | 5 ++ web/src/v2/styles.css | 52 ++++++++++++----- 9 files changed, 259 insertions(+), 34 deletions(-) diff --git a/web/src/v2/app.test.tsx b/web/src/v2/app.test.tsx index 880e0c72d..e235830b5 100644 --- a/web/src/v2/app.test.tsx +++ b/web/src/v2/app.test.tsx @@ -22,19 +22,25 @@ function LocaleSwitch() { return ; } +const branchNavigation = vi.hoisted(() => ({ complete: undefined as ((thread: ThreadView) => void) | undefined })); + vi.mock("./components/conversation", () => ({ - V2Conversation: ({ threadID, view = "conversation", draft, onDraftChange, reviewEntry }: { + V2Conversation: ({ threadID, view = "conversation", draft, onDraftChange, reviewEntry, onBranchCreated }: { threadID: string; view?: "conversation" | "inspector"; draft: string; onDraftChange: (value: string) => void; reviewEntry?: { threadID: string; runID: string }; - }) => ( + onBranchCreated?: (thread: ThreadView) => void; + }) => { + branchNavigation.complete = onBranchCreated; + return (
{threadID}