diff --git a/api/openapi.yaml b/api/openapi.yaml index 632aea36..f9c90305 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -300,6 +300,41 @@ components: - issue_uid - at type: object + CloseActionRequestBody: + additionalProperties: false + properties: + actor: + type: string + dry_run: + type: boolean + evidence: + items: + $ref: "#/components/schemas/Evidence" + type: + - array + - "null" + message: + type: string + reason: + enum: + - done + - wontfix + - duplicate + - superseded + - audit-no-change + - "" + type: string + retry_protocol: + enum: + - close-v1 + - "" + type: string + source: + enum: + - tui + - "" + type: string + type: object Comment: additionalProperties: true properties: @@ -4595,7 +4630,7 @@ components: type: object info: title: kata - version: 0.14.0 + version: 0.15.0 openapi: 3.1.0 paths: /api/v1/audit/closes: @@ -6275,11 +6310,19 @@ paths: required: true schema: type: string + - in: header + name: Idempotency-Key + schema: + type: string + - in: header + name: If-Match + schema: + type: string requestBody: content: application/json: schema: - $ref: "#/components/schemas/ActionRequestBody" + $ref: "#/components/schemas/CloseActionRequestBody" required: true responses: "200": diff --git a/cmd/kata/api_compat_test.go b/cmd/kata/api_compat_test.go index 9f5d19ee..7a5e5737 100644 --- a/cmd/kata/api_compat_test.go +++ b/cmd/kata/api_compat_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "net/http" "net/http/httptest" "sync/atomic" @@ -86,6 +87,45 @@ func TestFilteredListAllRejectsDaemonBeforeGlobalListFilters(t *testing.T) { assert.Zero(t, listCalls.Load(), "the unfiltered old endpoint must not be queried") } +func TestCloseRetryFlagsMakeOldDaemonRejectCloseBeforeMutation(t *testing.T) { + var closeCalls, mutations atomic.Int32 + var retryProtocol string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v1/projects/resolve": + _, _ = w.Write([]byte(`{"project":{"id":1,"name":"example-project"}}`)) + case "/api/v1/projects/1/issues/abc1/actions/close": + closeCalls.Add(1) + var body map[string]any + require.NoError(t, json.NewDecoder(r.Body).Decode(&body)) + retryProtocol, _ = body["retry_protocol"].(string) + if retryProtocol != "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"status":400,"error":{"code":"validation","message":"retry_protocol: unexpected property"}}`)) + return + } + mutations.Add(1) + _, _ = w.Write([]byte(`{"changed":true}`)) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(server.Close) + + _, _, err := executeRootCapture(t, + contextWithBaseURL(context.Background(), server.URL), + "--project", "example-project", "close", "abc1", + "--wontfix", + "--message", "Reviewed the request and recorded why the work should stop here.", + "--idempotency-key", "close-request-1") + require.Error(t, err) + assert.Contains(t, err.Error(), "retry_protocol") + assert.Equal(t, "close-v1", retryProtocol) + assert.Equal(t, int32(1), closeCalls.Load()) + assert.Zero(t, mutations.Load(), "the legacy request schema must reject before mutation") +} + func TestListAllDefaultsToUnlimited(t *testing.T) { var sentLimit atomic.Bool server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/cmd/kata/close.go b/cmd/kata/close.go index 433c3e2b..aea19686 100644 --- a/cmd/kata/close.go +++ b/cmd/kata/close.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "maps" "net/http" @@ -14,10 +15,12 @@ import ( func newCloseCmd() *cobra.Command { var ( - reason string - message string - evidence []string - dryRun bool + reason string + message string + evidence []string + dryRun bool + idempotencyKey string + ifMatch string sugarDone bool sugarWontfix bool @@ -50,6 +53,16 @@ Instead, label and comment: kata comment --body "what was attempted, what remains"`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { + if cmd.Flags().Changed("idempotency-key") && strings.TrimSpace(idempotencyKey) == "" { + return &cliError{ + Message: "--idempotency-key must not be blank", + Kind: kindValidation, + ExitCode: ExitValidation, + } + } + if err := validateMetaIfMatchFlag(cmd, ifMatch); err != nil { + return err + } // Resolve sugar -> reason (with conflict checks). Multiple sugar // flags are mutually exclusive: a sequential switch would silently // keep the first match and drop the rest (`--done --wontfix` would @@ -127,7 +140,18 @@ Instead, label and comment: if dryRun && currentOutputMode() == outputHuman && !flags.Quiet { _, _ = fmt.Fprintln(cmd.ErrOrStderr(), "close: dry-run (no mutations will occur)") } - return runAction(cmd, args[0], "close", extra) + headers := map[string]string{} + if idempotencyKey != "" { + headers["Idempotency-Key"] = idempotencyKey + } + if strings.TrimSpace(ifMatch) != "" { + etag, err := normalizeMetaIfMatch(ifMatch) + if err != nil { + return err + } + headers["If-Match"] = etag + } + return runActionWithHeaders(cmd, args[0], "close", extra, headers) }, } cmd.Flags().StringVar(&reason, "reason", "", @@ -144,6 +168,10 @@ Instead, label and comment: "duplicate-of:, superseded-by:") cmd.Flags().BoolVar(&dryRun, "dry-run", false, "validate without mutating; reports the would-be close event") + cmd.Flags().StringVar(&idempotencyKey, "idempotency-key", "", + "send Idempotency-Key header for safe retry") + cmd.Flags().StringVar(&ifMatch, "if-match", "", + "expected issue revision (N or rev-N)") cmd.Flags().BoolVar(&sugarDone, "done", false, "sugar for --reason done") cmd.Flags().BoolVar(&sugarWontfix, "wontfix", false, "sugar for --reason wontfix") @@ -213,6 +241,12 @@ func parseEvidenceFlags(raw []string) ([]api.Evidence, error) { // the daemon action endpoint. If --comment was passed on the command, the // comment is appended in a separate POST after the action succeeds. func runAction(cmd *cobra.Command, raw, action string, extra map[string]any) error { + return runActionWithHeaders(cmd, raw, action, extra, nil) +} + +func runActionWithHeaders( + cmd *cobra.Command, raw, action string, extra map[string]any, headers map[string]string, +) error { comment, err := commentFromFlag(cmd) if err != nil { return err @@ -224,20 +258,41 @@ func runAction(cmd *cobra.Command, raw, action string, extra map[string]any) err actor, _ := resolveActor(ctx, flags.As, nil) body := map[string]any{"actor": actor} maps.Copy(body, extra) + if action == "close" && len(headers) > 0 { + body["retry_protocol"] = api.CloseRetryProtocol + } client, err := httpClientFor(ctx, baseURL) if err != nil { return err } - status, bs, err := httpDoJSON(ctx, client, http.MethodPost, + status, bs, err := httpDoJSONHeaders(ctx, client, http.MethodPost, fmt.Sprintf("%s/api/v1/projects/%d/issues/%s/actions/%s", baseURL, pid, url.PathEscape(issue.RefForAPI), action), - body) + body, headers) if err != nil { return err } if status >= 400 { return apiErrFromBody(status, bs) } - if err := postFollowupComment(ctx, client, baseURL, pid, issue.RefForAPI, actor, comment); err != nil { + commentKey := "" + if key := headers["Idempotency-Key"]; key != "" { + commentKey = "close-comment:" + key + } + commentProjectID, commentIssueRef := pid, issue.RefForAPI + if comment != "" && commentKey != "" { + var response api.MutationResponse + if err := json.Unmarshal(bs, &response.Body); err != nil { + return fmt.Errorf("decode close response for follow-up comment: %w", err) + } + if response.Body.Issue.ProjectID <= 0 || response.Body.Issue.UID == "" { + return fmt.Errorf("close response is missing the issue identity for the follow-up comment") + } + commentProjectID = response.Body.Issue.ProjectID + commentIssueRef = response.Body.Issue.UID + } + if err := postFollowupCommentWithKey( + ctx, client, baseURL, commentProjectID, commentIssueRef, actor, comment, commentKey, + ); err != nil { return err } return printMutation(cmd, bs) diff --git a/cmd/kata/close_reopen_test.go b/cmd/kata/close_reopen_test.go index 5fac5064..23cb4c21 100644 --- a/cmd/kata/close_reopen_test.go +++ b/cmd/kata/close_reopen_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.kenn.io/kata/internal/db" ) func TestCloseReopen_RoundTrip(t *testing.T) { @@ -57,6 +58,51 @@ func TestClose_AgentOutputExternalEvidence(t *testing.T) { assert.Contains(t, audit, `"evidence_types":["external"]`) } +func TestCloseCmd_RetryFlagsReplayOriginalReceipt(t *testing.T) { + env, dir, _, ref := setupWorkspaceWithIssue(t, "test issue") + args := []string{ + "--json", "close", ref, + "--done", + "--message", "Implemented the requested behavior and ran the focused tests.", + "--test", "go test ./cmd/kata", + "--idempotency-key", "close-request-1", + "--if-match", "1", + } + first := runCLI(t, env, dir, args...) + assert.Contains(t, first, `"changed":true`) + + second := runCLI(t, env, dir, args...) + assert.Contains(t, second, `"changed":false`) + assert.Contains(t, second, `"reused":true`) + assert.Contains(t, second, `"original_event":`) +} + +func TestCloseCmd_RejectsBlankIfMatch(t *testing.T) { + env, dir, _, ref := setupWorkspaceWithIssue(t, "test issue") + _, stderr, err := runCLIWithErr(t, env, dir, + "close", ref, + "--done", + "--message", "Implemented the requested behavior and ran the focused tests.", + "--test", "go test ./cmd/kata", + "--if-match", "") + require.Error(t, err) + assert.Contains(t, stderr, "--if-match must not be blank") +} + +func TestCloseCmd_RejectsBlankIdempotencyKey(t *testing.T) { + env, dir, _, ref := setupWorkspaceWithIssue(t, "test issue") + _, stderr, err := runCLIWithErr(t, env, dir, + "close", ref, + "--wontfix", + "--message", "Reviewed the request and recorded why the work should stop here.", + "--idempotency-key", " ") + require.Error(t, err) + assert.Contains(t, stderr, "--idempotency-key must not be blank") + + show := runCLI(t, env, dir, "show", ref, "--json") + assert.Contains(t, show, `"status":"open"`) +} + func TestClose_AgentDryRunSuppressesHumanBanner(t *testing.T) { env, dir, _, ref := setupWorkspaceWithIssue(t, "test issue") @@ -404,6 +450,53 @@ func TestClose_WithComment_AppendsComment(t *testing.T) { assert.Equal(t, "closed", got.Issue.Status) } +func TestClose_RetryWithCommentAppendsOneComment(t *testing.T) { + env, dir, pid, ref := setupWorkspaceWithIssue(t, "test issue") + args := []string{ + "close", ref, + "--done", + "--message", "Implemented the requested behavior and ran the focused tests.", + "--test", "go test ./cmd/kata", + "--comment", "fixed in abc1234", + "--idempotency-key", "close-request-with-comment-1", + } + runCLI(t, env, dir, args...) + runCLI(t, env, dir, args...) + + got := fetchIssueViaHTTPWithComments(t, env, pid, ref) + require.Len(t, got.Comments, 1) + assert.Equal(t, "fixed in abc1234", got.Comments[0].Body) +} + +func TestClose_RetryWithCommentAfterMoveAppendsOneComment(t *testing.T) { + env, dir, sourceProjectID, ref := setupWorkspaceWithIssue(t, "test issue") + args := []string{ + "close", ref, + "--done", + "--message", "Implemented the requested behavior and ran the focused tests.", + "--test", "go test ./cmd/kata", + "--comment", "fixed in abc1234", + "--idempotency-key", "close-request-with-comment-then-move", + } + runCLI(t, env, dir, args...) + issue, err := env.DB.IssueByShortID(t.Context(), sourceProjectID, ref, db.IncludeDeletedNo) + require.NoError(t, err) + target, err := env.DB.CreateProject(t.Context(), "target-project") + require.NoError(t, err) + _, err = env.DB.MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: sourceProjectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + + runCLI(t, env, dir, args...) + + comments, err := env.DB.CommentsByIssue(t.Context(), issue.ID) + require.NoError(t, err) + require.Len(t, comments, 1) + assert.Equal(t, "fixed in abc1234", comments[0].Body) +} + func TestReopen_WithComment_AppendsComment(t *testing.T) { env, dir, pid, ref := setupWorkspaceWithIssue(t, "test issue") runCLI(t, env, dir, "close", ref, diff --git a/cmd/kata/comment_flag.go b/cmd/kata/comment_flag.go index f9df0e08..1ea554e8 100644 --- a/cmd/kata/comment_flag.go +++ b/cmd/kata/comment_flag.go @@ -47,21 +47,39 @@ func postFollowupComment( baseURL string, projectID int64, issueRef, actor, body string, +) error { + return postFollowupCommentWithKey(ctx, client, baseURL, projectID, issueRef, actor, body, "") +} + +func postFollowupCommentWithKey( + ctx context.Context, + client *http.Client, + baseURL string, + projectID int64, + issueRef, actor, body, idempotencyKey string, ) error { if body == "" { return nil } - status, bs, err := httpDoJSON(ctx, client, http.MethodPost, + headers := map[string]string{} + if idempotencyKey != "" { + headers["Idempotency-Key"] = idempotencyKey + } + retryInstruction := fmt.Sprintf("retry with: kata comment %s --body ...", issueRef) + if idempotencyKey != "" { + retryInstruction = "rerun the original kata close command with the same --idempotency-key" + } + status, bs, err := httpDoJSONWithHeader(ctx, client, http.MethodPost, fmt.Sprintf("%s/api/v1/projects/%d/issues/%s/comments", baseURL, projectID, url.PathEscape(issueRef)), - map[string]any{"actor": actor, "body": body}) + headers, map[string]any{"actor": actor, "body": body}) if err != nil { return fmt.Errorf("issue mutation succeeded but appending --comment failed: %w "+ - "(retry with: kata comment %s --body ...)", err, issueRef) + "(%s)", err, retryInstruction) } if status >= 400 { base := apiErrFromBody(status, bs) return fmt.Errorf("issue mutation succeeded but appending --comment failed: %w "+ - "(retry with: kata comment %s --body ...)", base, issueRef) + "(%s)", base, retryInstruction) } return nil } diff --git a/cmd/kata/comment_test.go b/cmd/kata/comment_test.go index be2e352d..8deb4c22 100644 --- a/cmd/kata/comment_test.go +++ b/cmd/kata/comment_test.go @@ -2,6 +2,8 @@ package main import ( "context" + "net/http" + "net/http/httptest" "strings" "testing" @@ -9,6 +11,18 @@ import ( "github.com/stretchr/testify/require" ) +func TestPostFollowupCommentFailureRecommendsSafeKeyedRetry(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { + http.Error(writer, "temporary failure", http.StatusServiceUnavailable) + })) + t.Cleanup(server.Close) + + err := postFollowupCommentWithKey(t.Context(), server.Client(), server.URL, + 1, "abc1", "example-agent", "finished work", "close-comment:close-request-1") + require.Error(t, err) + assert.Contains(t, err.Error(), "rerun the original kata close command with the same --idempotency-key") +} + func TestComment_AppendsToIssue(t *testing.T) { env, dir := setupCLIEnv(t) short := createIssueViaHTTP(t, env, dir, "x") diff --git a/docs/reference/cli.md b/docs/reference/cli.md index c1384502..4eb9d2c0 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -282,9 +282,21 @@ kata close --done --message \ [--pr ] \ [--test ] \ [--reviewed ] \ - [--evidence ] + [--evidence ] \ + [--idempotency-key ] \ + [--if-match ] ``` +`--idempotency-key` makes a close safe to retry after a lost response. For +seven days, an exact retry returns the original `issue.closed` event through +`original_event` and does not close the issue again. Reusing the key with a +different issue, actor, close payload, or revision returns a conflict. + +`--if-match` accepts `7` or `rev-7`. The daemon checks that revision inside the +close transaction and returns a revision conflict when it has changed. An +exact retry with a matching idempotency key returns the committed receipt even +after the original close advanced the issue state. + Evidence is validated against the close reason: | Reason | Evidence rule | diff --git a/docs/reference/http-api.md b/docs/reference/http-api.md index 1d3cd570..d331c01f 100644 --- a/docs/reference/http-api.md +++ b/docs/reference/http-api.md @@ -45,7 +45,7 @@ The schema carries a version in its `info.version` field { "ok": true, "schema_version": 7, - "api_schema_version": "0.14.0", + "api_schema_version": "0.15.0", "version": "1.4.2", "uptime": "5m0s", "db_path": "/path/to/kata.db", @@ -95,6 +95,11 @@ once at startup and requires API `0.11.0`, because its relationship tools always send the pinned-target fields and its close-audit paging relies on `event_id`. +Guarded close requests use a request-local compatibility check instead of a +separate health probe. A request that sends `Idempotency-Key` or `If-Match` +also sends `retry_protocol: "close-v1"`. Current daemons require the marker; +older daemons reject the unknown field before the close can run. + The field is **optional in the schema** even though current daemons always send it. That is deliberate: a version-detection field has to survive version skew, so a client generated from a schema that includes it can still parse the @@ -107,6 +112,7 @@ and decline to render issue detail. | Version | Change | | --- | --- | +| `0.15.0` | Added close idempotency and revision headers, the `close-v1` request marker, and retry receipt fields. | | `0.14.0` | Added `external` close evidence with its required `account` field. | | `0.13.0` | Added optimistic revision guards to issue and project metadata patch requests. | | `0.12.0` | Added the optional `idle_shutdown` health block for effective auto-start idle shutdown state and capability discovery. | diff --git a/docs/reference/mcp.md b/docs/reference/mcp.md index cb397822..3f4bf403 100644 --- a/docs/reference/mcp.md +++ b/docs/reference/mcp.md @@ -171,7 +171,10 @@ The tools use structured input and output. List-like results, including timestamps cannot skip or repeat rows and a project merge or issue purge during pagination fails the page with a restart error instead of silently skewing it. `kata.show` returns at most 100 comments. Create and -comment require idempotency keys. `kata.token_create`, recurrence creation, +comment require idempotency keys. `kata.close` accepts an optional +`idempotency_key` for safe retries and an optional `revision` for a conditional +close. An exact retry returns the original close event in the mutation output. +`kata.token_create`, recurrence creation, `kata.storage_import`, `kata.lease`, and `kata.sync_once` are annotated non-idempotent: the first two mint a new record on every identical retry, a forced storage import replaces the target again (with a fresh instance diff --git a/internal/api/metadata_resolvers.go b/internal/api/metadata_resolvers.go index 514d6b26..f0e3cad8 100644 --- a/internal/api/metadata_resolvers.go +++ b/internal/api/metadata_resolvers.go @@ -40,3 +40,12 @@ func (*PatchProjectMetadataRequest) Resolve(ctx huma.Context) []error { } return nil } + +// Resolve rejects an empty close revision guard before the action can be +// mistaken for an unconditional close. +func (*CloseActionRequest) Resolve(ctx huma.Context) []error { + if ifMatchPresentButEmpty(ctx) { + return []error{NewError(400, "validation", "If-Match header required", "", nil)} + } + return nil +} diff --git a/internal/api/types.go b/internal/api/types.go index f45e3283..f241b0ec 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -8,6 +8,10 @@ import ( "go.kenn.io/kata/internal/db" ) +// CloseRetryProtocol marks a close request whose retry headers must be +// understood by the receiving daemon before it may mutate an issue. +const CloseRetryProtocol = "close-v1" + // PingResponse mirrors the cheapest liveness response. type PingResponse struct { Body struct { @@ -822,7 +826,8 @@ type CommentResponse struct { } } -// ActionRequest is POST /api/v1/projects/{id}/issues/{ref}/actions/close|reopen. +// ActionRequest is POST /api/v1/projects/{id}/issues/{ref}/actions/reopen. +// CloseActionRequest adds the close-only retry and revision headers. // Reason is enforced to the schema's CHECK list so unsupported values surface // as 400 validation rather than a SQLite constraint failure (500 internal). // Message, Evidence, and DryRun are close-only inputs (anti-agent-justification); @@ -830,23 +835,43 @@ type CommentResponse struct { type ActionRequest struct { ProjectID int64 `path:"project_id" required:"true"` Ref string `path:"ref" required:"true"` - Body struct { - Actor string `json:"actor,omitempty"` - Reason string `json:"reason,omitempty" enum:"done,wontfix,duplicate,superseded,audit-no-change,"` - Message string `json:"message,omitempty"` - // Source signals the caller's UI surface. "tui" relaxes the - // substance / evidence validation so an interactive human close - // is one keystroke, but only over an owner-local Unix socket or - // direct, unforwarded loopback TCP connection. Forwarded TCP, - // non-loopback TCP, and identity-backed, trusted-proxy, or browser - // principals get full validation even when they send source="tui". - // Structural guards - // (parent-close, sibling throttle) always apply. Empty string means - // "agent / CLI" and gets full validation. - Source string `json:"source,omitempty" enum:"tui,"` - Evidence []Evidence `json:"evidence,omitempty"` - DryRun bool `json:"dry_run,omitempty"` - } + Body ActionRequestBody +} + +// CloseActionRequest adds retry and optimistic-concurrency headers to the +// shared issue action body. +type CloseActionRequest struct { + ProjectID int64 `path:"project_id" required:"true"` + Ref string `path:"ref" required:"true"` + IdempotencyKey string `header:"Idempotency-Key"` + IfMatch string `header:"If-Match"` + Body CloseActionRequestBody +} + +// CloseActionRequestBody extends the legacy action body with a close-only +// protocol marker. Older daemons reject the unknown marker before they can +// ignore retry headers and mutate an issue. +type CloseActionRequestBody struct { + ActionRequestBody + RetryProtocol string `json:"retry_protocol,omitempty" enum:"close-v1,"` +} + +// ActionRequestBody is the shared JSON body for close and reopen actions. +type ActionRequestBody struct { + Actor string `json:"actor,omitempty"` + Reason string `json:"reason,omitempty" enum:"done,wontfix,duplicate,superseded,audit-no-change,"` + Message string `json:"message,omitempty"` + // Source signals the caller's UI surface. "tui" relaxes the + // substance / evidence validation so an interactive human close + // is one keystroke, but only over an owner-local Unix socket or + // direct, unforwarded loopback TCP connection. Forwarded TCP, + // non-loopback TCP, and identity-backed, trusted-proxy, or browser + // principals get full validation even when they send source="tui". + // Structural guards (parent-close, sibling throttle) always apply. + // Empty string means "agent / CLI" and gets full validation. + Source string `json:"source,omitempty" enum:"tui,"` + Evidence []Evidence `json:"evidence,omitempty"` + DryRun bool `json:"dry_run,omitempty"` } // CreateLinkRequest is POST /api/v1/projects/{id}/issues/{ref}/links. diff --git a/internal/daemon/handlers_actions.go b/internal/daemon/handlers_actions.go index 3fa82853..6d580ce9 100644 --- a/internal/daemon/handlers_actions.go +++ b/internal/daemon/handlers_actions.go @@ -2,6 +2,9 @@ package daemon import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "errors" "fmt" "time" @@ -21,7 +24,12 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { OperationID: "closeIssue", Method: "POST", Path: "/api/v1/projects/{project_id}/issues/{ref}/actions/close", - }, func(ctx context.Context, in *api.ActionRequest) (*api.MutationResponse, error) { + }, func(ctx context.Context, in *api.CloseActionRequest) (*api.MutationResponse, error) { + if (in.IdempotencyKey != "" || in.IfMatch != "") && + in.Body.RetryProtocol != api.CloseRetryProtocol { + return nil, api.NewError(400, "retry_protocol_required", + "retry_protocol close-v1 is required with close retry headers", "", nil) + } actor, err := attributedActor(ctx, in.Body.Actor) if err != nil { return nil, err @@ -42,6 +50,10 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { if in.Body.Source == "tui" && in.Body.Reason == "" { in.Body.Reason = "done" } + ifMatchRev, err := parseOptionalIfMatchRevision(in.IfMatch) + if err != nil { + return nil, err + } // The TUI bypass is scoped to reason="done" — the only shape // the interactive "press x to close" path ever produces. A // caller sending source="tui" with reason="duplicate" or @@ -51,10 +63,40 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { // so duplicate/superseded closes still must carry their typed // targets and won't corrupt the audit trail. tuiBypass := tuiBypassAllowed(ctx, in.Body.Source, in.Body.Reason) + idempotencyFingerprint := "" + if in.IdempotencyKey != "" { + if _, err := activeProjectByID(ctx, cfg.DB, in.ProjectID); err != nil { + return nil, err + } + release, err := cfg.DB.AcquireIdempotencyLock(ctx, in.ProjectID, in.IdempotencyKey) + if err != nil { + return nil, internalAPIError(err) + } + defer func() { _ = release() }() + match, err := lookupCloseIdempotencyMatch(ctx, cfg, in.ProjectID, in.IdempotencyKey) + if err != nil { + return nil, err + } + if match != nil { + idempotencyFingerprint = closeIdempotencyFingerprint( + match.IssueUID, in.Ref, actor, in.Body.Reason, in.Body.Message, in.Body.Source, + in.Body.Evidence, in.Body.DryRun, ifMatchRev) + return closeIdempotencyResponse(ctx, cfg, match, idempotencyFingerprint) + } + } issue, err := activeIssueByRef(ctx, cfg.DB, in.ProjectID, in.Ref, db.IncludeDeletedNo) if err != nil { return nil, err } + if in.IdempotencyKey != "" { + idempotencyFingerprint = closeIdempotencyFingerprint( + issue.UID, in.Ref, actor, in.Body.Reason, in.Body.Message, in.Body.Source, + in.Body.Evidence, in.Body.DryRun, ifMatchRev) + } + if ifMatchRev != nil && issue.Revision != *ifMatchRev { + return nil, api.NewError(412, "revision_conflict", + fmt.Sprintf("issue revision is %d", issue.Revision), "", nil) + } // Already-closed short-circuit. CloseIssue itself returns // changed=false for this case; short-circuiting before the // guards (and substance / evidence validation) keeps idempotent @@ -63,6 +105,11 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { // has landed since the original close, or when the throttle // window is hot. Validation only gates real state transitions. if issue.Status == "closed" { + if in.IdempotencyKey != "" { + return nil, api.NewError(409, "issue_already_closed", + "issue was already closed by another request", + "omit the idempotency key to accept the current state", nil) + } out := &api.MutationResponse{} out.Body.Issue = issue return out, nil @@ -137,14 +184,42 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { var err error evt = nil events = nil - updated, events, changed, err = cfg.DB.CloseIssueWithEvents(ctx, issue.ID, - in.Body.Reason, actor, in.Body.Message, dbEvidence) + updated, events, changed, err = cfg.DB.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: issue.ID, ExpectedProjectID: in.ProjectID, + Reason: in.Body.Reason, Actor: actor, + Message: in.Body.Message, Evidence: dbEvidence, IfMatchRev: ifMatchRev, + IdempotencyKey: in.IdempotencyKey, IdempotencyFingerprint: idempotencyFingerprint, + }) if len(events) > 0 { evt = &events[0] } return err }) + if err == nil && changed { + cfg.Publish().Events(in.ProjectID, events) + } if err != nil { + // A connection can fail after the database commits. The keyed + // receipt proves whether this attempt landed. When it did, the + // batch this attempt returned is the committed batch, so publish + // it exactly once before replying with the receipt. + if in.IdempotencyKey != "" { + match, lookupErr := lookupCloseIdempotencyMatch( + ctx, cfg, in.ProjectID, in.IdempotencyKey) + if lookupErr != nil { + return nil, lookupErr + } + if match != nil { + if len(events) > 0 && events[0].UID == match.Event.UID { + cfg.Publish().Events(in.ProjectID, events) + } + return closeIdempotencyResponse(ctx, cfg, match, idempotencyFingerprint) + } + } + if revisionConflict, ok := errors.AsType[*db.RevisionConflictError](err); ok { + return nil, api.NewError(412, "revision_conflict", + fmt.Sprintf("issue revision is %d", revisionConflict.CurrentRevision), "", nil) + } // In-transaction guard re-fires when a concurrent link/create // added an open child between the read-side guard and the // close write. Map it to the same 409 code so clients see @@ -158,13 +233,31 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { } return nil, api.NewError(409, "parent_has_open_children", detail, "", nil) } + if errors.Is(err, db.ErrIssueProjectChanged) { + return nil, api.NewError(409, "issue_moved", + "issue moved to another project before the close committed", + "resolve the issue in its current project and retry with a fresh idempotency key", nil) + } if errors.Is(err, db.ErrFederatedReadOnly) { return nil, federationReadOnlyError(err) } return nil, internalAPIError(err) } - if changed { - cfg.Publish().Events(in.ProjectID, events) + // A database retry can observe that its first attempt committed even + // when the commit response was lost. Recover that attempt's receipt + // before returning a plain no-op response. + if !changed && in.IdempotencyKey != "" { + reuse, err := tryCloseIdempotencyMatch( + ctx, cfg, in.ProjectID, in.IdempotencyKey, idempotencyFingerprint) + if err != nil { + return nil, err + } + if reuse != nil { + return reuse, nil + } + return nil, api.NewError(409, "issue_already_closed", + "issue was closed by another request before this close committed", + "retry after reopening, or omit the idempotency key to accept the current state", nil) } out := &api.MutationResponse{} out.Body.Issue = updated @@ -219,6 +312,87 @@ func registerActionsHandlers(humaAPI huma.API, cfg ServerConfig) { }) } +func tryCloseIdempotencyMatch( + ctx context.Context, + cfg ServerConfig, + projectID int64, + key, fingerprint string, +) (*api.MutationResponse, error) { + match, err := lookupCloseIdempotencyMatch(ctx, cfg, projectID, key) + if err != nil { + return nil, err + } + if match == nil { + return nil, nil + } + return closeIdempotencyResponse(ctx, cfg, match, fingerprint) +} + +func lookupCloseIdempotencyMatch( + ctx context.Context, cfg ServerConfig, projectID int64, key string, +) (*db.IdempotencyMatch, error) { + match, err := cfg.DB.LookupIssueMutationIdempotency( + ctx, projectID, "issue.closed", key, time.Now().Add(-idempotencyWindow)) + if err != nil { + return nil, internalAPIError(err) + } + return match, nil +} + +func closeIdempotencyResponse( + ctx context.Context, cfg ServerConfig, match *db.IdempotencyMatch, fingerprint string, +) (*api.MutationResponse, error) { + if match.Fingerprint != fingerprint { + return nil, api.NewError(409, "idempotency_mismatch", + "idempotency key matched a prior close with a different fingerprint", + "use a fresh key or send the exact original close request", nil) + } + current, err := cfg.DB.IssueByID(ctx, match.IssueID) + if err != nil { + return nil, internalAPIError(err) + } + // The issue may have moved since the original close. Replaying the + // receipt exposes its current state, so the project it lives in now must + // be active and inside the caller's host scope. + if _, err := activeProjectByID(ctx, cfg.DB, current.ProjectID); err != nil { + return nil, err + } + if _, err := authorizeHostProjectScope(ctx, []int64{current.ProjectID}, nil, false); err != nil { + return nil, err + } + original := match.Event + out := &api.MutationResponse{} + out.Body.Issue = current + out.Body.OriginalEvent = &original + out.Body.Reused = true + return out, nil +} + +func closeIdempotencyFingerprint( + issueUID, requestRef, actor, reason, message, source string, + evidence []api.Evidence, + dryRun bool, + ifMatchRev *int64, +) string { + encoded, _ := json.Marshal(struct { + IssueUID string `json:"issue_uid"` + RequestRef string `json:"request_ref"` + Actor string `json:"actor"` + Reason string `json:"reason"` + Message string `json:"message"` + Source string `json:"source"` + Evidence []api.Evidence `json:"evidence"` + DryRun bool `json:"dry_run"` + IfMatchRev *int64 `json:"if_match_revision"` + }{ + IssueUID: issueUID, RequestRef: requestRef, + Actor: actor, Reason: reason, Message: message, + Source: source, Evidence: evidence, DryRun: dryRun, IfMatchRev: ifMatchRev, + }) + sum := sha256.Sum256(encoded) + return hex.EncodeToString(sum[:]) +} + // validateEvidenceTargets resolves duplicate-of and superseded-by issue // refs in the same project and rejects targets that are missing or that // point at the issue being closed. ValidateCloseInput already checks that diff --git a/internal/daemon/handlers_actions_retry_test.go b/internal/daemon/handlers_actions_retry_test.go new file mode 100644 index 00000000..9d2d9e41 --- /dev/null +++ b/internal/daemon/handlers_actions_retry_test.go @@ -0,0 +1,546 @@ +package daemon_test + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.kenn.io/kata/internal/api" + "go.kenn.io/kata/internal/daemon" + "go.kenn.io/kata/internal/db" +) + +type closeRaceStore struct { + db.Storage + raceNext bool +} + +type closeMoveRaceStore struct { + db.Storage + toProjectID int64 + moveNext bool +} + +func (s *closeMoveRaceStore) CloseIssueGuarded( + ctx context.Context, params db.CloseIssueParams, +) (db.Issue, []db.Event, bool, error) { + if s.moveNext { + s.moveNext = false + issue, err := s.IssueByID(ctx, params.IssueID) + if err != nil { + return db.Issue{}, nil, false, err + } + _, err = s.MoveIssueProject(ctx, db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: issue.ProjectID, ToProjectID: s.toProjectID, + IfMatchRev: issue.Revision, Actor: "moving-agent", + }) + if err != nil { + return db.Issue{}, nil, false, err + } + } + return s.Storage.CloseIssueGuarded(ctx, params) +} + +func (s *closeRaceStore) CloseIssueGuarded( + ctx context.Context, params db.CloseIssueParams, +) (db.Issue, []db.Event, bool, error) { + if s.raceNext { + s.raceNext = false + _, _, _, err := s.Storage.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: params.IssueID, Reason: "wontfix", Actor: "racing-agent", + }) + if err != nil { + return db.Issue{}, nil, false, err + } + } + return s.Storage.CloseIssueGuarded(ctx, params) +} + +type lostCloseResponseStore struct { + db.Storage + failNext bool + failIssueReadOnce bool + committed bool + committedEvents []db.Event +} + +func (s *lostCloseResponseStore) CloseIssueGuarded( + ctx context.Context, params db.CloseIssueParams, +) (db.Issue, []db.Event, bool, error) { + issue, events, changed, err := s.Storage.CloseIssueGuarded(ctx, params) + if err == nil && changed && s.failNext { + s.failNext = false + s.committed = true + s.committedEvents = append([]db.Event(nil), events...) + return issue, events, changed, errors.New("commit response lost") + } + return issue, events, changed, err +} + +func (s *lostCloseResponseStore) IssueByID(ctx context.Context, id int64) (db.Issue, error) { + if s.committed && s.failIssueReadOnce { + s.failIssueReadOnce = false + return db.Issue{}, errors.New("issue read unavailable") + } + return s.Storage.IssueByID(ctx, id) +} + +func TestClose_RetryControlsRequireProtocolMarker(t *testing.T) { + _, ts, projectID, issueID := bootstrapProjectWithIssue(t) + response := postWithHeader(t, ts, issueURL(projectID, issueID, "actions/close"), + map[string]string{"Idempotency-Key": "close-without-protocol"}, map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + }) + assertAPIError(t, response.status, response.body, + http.StatusBadRequest, "retry_protocol_required") +} + +func TestClose_RetryProtocolMarkerIsCloseOnly(t *testing.T) { + _, ts, projectID, issueID := bootstrapProjectWithIssue(t) + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + response := postWithHeader(t, ts, issueURL(projectID, issueID, "actions/close"), + map[string]string{"Idempotency-Key": "close-with-protocol"}, body) + requireOK(t, response) + + reopen := postWithHeader(t, ts, issueURL(projectID, issueID, "actions/reopen"), nil, + map[string]any{"actor": "agent-one", "retry_protocol": "close-v1"}) + assert.Equal(t, http.StatusBadRequest, reopen.status, string(reopen.body)) + assert.Contains(t, string(reopen.body), "retry_protocol") +} + +func TestClose_IdempotencyReplaysCommittedReceipt(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(context.Background(), issueID) + require.NoError(t, err) + path := issueURLRef(projectID, issue.ShortID, "actions/close") + headers := map[string]string{ + "Idempotency-Key": "close-request-1", + "If-Match": `"rev-1"`, + } + body := map[string]any{ + "actor": "agent-one", + "reason": "done", + "message": "Implemented the requested behavior and ran the focused tests.", + "retry_protocol": "close-v1", + "evidence": []map[string]any{{ + "type": "test", "command": "go test ./internal/daemon", + }}, + } + + first := postWithHeader(t, ts, path, headers, body) + requireOK(t, first) + var firstOut api.MutationResponse + require.NoError(t, json.Unmarshal(first.body, &firstOut.Body)) + require.True(t, firstOut.Body.Changed) + require.NotNil(t, firstOut.Body.Event) + assert.False(t, firstOut.Body.Reused) + + second := postWithHeader(t, ts, path, headers, body) + requireOK(t, second) + var secondOut api.MutationResponse + require.NoError(t, json.Unmarshal(second.body, &secondOut.Body)) + assert.False(t, secondOut.Body.Changed) + assert.True(t, secondOut.Body.Reused) + assert.Nil(t, secondOut.Body.Event) + require.NotNil(t, secondOut.Body.OriginalEvent) + assert.Equal(t, firstOut.Body.Event.UID, secondOut.Body.OriginalEvent.UID) + + events, err := h.DB().EventsAfter(context.Background(), db.EventsAfterParams{ + ProjectID: projectID, + Limit: 100, + }) + require.NoError(t, err) + closed := 0 + for _, event := range events { + if event.Type == "issue.closed" && event.IssueID != nil && *event.IssueID == issueID { + closed++ + } + } + assert.Equal(t, 1, closed) +} + +func TestClose_IdempotencyReplaysReceiptAfterSoftDelete(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(context.Background(), issueID) + require.NoError(t, err) + path := issueURLRef(projectID, issue.ShortID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-request-then-delete"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + + first := postWithHeader(t, ts, path, headers, body) + requireOK(t, first) + var firstOut api.MutationResponse + require.NoError(t, json.Unmarshal(first.body, &firstOut.Body)) + require.NotNil(t, firstOut.Body.Event) + + _, _, changed, err := h.DB().SoftDeleteIssue(context.Background(), issueID, "agent-one") + require.NoError(t, err) + require.True(t, changed) + + retry := postWithHeader(t, ts, path, headers, body) + requireOK(t, retry) + var retryOut api.MutationResponse + require.NoError(t, json.Unmarshal(retry.body, &retryOut.Body)) + assert.False(t, retryOut.Body.Changed) + assert.True(t, retryOut.Body.Reused) + require.NotNil(t, retryOut.Body.OriginalEvent) + assert.Equal(t, firstOut.Body.Event.UID, retryOut.Body.OriginalEvent.UID) + require.NotNil(t, retryOut.Body.Issue.DeletedAt) +} + +func TestClose_IdempotencyReplaysReceiptAfterMove(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + path := issueURLRef(projectID, issue.ShortID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-request-then-move"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + + first := postWithHeader(t, ts, path, headers, body) + requireOK(t, first) + var firstOut api.MutationResponse + require.NoError(t, json.Unmarshal(first.body, &firstOut.Body)) + require.NotNil(t, firstOut.Body.Event) + + target, err := h.DB().CreateProject(t.Context(), "target") + require.NoError(t, err) + moved, err := h.DB().MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issueID, FromProjectID: projectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + + retry := postWithHeader(t, ts, path, headers, body) + requireOK(t, retry) + var retryOut api.MutationResponse + require.NoError(t, json.Unmarshal(retry.body, &retryOut.Body)) + assert.False(t, retryOut.Body.Changed) + assert.True(t, retryOut.Body.Reused) + require.NotNil(t, retryOut.Body.OriginalEvent) + assert.Equal(t, firstOut.Body.Event.UID, retryOut.Body.OriginalEvent.UID) + assert.Equal(t, moved.Issue.ProjectID, retryOut.Body.Issue.ProjectID) +} + +func TestClose_RejectsKeyWhenAnotherCloseWinsBeforeWrite(t *testing.T) { + database := openTestDB(t) + project, err := database.db.CreateProject(t.Context(), "kata") + require.NoError(t, err) + issue, _, err := database.db.CreateIssue(t.Context(), db.CreateIssueParams{ + ProjectID: project.ID, Title: "race", Author: "agent-one", + }) + require.NoError(t, err) + store := &closeRaceStore{Storage: database.db, raceNext: true} + ts := startTestServer(t, daemon.ServerConfig{DB: store, StartedAt: database.now}) + path := issueURLRef(project.ID, issue.ShortID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-race-1"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + + first := postWithHeader(t, ts, path, headers, body) + assertAPIError(t, first.status, first.body, http.StatusConflict, "issue_already_closed") + + _, _, changed, err := database.db.ReopenIssue(t.Context(), issue.ID, "coordinator") + require.NoError(t, err) + require.True(t, changed) + + retry := postWithHeader(t, ts, path, headers, body) + requireOK(t, retry) + var retryOut api.MutationResponse + require.NoError(t, json.Unmarshal(retry.body, &retryOut.Body)) + assert.True(t, retryOut.Body.Changed) + assert.False(t, retryOut.Body.Reused) +} + +func TestClose_RejectsMoveThatWinsBeforeGuardedWrite(t *testing.T) { + database := openTestDB(t) + source, err := database.db.CreateProject(t.Context(), "source") + require.NoError(t, err) + target, err := database.db.CreateProject(t.Context(), "target") + require.NoError(t, err) + issue, _, err := database.db.CreateIssue(t.Context(), db.CreateIssueParams{ + ProjectID: source.ID, Title: "move race", Author: "agent-one", + }) + require.NoError(t, err) + store := &closeMoveRaceStore{Storage: database.db, toProjectID: target.ID, moveNext: true} + ts := startTestServer(t, daemon.ServerConfig{DB: store, StartedAt: database.now}) + response := postWithHeader(t, ts, + issueURLRef(source.ID, issue.ShortID, "actions/close"), + map[string]string{"Idempotency-Key": "close-move-race-1"}, map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + }) + assertAPIError(t, response.status, response.body, http.StatusConflict, "issue_moved") + + current, err := database.db.IssueByID(t.Context(), issue.ID) + require.NoError(t, err) + assert.Equal(t, target.ID, current.ProjectID) + assert.Equal(t, "open", current.Status) +} + +func TestClose_RecoversCommittedReceiptAndPublishesEventsOnce(t *testing.T) { + database := openTestDB(t) + project, issue := createClaimHubIssueInDB(t, database.db) + _, err := database.db.AcquireClaim(t.Context(), db.AcquireClaimParams{ + ProjectID: project.ID, IssueRef: issue.ShortID, + Principal: db.ClaimPrincipal{ + HolderInstanceUID: database.db.InstanceUID(), Holder: "agent-one", ClientKind: "cli", + }, + ClaimKind: "hard", Now: time.Now().UTC(), + }) + require.NoError(t, err) + store := &lostCloseResponseStore{Storage: database.db, failNext: true} + sink := &recordingSink{} + broadcaster := daemon.NewEventBroadcaster() + subscription := broadcaster.Subscribe(daemon.SubFilter{ProjectID: project.ID}) + defer subscription.Unsub() + ts := startTestServer(t, daemon.ServerConfig{ + DB: store, StartedAt: database.now, Hooks: sink, Broadcaster: broadcaster, + }) + path := issueURLRef(project.ID, issue.ShortID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-lost-response-1"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + + first := postWithHeader(t, ts, path, headers, body) + requireOK(t, first) + var firstOut api.MutationResponse + require.NoError(t, json.Unmarshal(first.body, &firstOut.Body)) + assert.False(t, firstOut.Body.Changed) + assert.True(t, firstOut.Body.Reused) + require.Len(t, store.committedEvents, 2) + assert.Equal(t, []string{"issue.closed", "claim.released"}, []string{ + store.committedEvents[0].Type, store.committedEvents[1].Type, + }) + assert.Equal(t, store.committedEvents, sink.snapshot()) + assert.Equal(t, []int64{store.committedEvents[0].ID, store.committedEvents[1].ID}, + drainBroadcastIDs(t, subscription.Ch, 50*time.Millisecond)) + + second := postWithHeader(t, ts, path, headers, body) + requireOK(t, second) + assert.Equal(t, store.committedEvents, sink.snapshot(), "an exact retry must not publish twice") + assert.Empty(t, drainBroadcastIDs(t, subscription.Ch, 50*time.Millisecond)) +} + +func TestClose_LostResponsePublishesOnceWhenReceiptReadFails(t *testing.T) { + database := openTestDB(t) + project, issue := createClaimHubIssueInDB(t, database.db) + _, err := database.db.AcquireClaim(t.Context(), db.AcquireClaimParams{ + ProjectID: project.ID, IssueRef: issue.ShortID, + Principal: db.ClaimPrincipal{ + HolderInstanceUID: database.db.InstanceUID(), Holder: "agent-one", ClientKind: "cli", + }, + ClaimKind: "hard", Now: time.Now().UTC(), + }) + require.NoError(t, err) + store := &lostCloseResponseStore{ + Storage: database.db, failNext: true, failIssueReadOnce: true, + } + sink := &recordingSink{} + broadcaster := daemon.NewEventBroadcaster() + subscription := broadcaster.Subscribe(daemon.SubFilter{ProjectID: project.ID}) + defer subscription.Unsub() + ts := startTestServer(t, daemon.ServerConfig{ + DB: store, StartedAt: database.now, Hooks: sink, Broadcaster: broadcaster, + }) + path := issueURLRef(project.ID, issue.ShortID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-delivery-retry-1"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + + first := postWithHeader(t, ts, path, headers, body) + assertAPIError(t, first.status, first.body, http.StatusInternalServerError, "internal") + require.Len(t, store.committedEvents, 2) + assert.Equal(t, store.committedEvents, sink.snapshot(), + "the committed batch is published even when the receipt response fails") + assert.Equal(t, []int64{store.committedEvents[0].ID, store.committedEvents[1].ID}, + drainBroadcastIDs(t, subscription.Ch, 50*time.Millisecond)) + + second := postWithHeader(t, ts, path, headers, body) + requireOK(t, second) + var secondOut api.MutationResponse + require.NoError(t, json.Unmarshal(second.body, &secondOut.Body)) + assert.True(t, secondOut.Body.Reused) + assert.Equal(t, store.committedEvents, sink.snapshot(), "a replay must not publish twice") + assert.Empty(t, drainBroadcastIDs(t, subscription.Ch, 50*time.Millisecond)) +} + +func TestClose_IdempotencyRejectsDifferentRequest(t *testing.T) { + _, ts, projectID, issueID := bootstrapProjectWithIssue(t) + path := issueURL(projectID, issueID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-request-1"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + requireOK(t, postWithHeader(t, ts, path, headers, body)) + + body["message"] = "A different explanation must not reuse the original close receipt." + retry := postWithHeader(t, ts, path, headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusConflict, "idempotency_mismatch") +} + +func TestClose_IdempotencyRejectsDifferentIssueWithSameBody(t *testing.T) { + h, ts, projectID, firstID := bootstrapProjectWithIssue(t) + first, err := h.DB().IssueByID(t.Context(), firstID) + require.NoError(t, err) + second, _, err := h.DB().CreateIssue(t.Context(), db.CreateIssueParams{ + ProjectID: projectID, Title: "second issue", Author: "agent-one", + }) + require.NoError(t, err) + headers := map[string]string{"Idempotency-Key": "close-request-1"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + requireOK(t, postWithHeader(t, ts, + issueURLRef(projectID, first.ShortID, "actions/close"), headers, body)) + + retry := postWithHeader(t, ts, + issueURLRef(projectID, second.ShortID, "actions/close"), headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusConflict, "idempotency_mismatch") + current, err := h.DB().IssueByID(t.Context(), second.ID) + require.NoError(t, err) + assert.Equal(t, "open", current.Status) +} + +func TestClose_IfMatchRejectsStaleRevision(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + _, err := h.DB().PatchIssueMetadata(context.Background(), db.PatchIssueMetadataIn{ + IssueID: issueID, + Actor: "coordinator", + Patch: map[string]json.RawMessage{ + "work.state": json.RawMessage(`"ready"`), + }, + }) + require.NoError(t, err) + + body := map[string]any{ + "actor": "agent-one", + "reason": "done", + "message": "Implemented the requested behavior and ran the focused tests.", + "retry_protocol": "close-v1", + "evidence": []map[string]any{{ + "type": "test", "command": "go test ./internal/daemon", + }}, + } + response := postWithHeader(t, ts, issueURL(projectID, issueID, "actions/close"), + map[string]string{"If-Match": `"rev-1"`}, body) + assertAPIError(t, response.status, response.body, + http.StatusPreconditionFailed, "revision_conflict") + + issue, err := h.DB().IssueByID(context.Background(), issueID) + require.NoError(t, err) + assert.Equal(t, "open", issue.Status) +} + +func TestClose_PresentEmptyIfMatchRejectsRequest(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + } + response := postWithHeader(t, ts, issueURL(projectID, issueID, "actions/close"), + map[string]string{"If-Match": ""}, body) + assertAPIError(t, response.status, response.body, + http.StatusBadRequest, "validation") + + issue, err := h.DB().IssueByID(context.Background(), issueID) + require.NoError(t, err) + assert.Equal(t, "open", issue.Status) +} + +func TestClose_IfMatchRejectsStaleRevisionAfterAnotherClose(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + _, err := h.DB().PatchIssueMetadata(context.Background(), db.PatchIssueMetadataIn{ + IssueID: issueID, + Actor: "coordinator", + Patch: map[string]json.RawMessage{ + "work.state": json.RawMessage(`"ready"`), + }, + }) + require.NoError(t, err) + path := issueURL(projectID, issueID, "actions/close") + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + } + requireOK(t, postWithHeader(t, ts, path, nil, body)) + + body["retry_protocol"] = "close-v1" + response := postWithHeader(t, ts, path, + map[string]string{"If-Match": `"rev-1"`}, body) + assertAPIError(t, response.status, response.body, + http.StatusPreconditionFailed, "revision_conflict") +} + +func TestClose_ReplayRejectsArchivedCurrentProject(t *testing.T) { + h, ts, projectID, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + path := issueURLRef(projectID, issue.UID, "actions/close") + headers := map[string]string{"Idempotency-Key": "close-then-archive"} + body := map[string]any{ + "actor": "agent-one", + "reason": "wontfix", + "message": "Reviewed the request and recorded why the work should stop here.", + "retry_protocol": "close-v1", + } + requireOK(t, postWithHeader(t, ts, path, headers, body)) + + target, err := h.DB().CreateProject(t.Context(), "archived-target") + require.NoError(t, err) + issue, err = h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + _, err = h.DB().MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: projectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + archiveProject(t, h, target.ID, true) + + retry := postWithHeader(t, ts, path, headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusNotFound, "project_not_found") +} diff --git a/internal/daemon/handlers_comments.go b/internal/daemon/handlers_comments.go index 96dc9b0a..8b724aa0 100644 --- a/internal/daemon/handlers_comments.go +++ b/internal/daemon/handlers_comments.go @@ -13,6 +13,8 @@ import ( "go.kenn.io/kata/internal/api" "go.kenn.io/kata/internal/db" + "go.kenn.io/kata/internal/shortid" + "go.kenn.io/kata/internal/uid" ) // registerCommentsHandlers installs POST /comments. CreateComment writes the @@ -28,42 +30,59 @@ func registerCommentsHandlers(humaAPI huma.API, cfg ServerConfig) { if err != nil { return nil, err } - issue, err := activeIssueByRef(ctx, cfg.DB, in.ProjectID, in.Ref, db.IncludeDeletedNo) - if err != nil { - return nil, err - } + var issue db.Issue + resolved := false fingerprint := "" if in.IdempotencyKey != "" { - release, err := cfg.DB.AcquireIdempotencyLock(ctx, in.ProjectID, in.IdempotencyKey) + routeProject, err := activeProjectByID(ctx, cfg.DB, in.ProjectID) + if err != nil { + return nil, err + } + // Comment keys are scoped to one issue UID. A full ULID ref needs + // no resolution, so its retry survives a project move; any other + // ref form resolves inside the route project first and falls back + // to the receipt this project already holds for the key. + issueUID := "" + if uid.Valid(in.Ref) { + issueUID = strings.ToUpper(in.Ref) + } else { + issue, err = activeIssueByRef(ctx, cfg.DB, in.ProjectID, in.Ref, db.IncludeDeletedNo) + if err == nil { + resolved = true + issueUID = issue.UID + } else if issueUID, err = receiptIssueUID( + ctx, cfg, routeProject, in.Ref, in.IdempotencyKey, err, + ); err != nil { + return nil, err + } + } + // Project IDs change when an issue moves. Zero is outside the + // persisted project ID range and gives every keyed comment one + // stable, backend-wide lock scope. + release, err := cfg.DB.AcquireIdempotencyLock(ctx, 0, issueUID+"\x00"+in.IdempotencyKey) if err != nil { return nil, internalAPIError(err) } defer func() { _ = release() }() - fingerprint = commentIdempotencyFingerprint(issue.UID, actor, in.Body.Body) match, err := cfg.DB.LookupCommentIdempotency( - ctx, in.ProjectID, in.IdempotencyKey, time.Now().Add(-idempotencyWindow)) + ctx, issueUID, in.IdempotencyKey, time.Now().Add(-idempotencyWindow)) if err != nil { return nil, internalAPIError(err) } if match != nil { - if match.Fingerprint != fingerprint { - return nil, api.NewError(409, "idempotency_mismatch", - "idempotency key matched a prior comment with a different fingerprint", - "use a fresh key or send the exact original comment", nil) - } - updated, err := cfg.DB.IssueByID(ctx, issue.ID) - if err != nil { - return nil, internalAPIError(err) - } - out := &api.CommentResponse{} - out.Body.Issue = updated - out.Body.Comment = match.Comment - out.Body.Event = nil - out.Body.Changed = false - return out, nil + return replayComment(ctx, cfg, in.ProjectID, match, actor, in.Body.Body) } } + if !resolved { + issue, err = activeIssueByRef(ctx, cfg.DB, in.ProjectID, in.Ref, db.IncludeDeletedNo) + if err != nil { + return nil, err + } + } + if in.IdempotencyKey != "" { + fingerprint = commentIdempotencyFingerprint(issue.UID, actor, in.Body.Body) + } c, evt, err := cfg.DB.CreateComment(ctx, db.CreateCommentParams{ IssueID: issue.ID, Author: actor, @@ -145,6 +164,76 @@ func registerCommentsHandlers(humaAPI huma.API, cfg ServerConfig) { }) } +// replayComment returns a committed comment receipt to an exact retry. The +// route must be the project the comment was written in or the project the +// issue lives in now, and that current project must still be active and +// inside the caller's host scope, because the reply exposes current state. +func replayComment( + ctx context.Context, + cfg ServerConfig, + routeProjectID int64, + match *db.CommentIdempotencyMatch, + actor, body string, +) (*api.CommentResponse, error) { + if match.Fingerprint != commentIdempotencyFingerprint(match.IssueUID, actor, body) { + return nil, api.NewError(409, "idempotency_mismatch", + "idempotency key matched a prior comment with a different fingerprint", + "use a fresh key or send the exact original comment", nil) + } + current, err := cfg.DB.IssueByID(ctx, match.Comment.IssueID) + if err != nil { + return nil, internalAPIError(err) + } + if current.DeletedAt != nil || + (routeProjectID != match.Event.ProjectID && routeProjectID != current.ProjectID) { + return nil, api.NewError(404, "issue_not_found", "issue not found", "", nil) + } + if _, err := activeProjectByID(ctx, cfg.DB, current.ProjectID); err != nil { + return nil, err + } + if _, err := authorizeHostProjectScope(ctx, []int64{current.ProjectID}, nil, false); err != nil { + return nil, err + } + out := &api.CommentResponse{} + out.Body.Issue = current + out.Body.Comment = match.Comment + out.Body.Event = nil + out.Body.Changed = false + return out, nil +} + +// receiptIssueUID recovers the issue a short-id retry addresses after that +// issue moved out of the route project. The receipt written in this project +// names the issue; the ref must still be a suffix of that issue's ULID, and a +// qualifier must name this project, so a key cannot steer a retry elsewhere. +// Any other outcome returns the original resolution error. +func receiptIssueUID( + ctx context.Context, + cfg ServerConfig, + routeProject db.Project, + ref, key string, + resolveErr error, +) (string, error) { + parsed, err := shortid.Parse(ref) + if err != nil || parsed.ShortID == "" || + (parsed.Project != "" && parsed.Project != routeProject.Name) { + return "", resolveErr + } + match, err := cfg.DB.LookupIssueMutationIdempotency( + ctx, routeProject.ID, "issue.commented", key, time.Now().Add(-idempotencyWindow)) + if err != nil { + return "", internalAPIError(err) + } + if match == nil { + return "", resolveErr + } + derived, err := shortid.Derive(match.IssueUID, len(parsed.ShortID)) + if err != nil || derived != parsed.ShortID { + return "", resolveErr + } + return match.IssueUID, nil +} + func commentIdempotencyFingerprint(issueUID, actor, body string) string { encoded, _ := json.Marshal(struct { IssueUID string `json:"issue_uid"` diff --git a/internal/daemon/handlers_comments_test.go b/internal/daemon/handlers_comments_test.go index 86828724..ac257304 100644 --- a/internal/daemon/handlers_comments_test.go +++ b/internal/daemon/handlers_comments_test.go @@ -3,14 +3,32 @@ package daemon_test import ( "context" "encoding/json" + "net/http" + "net/http/httptest" + "slices" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.kenn.io/kata/internal/daemon" "go.kenn.io/kata/internal/db" ) +type commentProjectHostAccess struct { + deniedProjectID int64 +} + +func (a commentProjectHostAccess) Authorize( + _ context.Context, + request daemon.HostAccessRequest, +) (daemon.HostAccessDecision, error) { + if slices.Contains(request.Operation.ProjectIDs, a.deniedProjectID) { + return daemon.HostAccessDecision{}, daemon.ErrHostAccessDenied + } + return daemon.HostAccessDecision{}, nil +} + func TestCommentEndpoint_AppendsAndEmitsEvent(t *testing.T) { _, ts, pid, num := bootstrapProjectWithIssue(t) @@ -56,6 +74,72 @@ func TestCommentEndpoint_IdempotencyReusesCommittedComment(t *testing.T) { require.Len(t, comments, 1) } +func TestCommentEndpoint_IdempotencyReplaysCommittedCommentAfterMove(t *testing.T) { + h, ts, sourceProjectID, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + body := map[string]any{"actor": "agent", "body": "first comment"} + headers := map[string]string{"Idempotency-Key": "comment-request-then-move"} + + first := postWithHeader(t, ts, issueURLRef(sourceProjectID, issue.UID, "comments"), headers, body) + requireOK(t, first) + target, err := h.DB().CreateProject(t.Context(), "target-project") + require.NoError(t, err) + issue, err = h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + _, err = h.DB().MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: sourceProjectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + + retry := postWithHeader(t, ts, issueURLRef(target.ID, issue.UID, "comments"), headers, body) + requireOK(t, retry) + var reused struct { + Changed bool `json:"changed"` + } + require.NoError(t, json.Unmarshal(retry.body, &reused)) + assert.False(t, reused.Changed) + comments, err := h.DB().CommentsByIssue(t.Context(), issueID) + require.NoError(t, err) + require.Len(t, comments, 1) +} + +func TestCommentEndpoint_IdempotencyReauthorizesMovedIssue(t *testing.T) { + dbh, initialServer, sourceProjectID, issueID := bootstrapProjectWithIssue(t) + issue, err := dbh.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + body := map[string]any{"actor": "agent", "body": "first comment"} + headers := map[string]string{"Idempotency-Key": "comment-request-before-move"} + path := issueURLRef(sourceProjectID, issue.UID, "comments") + requireOK(t, postWithHeader(t, initialServer, path, headers, body)) + + target, err := dbh.DB().CreateProject(t.Context(), "denied-target") + require.NoError(t, err) + issue, err = dbh.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + _, err = dbh.DB().MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: sourceProjectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + + server := daemon.NewServer(daemon.ServerConfig{ + DB: dbh.DB(), StartedAt: time.Now(), + HostAccess: commentProjectHostAccess{deniedProjectID: target.ID}, + }) + hostServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + ctx := daemon.WithPrincipal(request.Context(), daemon.Principal{ + Kind: daemon.PrincipalHost, Subject: "host-user", Actor: "agent", + }) + server.Handler().ServeHTTP(writer, request.WithContext(ctx)) + })) + t.Cleanup(hostServer.Close) + + retry := postWithHeader(t, hostServer, path, headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusNotFound, "not_found") +} + func TestCommentEndpoint_IdempotencyRejectsDifferentBody(t *testing.T) { _, ts, pid, issueID := bootstrapProjectWithIssue(t) path := issueURL(pid, issueID, "comments") @@ -186,3 +270,124 @@ func TestReopenIssue_BlankActorIs400(t *testing.T) { map[string]any{"actor": " "}) assertAPIError(t, resp.StatusCode, bs, 400, "validation") } + +func TestCommentEndpoint_IdempotencyKeyIsScopedToIssue(t *testing.T) { + h, ts, pid, firstIssueID := bootstrapProjectWithIssue(t) + second, _, err := h.DB().CreateIssue(t.Context(), db.CreateIssueParams{ + ProjectID: pid, Title: "second issue", Author: "agent", + }) + require.NoError(t, err) + body := map[string]any{"actor": "agent", "body": "same body"} + headers := map[string]string{"Idempotency-Key": "shared-key"} + + first := postWithHeader(t, ts, issueURL(pid, firstIssueID, "comments"), headers, body) + requireOK(t, first) + var firstOut struct { + Comment struct { + UID string `json:"uid"` + } `json:"comment"` + } + require.NoError(t, json.Unmarshal(first.body, &firstOut)) + + other := postWithHeader(t, ts, issueURLRef(pid, second.ShortID, "comments"), headers, body) + requireOK(t, other) + var otherOut struct { + Comment struct { + UID string `json:"uid"` + } `json:"comment"` + Changed bool `json:"changed"` + } + require.NoError(t, json.Unmarshal(other.body, &otherOut)) + assert.True(t, otherOut.Changed, "a key used on another issue must not replay the first issue's comment") + assert.NotEqual(t, firstOut.Comment.UID, otherOut.Comment.UID) + comments, err := h.DB().CommentsByIssue(t.Context(), second.ID) + require.NoError(t, err) + require.Len(t, comments, 1) +} + +func TestCommentEndpoint_IdempotencyReplayRequiresRelatedProject(t *testing.T) { + h, ts, pid, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + body := map[string]any{"actor": "agent", "body": "first comment"} + headers := map[string]string{"Idempotency-Key": "comment-request-unrelated-route"} + requireOK(t, postWithHeader(t, ts, issueURLRef(pid, issue.UID, "comments"), headers, body)) + + unrelated, err := h.DB().CreateProject(t.Context(), "unrelated-project") + require.NoError(t, err) + retry := postWithHeader(t, ts, issueURLRef(unrelated.ID, issue.UID, "comments"), headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusNotFound, "issue_not_found") +} + +func TestCommentEndpoint_IdempotencyReplayRejectsArchivedCurrentProject(t *testing.T) { + h, ts, sourceProjectID, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + body := map[string]any{"actor": "agent", "body": "first comment"} + headers := map[string]string{"Idempotency-Key": "comment-request-then-archive"} + path := issueURLRef(sourceProjectID, issue.UID, "comments") + requireOK(t, postWithHeader(t, ts, path, headers, body)) + + target, err := h.DB().CreateProject(t.Context(), "archived-target") + require.NoError(t, err) + issue, err = h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + _, err = h.DB().MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: sourceProjectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + archiveProject(t, h, target.ID, true) + + retry := postWithHeader(t, ts, path, headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusNotFound, "project_not_found") +} + +func TestCommentEndpoint_IdempotencyReplaysShortIDRetryAfterMove(t *testing.T) { + h, ts, sourceProjectID, issueID := bootstrapProjectWithIssue(t) + body := map[string]any{"actor": "agent", "body": "first comment"} + headers := map[string]string{"Idempotency-Key": "short-id-comment-then-move"} + path := issueURL(sourceProjectID, issueID, "comments") + requireOK(t, postWithHeader(t, ts, path, headers, body)) + + target, err := h.DB().CreateProject(t.Context(), "target-project") + require.NoError(t, err) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + _, err = h.DB().MoveIssueProject(t.Context(), db.MoveIssueProjectIn{ + IssueID: issue.ID, FromProjectID: sourceProjectID, ToProjectID: target.ID, + IfMatchRev: issue.Revision, Actor: "coordinator", + }) + require.NoError(t, err) + + retry := postWithHeader(t, ts, path, headers, body) + requireOK(t, retry) + var reused struct { + Changed bool `json:"changed"` + Issue struct { + ProjectID int64 `json:"project_id"` + } `json:"issue"` + } + require.NoError(t, json.Unmarshal(retry.body, &reused)) + assert.False(t, reused.Changed) + assert.Equal(t, target.ID, reused.Issue.ProjectID) + comments, err := h.DB().CommentsByIssue(t.Context(), issueID) + require.NoError(t, err) + require.Len(t, comments, 1) +} + +func TestCommentEndpoint_IdempotencyReplayRejectsSoftDeletedIssue(t *testing.T) { + h, ts, pid, issueID := bootstrapProjectWithIssue(t) + issue, err := h.DB().IssueByID(t.Context(), issueID) + require.NoError(t, err) + body := map[string]any{"actor": "agent", "body": "first comment"} + headers := map[string]string{"Idempotency-Key": "comment-then-delete"} + path := issueURLRef(pid, issue.UID, "comments") + requireOK(t, postWithHeader(t, ts, path, headers, body)) + + _, _, _, err = h.DB().SoftDeleteIssue(t.Context(), issueID, "agent") + require.NoError(t, err) + + retry := postWithHeader(t, ts, path, headers, body) + assertAPIError(t, retry.status, retry.body, http.StatusNotFound, "issue_not_found") +} diff --git a/internal/daemon/openapi.go b/internal/daemon/openapi.go index ebc26e52..41cf3d25 100644 --- a/internal/daemon/openapi.go +++ b/internal/daemon/openapi.go @@ -16,7 +16,7 @@ import ( // (info.version). It tracks the HTTP API contract, not the build version, so // the committed schema artifact stays stable across builds and is bumped // deliberately when the wire contract changes. -const APISchemaVersion = "0.14.0" +const APISchemaVersion = "0.15.0" // OpenAPIDocument builds the daemon's complete OpenAPI model by wiring every // route through NewServer with a zero ServerConfig. It binds no listener and diff --git a/internal/daemon/openapi_test.go b/internal/daemon/openapi_test.go index 5d66308f..f95a8d8d 100644 --- a/internal/daemon/openapi_test.go +++ b/internal/daemon/openapi_test.go @@ -287,9 +287,9 @@ func TestOpenAPIDocumentIncludesUIReadContract(t *testing.T) { } } -func TestOpenAPISchemaVersionReflectsExternalEvidence(t *testing.T) { - if APISchemaVersion != "0.14.0" { - t.Fatalf("APISchemaVersion = %q, want 0.14.0 for external evidence", APISchemaVersion) +func TestOpenAPISchemaVersionReflectsCloseRetrySafety(t *testing.T) { + if APISchemaVersion != "0.15.0" { + t.Fatalf("APISchemaVersion = %q, want 0.15.0 for close retry safety", APISchemaVersion) } } diff --git a/internal/db/dbtest/conformance.go b/internal/db/dbtest/conformance.go index 5a4b4225..615077df 100644 --- a/internal/db/dbtest/conformance.go +++ b/internal/db/dbtest/conformance.go @@ -71,7 +71,7 @@ var storageScenarios = []scenario{ }, { name: "idempotency", - methods: []string{"AcquireIdempotencyLock", "CreateComment", "CreateIssue", "CreateProject", "LookupCommentIdempotency", "LookupIdempotency"}, + methods: []string{"AcquireIdempotencyLock", "CreateComment", "CreateIssue", "CreateProject", "LookupCommentIdempotency", "LookupIdempotency", "LookupIssueMutationIdempotency"}, run: checkIdempotency, }, { @@ -99,7 +99,7 @@ var storageScenarios = []scenario{ { name: "issue lifecycle", methods: []string{ - "BatchProjectStats", "ClaimOwner", "CloseIssue", "CloseIssueWithEvents", "CreateIssue", "CreateProject", "EditIssue", + "BatchProjectStats", "ClaimOwner", "CloseIssue", "CloseIssueGuarded", "CloseIssueWithEvents", "CreateIssue", "CreateProject", "EditIssue", "IssueByShortID", "IssueUIDPrefixMatch", "ListAllIssues", "ReopenIssue", "RestoreIssue", "SoftDeleteIssue", "UpdateOwner", "UpdatePriority", }, diff --git a/internal/db/dbtest/conformance_core.go b/internal/db/dbtest/conformance_core.go index 197e8485..0140ff5f 100644 --- a/internal/db/dbtest/conformance_core.go +++ b/internal/db/dbtest/conformance_core.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "sync" "testing" "time" @@ -328,20 +329,107 @@ func checkIdempotency(t *testing.T, store db.Storage) error { if err != nil { return fmt.Errorf("create comment: %w", err) } - commentMatch, err := store.LookupCommentIdempotency(ctx, project.ID, "comment-request-1", since) + commentMatch, err := store.LookupCommentIdempotency(ctx, issue.UID, "comment-request-1", since) if err != nil { return fmt.Errorf("lookup comment idempotency: %w", err) } require.NotNil(t, commentMatch) assert.Equal(t, comment.UID, commentMatch.Comment.UID) + assert.Equal(t, issue.UID, commentMatch.IssueUID) assert.Equal(t, commentEvent.UID, commentMatch.Event.UID) assert.Equal(t, "comment-fingerprint-1", commentMatch.Fingerprint) - missingComment, err := store.LookupCommentIdempotency(ctx, project.ID, "comment-request-2", since) + missingComment, err := store.LookupCommentIdempotency(ctx, issue.UID, "comment-request-2", since) if err != nil { return fmt.Errorf("lookup missing comment idempotency key: %w", err) } assert.Nil(t, missingComment) + closeFingerprint := strings.Repeat("a", 64) + closed, closeEvents, changed, err := store.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: issue.ID, Reason: "wontfix", Actor: "conformance-agent", + Message: "Recorded the reason for stopping this conformance task.", + IdempotencyKey: "close-request-1", IdempotencyFingerprint: closeFingerprint, + IfMatchRev: new(issue.Revision), + }) + if err != nil { + return fmt.Errorf("guarded close: %w", err) + } + assert.True(t, changed) + assert.Equal(t, "closed", closed.Status) + require.NotEmpty(t, closeEvents) + closeMatch, err := store.LookupIssueMutationIdempotency( + ctx, project.ID, "issue.closed", "close-request-1", since) + if err != nil { + return fmt.Errorf("lookup close idempotency: %w", err) + } + require.NotNil(t, closeMatch) + assert.Equal(t, issue.ID, closeMatch.IssueID) + assert.Equal(t, closeEvents[0].UID, closeMatch.Event.UID) + assert.Equal(t, closeFingerprint, closeMatch.Fingerprint) + + projectGuardIssue, _, err := store.CreateIssue(ctx, db.CreateIssueParams{ + ProjectID: project.ID, Title: "project-pinned close", Author: "conformance-agent", + }) + if err != nil { + return fmt.Errorf("create project-pinned close issue: %w", err) + } + _, _, _, err = store.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: projectGuardIssue.ID, ExpectedProjectID: project.ID + 9999, + Reason: "wontfix", Actor: "conformance-agent", + }) + if !errors.Is(err, db.ErrIssueProjectChanged) { + return fmt.Errorf("project-pinned close returned %v, want issue project changed", err) + } + projectGuardIssue, err = store.IssueByID(ctx, projectGuardIssue.ID) + if err != nil { + return fmt.Errorf("read project-pinned close issue: %w", err) + } + assert.Equal(t, "open", projectGuardIssue.Status) + + staleIssue, _, err := store.CreateIssue(ctx, db.CreateIssueParams{ + ProjectID: project.ID, Title: "stale close guard", Author: "conformance-agent", + }) + if err != nil { + return fmt.Errorf("create stale close issue: %w", err) + } + if _, err := store.PatchIssueMetadata(ctx, db.PatchIssueMetadataIn{ + IssueID: staleIssue.ID, Actor: "conformance-agent", + Patch: map[string]json.RawMessage{"work.state": json.RawMessage(`"ready"`)}, + }); err != nil { + return fmt.Errorf("advance close revision: %w", err) + } + _, _, _, err = store.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: staleIssue.ID, Reason: "wontfix", Actor: "conformance-agent", + IfMatchRev: new(staleIssue.Revision), + }) + conflict, ok := errors.AsType[*db.RevisionConflictError](err) + if !ok || conflict == nil { + return fmt.Errorf("guarded close returned %v, want revision conflict", err) + } + assert.Equal(t, staleIssue.Revision+1, conflict.CurrentRevision) + if _, _, _, err := store.CloseIssueWithEvents( + ctx, staleIssue.ID, "wontfix", "conformance-agent", "stopped", nil, + ); err != nil { + return fmt.Errorf("close after revision conflict: %w", err) + } + _, _, _, err = store.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: staleIssue.ID, Reason: "wontfix", Actor: "conformance-agent", + IfMatchRev: new(staleIssue.Revision), + }) + conflict, ok = errors.AsType[*db.RevisionConflictError](err) + if !ok || conflict == nil { + return fmt.Errorf("guarded closed-issue retry returned %v, want revision conflict", err) + } + assert.Equal(t, staleIssue.Revision+1, conflict.CurrentRevision) + + nulRelease, err := store.AcquireIdempotencyLock(ctx, 0, "issue-uid\x00comment-request") + if err != nil { + return fmt.Errorf("acquire idempotency lock containing NUL: %w", err) + } + if err := nulRelease(); err != nil { + return fmt.Errorf("release idempotency lock containing NUL: %w", err) + } + releaseFirst, err := store.AcquireIdempotencyLock(ctx, project.ID, "serialized-request") if err != nil { return fmt.Errorf("acquire first idempotency lock: %w", err) diff --git a/internal/db/errors.go b/internal/db/errors.go index 0b5cc98a..658f5c20 100644 --- a/internal/db/errors.go +++ b/internal/db/errors.go @@ -19,6 +19,10 @@ var ( // has open child issues. ErrOpenChildren = errors.New("issue has open children") + // ErrIssueProjectChanged is returned when a guarded mutation finds that + // its issue moved after the caller resolved the project-scoped route. + ErrIssueProjectChanged = errors.New("issue project changed") + // ErrNoFields is returned by EditIssue when no field changes are // requested. ErrNoFields = errors.New("no fields to update") @@ -346,8 +350,8 @@ func (e *RecurrencePinnedError) Error() string { return "cannot move: issue is part of a recurrence series" } -// RevisionConflictError is returned by MoveIssueProject when the caller's -// IfMatchRev does not match the issue's current revision. +// RevisionConflictError reports that a caller's IfMatchRev does not match the +// current revision of the guarded record. type RevisionConflictError struct { CurrentRevision int64 } diff --git a/internal/db/params.go b/internal/db/params.go index 4ad1b273..04f2ea63 100644 --- a/internal/db/params.go +++ b/internal/db/params.go @@ -82,6 +82,22 @@ type CreateIssueParams struct { Metadata map[string]json.RawMessage } +// CloseIssueParams carries optional retry and concurrency guards for one close. +// A nil IfMatchRev preserves the unconditional close behavior. Idempotency +// fields are written to the close event so a caller can recover its original +// receipt after losing the response. +type CloseIssueParams struct { + IssueID int64 + ExpectedProjectID int64 + Reason string + Actor string + Message string + Evidence []Evidence + IfMatchRev *int64 + IdempotencyKey string + IdempotencyFingerprint string +} + // ListIssuesParams filters single-project list output. type ListIssuesParams struct { ProjectID int64 diff --git a/internal/db/pgstore/idempotency.go b/internal/db/pgstore/idempotency.go index a5727aaf..1ac7d763 100644 --- a/internal/db/pgstore/idempotency.go +++ b/internal/db/pgstore/idempotency.go @@ -18,12 +18,24 @@ func (s *Store) LookupIdempotency( key string, since time.Time, ) (*db.IdempotencyMatch, error) { - query := eventSelect + ` WHERE e.type = 'issue.created' - AND e.project_id = $1 - AND e.payload::jsonb ->> 'idempotency_key' = $2 - AND e.created_at >= $3 - ORDER BY e.id DESC LIMIT 1` - event, err := scanEvent(s.QueryRowContext(ctx, query, projectID, key, formatStoredTime(since))) + return s.LookupIssueMutationIdempotency(ctx, projectID, "issue.created", key, since) +} + +// LookupIssueMutationIdempotency finds the newest recent issue event of the +// requested type carrying key. +func (s *Store) LookupIssueMutationIdempotency( + ctx context.Context, + projectID int64, + eventType string, + key string, + since time.Time, +) (*db.IdempotencyMatch, error) { + query := eventSelect + ` WHERE e.type = $1 + AND e.project_id = $2 + AND e.payload::jsonb ->> 'idempotency_key' = $3 + AND e.created_at >= $4 + ORDER BY e.id DESC LIMIT 1` + event, err := scanEvent(s.QueryRowContext(ctx, query, eventType, projectID, key, formatStoredTime(since))) if errors.Is(err, db.ErrNotFound) { return nil, nil } @@ -44,33 +56,33 @@ func (s *Store) LookupIdempotency( return nil, mapSQLError(err, nil) } return &db.IdempotencyMatch{ - IssueID: issue.ID, IssueShortID: issue.ShortID, + IssueID: issue.ID, IssueUID: issue.UID, IssueShortID: issue.ShortID, Fingerprint: fingerprint.String, Event: event, }, nil } -// LookupCommentIdempotency finds the newest recent issue.commented event -// carrying key and returns the comment that event created. +// LookupCommentIdempotency finds the newest recent issue.commented event on +// issueUID carrying key and returns the comment that event created. func (s *Store) LookupCommentIdempotency( ctx context.Context, - projectID int64, + issueUID string, key string, since time.Time, ) (*db.CommentIdempotencyMatch, error) { query := eventSelect + ` WHERE e.type = 'issue.commented' - AND e.project_id = $1 + AND e.issue_uid = $1 AND e.payload::jsonb ->> 'idempotency_key' = $2 AND e.created_at >= $3 ORDER BY e.id DESC LIMIT 1` - event, err := scanEvent(s.QueryRowContext(ctx, query, projectID, key, formatStoredTime(since))) + event, err := scanEvent(s.QueryRowContext(ctx, query, issueUID, key, formatStoredTime(since))) if errors.Is(err, db.ErrNotFound) { return nil, nil } if err != nil { return nil, err } - if event.IssueID == nil { - return nil, fmt.Errorf("comment idempotency match has no issue_id") + if event.IssueID == nil || event.IssueUID == nil { + return nil, fmt.Errorf("comment idempotency match has no issue identity") } var payload struct { CommentUID string `json:"comment_uid"` @@ -85,6 +97,7 @@ func (s *Store) LookupCommentIdempotency( return nil, fmt.Errorf("comment idempotency match comment: %w", err) } return &db.CommentIdempotencyMatch{ - Comment: comment, Fingerprint: payload.Fingerprint, Event: event, + Comment: comment, IssueUID: *event.IssueUID, + Fingerprint: payload.Fingerprint, Event: event, }, nil } diff --git a/internal/db/pgstore/idempotency_lock.go b/internal/db/pgstore/idempotency_lock.go index 92a17ece..9f6b14cb 100644 --- a/internal/db/pgstore/idempotency_lock.go +++ b/internal/db/pgstore/idempotency_lock.go @@ -2,6 +2,7 @@ package pgstore import ( "context" + "crypto/sha256" "errors" "fmt" "time" @@ -23,7 +24,8 @@ func (s *Store) AcquireIdempotencyLock( if s.idempotencyDB == nil { return nil, errors.New("postgres idempotency coordinator is unavailable") } - lockIdentity := fmt.Sprintf("kata:pgstore:idempotency:%s:%d:%s", s.schema, projectID, key) + rawLockIdentity := fmt.Sprintf("kata:pgstore:idempotency:%s:%d:%s", s.schema, projectID, key) + lockIdentity := fmt.Sprintf("%x", sha256.Sum256([]byte(rawLockIdentity))) for { conn, err := s.idempotencyDB.Conn(ctx) if err != nil { diff --git a/internal/db/pgstore/issue_lifecycle.go b/internal/db/pgstore/issue_lifecycle.go index f2707bfc..8f061a58 100644 --- a/internal/db/pgstore/issue_lifecycle.go +++ b/internal/db/pgstore/issue_lifecycle.go @@ -273,22 +273,28 @@ func (s *Store) CloseIssueWithEvents( actor string, message string, evidence []db.Evidence, +) (db.Issue, []db.Event, bool, error) { + return s.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: issueID, Reason: reason, Actor: actor, Message: message, Evidence: evidence, + }) +} + +// CloseIssueGuarded applies the close and its optional revision precondition in +// the same transaction, then returns every event committed by that mutation. +func (s *Store) CloseIssueGuarded( + ctx context.Context, p db.CloseIssueParams, ) (db.Issue, []db.Event, bool, error) { return s.closeIssueWithEvents( - ctx, issueID, reason, actor, message, evidence, s.withSerializableTx, + ctx, p, s.withSerializableTx, ) } func (s *Store) closeIssueWithEvents( ctx context.Context, - issueID int64, - reason string, - actor string, - message string, - evidence []db.Evidence, + p db.CloseIssueParams, runTx func(context.Context, transactionFunc) error, ) (db.Issue, []db.Event, bool, error) { - if reason == "" { + if p.Reason == "" { return db.Issue{}, nil, false, fmt.Errorf("close reason is required") } var issue db.Issue @@ -296,10 +302,16 @@ func (s *Store) closeIssueWithEvents( var changed bool err := runTx(ctx, func(tx *sql.Tx) error { issue, events, changed = db.Issue{}, nil, false - current, project, err := lockedIssueTx(ctx, tx, issueID, false) + current, project, err := lockedIssueTx(ctx, tx, p.IssueID, false) if err != nil { return err } + if p.ExpectedProjectID != 0 && current.ProjectID != p.ExpectedProjectID { + return db.ErrIssueProjectChanged + } + if p.IfMatchRev != nil && current.Revision != *p.IfMatchRev { + return &db.RevisionConflictError{CurrentRevision: current.Revision} + } if current.Status == "closed" { issue = current return nil @@ -317,7 +329,7 @@ func (s *Store) closeIssueWithEvents( } closedAt := mutationTimestamp() if _, err := tx.ExecContext(ctx, `UPDATE issues SET status = 'closed', closed_reason = $1, - closed_at = $2, updated_at = $2 WHERE id = $3`, reason, closedAt, current.ID); err != nil { + closed_at = $2, updated_at = $2 WHERE id = $3`, p.Reason, closedAt, current.ID); err != nil { return mapSQLError(err, nil) } parentUID, parentShortID := new(string), new(string) @@ -332,27 +344,30 @@ func (s *Store) closeIssueWithEvents( return mapSQLError(parentErr, nil) } body, err := json.Marshal(struct { - Reason string `json:"reason"` - ClosedAt string `json:"closed_at"` - Message string `json:"message,omitempty"` - Evidence []db.Evidence `json:"evidence,omitempty"` - ParentUID *string `json:"parent_uid,omitempty"` - ParentShortID *string `json:"parent_short_id,omitempty"` + Reason string `json:"reason"` + ClosedAt string `json:"closed_at"` + Message string `json:"message,omitempty"` + Evidence []db.Evidence `json:"evidence,omitempty"` + ParentUID *string `json:"parent_uid,omitempty"` + ParentShortID *string `json:"parent_short_id,omitempty"` + IdempotencyKey string `json:"idempotency_key,omitempty"` + IdempotencyFingerprint string `json:"idempotency_fingerprint,omitempty"` }{ - Reason: reason, ClosedAt: closedAt, Message: message, Evidence: evidence, + Reason: p.Reason, ClosedAt: closedAt, Message: p.Message, Evidence: p.Evidence, ParentUID: parentUID, ParentShortID: parentShortID, + IdempotencyKey: p.IdempotencyKey, IdempotencyFingerprint: p.IdempotencyFingerprint, }) if err != nil { return err } created, err := s.insertEventTx(ctx, tx, - issueEventInput(current, project, "issue.closed", actor, string(body))) + issueEventInput(current, project, "issue.closed", p.Actor, string(body))) if err != nil { return err } events, changed = []db.Event{created}, true auditEvents, err := s.annotateClaimWorkMutationTx(ctx, tx, claimWorkMutationInput{ - Project: project, Issue: current, EventType: "issue.closed", Actor: actor, + Project: project, Issue: current, EventType: "issue.closed", Actor: p.Actor, HolderInstanceUID: s.InstanceUID(), OffendingEventUID: created.UID, }) if err != nil { @@ -363,9 +378,9 @@ func (s *Store) closeIssueWithEvents( if len(auditEvents) > 0 { lastEventID = auditEvents[len(auditEvents)-1].ID } - if reason == "done" && current.RecurrenceID != nil && current.OccurrenceKey != nil { + if p.Reason == "done" && current.RecurrenceID != nil && current.OccurrenceKey != nil { if _, err := s.materializeNextTx( - ctx, tx, *current.RecurrenceID, *current.OccurrenceKey, actor, + ctx, tx, *current.RecurrenceID, *current.OccurrenceKey, p.Actor, ); err != nil { return fmt.Errorf("materialize next recurrence: %w", err) } diff --git a/internal/db/pgstore/issue_lifecycle_retry_test.go b/internal/db/pgstore/issue_lifecycle_retry_test.go index ee86bde3..b857c990 100644 --- a/internal/db/pgstore/issue_lifecycle_retry_test.go +++ b/internal/db/pgstore/issue_lifecycle_retry_test.go @@ -31,7 +31,7 @@ func TestCloseIssueRetryClearsRolledBackAttemptOutput(t *testing.T) { closedAt := mutationTimestamp() closed, events, changed, err := store.closeIssueWithEvents( - ctx, issue.ID, "done", "tester", "", nil, + ctx, db.CloseIssueParams{IssueID: issue.ID, Reason: "done", Actor: "tester"}, rollbackThenRetry(t, store, func() { _, updateErr := store.ExecContext(ctx, `UPDATE issues SET status='closed', closed_reason='done', closed_at=$1, updated_at=$1 WHERE id=$2`, diff --git a/internal/db/pgstore/stubgen/main.go b/internal/db/pgstore/stubgen/main.go index 18bb6d54..6d9999db 100644 --- a/internal/db/pgstore/stubgen/main.go +++ b/internal/db/pgstore/stubgen/main.go @@ -106,6 +106,7 @@ var alreadyImplemented = map[string]bool{ "ClearPendingExternalComment": true, // external_roots.go "Close": true, // inherited from embedded *sql.DB "CloseIssue": true, // issue_lifecycle.go + "CloseIssueGuarded": true, // issue_lifecycle.go "CloseIssueWithEvents": true, // issue_lifecycle.go "CommentBodyByID": true, // comments.go "CommentsByIssue": true, // comments.go @@ -208,6 +209,7 @@ var alreadyImplemented = map[string]bool{ "LinksByIssue": true, // links.go "Path": true, // store.go "LookupIdempotency": true, // idempotency.go + "LookupIssueMutationIdempotency": true, // idempotency.go "LookupCommentIdempotency": true, // idempotency.go "MaxEventID": true, // events.go "MaxFederationBaselineEventID": true, // events.go diff --git a/internal/db/pgstore/stubgen/main_test.go b/internal/db/pgstore/stubgen/main_test.go index a5f85721..5aa3c4ed 100644 --- a/internal/db/pgstore/stubgen/main_test.go +++ b/internal/db/pgstore/stubgen/main_test.go @@ -38,7 +38,7 @@ type Storage interface { Only(context.Context) error } func TestStorageMethodInventoryClassifiesEveryMethod(t *testing.T) { methods, err := CollectStorageMethodInventory("../../storage.go") require.NoError(t, err) - require.Len(t, methods, 249) + require.Len(t, methods, 251) var implemented []string var stubbed []string @@ -88,6 +88,7 @@ func TestStorageMethodInventoryClassifiesEveryMethod(t *testing.T) { "ClearPendingExternalComment", "Close", "CloseIssue", + "CloseIssueGuarded", "CloseIssueWithEvents", "CommentBodyByID", "CommentsByIssue", @@ -208,6 +209,7 @@ func TestStorageMethodInventoryClassifiesEveryMethod(t *testing.T) { "ListRecurrencesByProject", "LookupCommentIdempotency", "LookupIdempotency", + "LookupIssueMutationIdempotency", "MarkClaimStatusRefreshError", "MarkPendingClaimAttempt", "MaterializeFederatedProject", diff --git a/internal/db/sqlitestore/idempotency_lock.go b/internal/db/sqlitestore/idempotency_lock.go index db2d6a21..e11edcd6 100644 --- a/internal/db/sqlitestore/idempotency_lock.go +++ b/internal/db/sqlitestore/idempotency_lock.go @@ -22,7 +22,7 @@ func newIdempotencyLockSet() *idempotencyLockSet { return set } -// AcquireIdempotencyLock serializes concurrent create retries within the one +// AcquireIdempotencyLock serializes concurrent mutation retries within the one // SQLite daemon that owns this database. SQLite daemon discovery prevents a // second process from serving the same file; fixed stripes keep memory bounded. func (d *Store) AcquireIdempotencyLock( diff --git a/internal/db/sqlitestore/queries.go b/internal/db/sqlitestore/queries.go index cb69cbe3..d726e070 100644 --- a/internal/db/sqlitestore/queries.go +++ b/internal/db/sqlitestore/queries.go @@ -1452,19 +1452,26 @@ func (d *Store) CloseIssueWithEvents( issueID int64, reason, actor, message string, evidence []db.Evidence, +) (db.Issue, []db.Event, bool, error) { + return d.CloseIssueGuarded(ctx, db.CloseIssueParams{ + IssueID: issueID, Reason: reason, Actor: actor, Message: message, Evidence: evidence, + }) +} + +// CloseIssueGuarded applies the close and its optional revision precondition in +// the same transaction, then returns every event committed by that mutation. +func (d *Store) CloseIssueGuarded( + ctx context.Context, p db.CloseIssueParams, ) (db.Issue, []db.Event, bool, error) { return retryWrite3(ctx, d, func() (db.Issue, []db.Event, bool, error) { - return d.closeIssueWithEvents(ctx, issueID, reason, actor, message, evidence) + return d.closeIssueGuarded(ctx, p) }) } -func (d *Store) closeIssueWithEvents( - ctx context.Context, - issueID int64, - reason, actor, message string, - evidence []db.Evidence, +func (d *Store) closeIssueGuarded( + ctx context.Context, p db.CloseIssueParams, ) (db.Issue, []db.Event, bool, error) { - if reason == "" { + if p.Reason == "" { return db.Issue{}, nil, false, fmt.Errorf("close: reason is required") } tx, err := d.BeginTx(ctx, nil) @@ -1473,17 +1480,23 @@ func (d *Store) closeIssueWithEvents( } defer func() { _ = tx.Rollback() }() - issue, projectName, err := lookupIssueForEvent(ctx, tx, issueID) + issue, projectName, err := lookupIssueForEvent(ctx, tx, p.IssueID) if err != nil { return db.Issue{}, nil, false, err } + if p.ExpectedProjectID != 0 && issue.ProjectID != p.ExpectedProjectID { + return db.Issue{}, nil, false, db.ErrIssueProjectChanged + } + if p.IfMatchRev != nil && issue.Revision != *p.IfMatchRev { + return db.Issue{}, nil, false, &db.RevisionConflictError{CurrentRevision: issue.Revision} + } if issue.Status == "closed" { if err := tx.Commit(); err != nil { return db.Issue{}, nil, false, err } return issue, nil, false, nil } - if hasOpen, err := txHasOpenChildren(ctx, tx, issueID); err != nil { + if hasOpen, err := txHasOpenChildren(ctx, tx, p.IssueID); err != nil { return db.Issue{}, nil, false, err } else if hasOpen { return db.Issue{}, nil, false, db.ErrOpenChildren @@ -1495,7 +1508,7 @@ func (d *Store) closeIssueWithEvents( closed_reason = ?, closed_at = ?, updated_at = ? - WHERE id = ?`, reason, closedAt, closedAt, issueID); err != nil { + WHERE id = ?`, p.Reason, closedAt, closedAt, p.IssueID); err != nil { return db.Issue{}, nil, false, fmt.Errorf("close: %w", err) } @@ -1508,7 +1521,7 @@ func (d *Store) closeIssueWithEvents( // at close" (non-nil empty) from "legacy event that predates these // fields" (nil) — the audit projection falls back to a live links // lookup only for the legacy case. - parentUID, parentSID, hasParent, err := txParentIdentity(ctx, tx, issueID) + parentUID, parentSID, hasParent, err := txParentIdentity(ctx, tx, p.IssueID) if err != nil { return db.Issue{}, nil, false, err } @@ -1518,19 +1531,23 @@ func (d *Store) closeIssueWithEvents( *parentSIDForPayload = parentSID } payloadBytes, err := json.Marshal(struct { - Reason string `json:"reason"` - ClosedAt string `json:"closed_at"` - Message string `json:"message,omitempty"` - Evidence []db.Evidence `json:"evidence,omitempty"` - ParentUID *string `json:"parent_uid,omitempty"` - ParentShortID *string `json:"parent_short_id,omitempty"` + Reason string `json:"reason"` + ClosedAt string `json:"closed_at"` + Message string `json:"message,omitempty"` + Evidence []db.Evidence `json:"evidence,omitempty"` + ParentUID *string `json:"parent_uid,omitempty"` + ParentShortID *string `json:"parent_short_id,omitempty"` + IdempotencyKey string `json:"idempotency_key,omitempty"` + IdempotencyFingerprint string `json:"idempotency_fingerprint,omitempty"` }{ - Reason: reason, - ClosedAt: closedAt, - Message: message, - Evidence: evidence, - ParentUID: parentUIDForPayload, - ParentShortID: parentSIDForPayload, + Reason: p.Reason, + ClosedAt: closedAt, + Message: p.Message, + Evidence: p.Evidence, + ParentUID: parentUIDForPayload, + ParentShortID: parentSIDForPayload, + IdempotencyKey: p.IdempotencyKey, + IdempotencyFingerprint: p.IdempotencyFingerprint, }) if err != nil { return db.Issue{}, nil, false, fmt.Errorf("close payload: %w", err) @@ -1541,7 +1558,7 @@ func (d *Store) closeIssueWithEvents( ProjectName: projectName, IssueID: &issue.ID, Type: "issue.closed", - Actor: actor, + Actor: p.Actor, Payload: string(payloadBytes), }) if err != nil { @@ -1554,7 +1571,7 @@ func (d *Store) closeIssueWithEvents( IssueID: issue.ID, IssueUID: issue.UID, EventType: "issue.closed", - Actor: actor, + Actor: p.Actor, HolderInstanceUID: d.InstanceUID(), }) if err != nil { @@ -1565,9 +1582,9 @@ func (d *Store) closeIssueWithEvents( events = append(events, auditEvents...) lastEventID = auditEvents[len(auditEvents)-1].ID } - if reason == "done" && issue.RecurrenceID != nil && issue.OccurrenceKey != nil { + if p.Reason == "done" && issue.RecurrenceID != nil && issue.OccurrenceKey != nil { if _, err := d.materializeNextTx(ctx, tx, *issue.RecurrenceID, - *issue.OccurrenceKey, actor); err != nil { + *issue.OccurrenceKey, p.Actor); err != nil { return db.Issue{}, nil, false, fmt.Errorf("materialize next recurrence: %w", err) } generated, err := eventsAfterTx(ctx, tx, lastEventID) @@ -1576,12 +1593,14 @@ func (d *Store) closeIssueWithEvents( } events = append(events, generated...) } - updated, err := issueByIDTx(ctx, tx, issueID) + updated, err := issueByIDTx(ctx, tx, p.IssueID) if err != nil { return db.Issue{}, nil, false, err } if err := tx.Commit(); err != nil { - return db.Issue{}, nil, false, err + // Preserve the attempted result so the daemon can match its keyed + // receipt after an ambiguous commit response and publish every event. + return updated, events, true, err } return updated, events, true, nil } diff --git a/internal/db/sqlitestore/queries_idempotency.go b/internal/db/sqlitestore/queries_idempotency.go index cf6af80e..1578b977 100644 --- a/internal/db/sqlitestore/queries_idempotency.go +++ b/internal/db/sqlitestore/queries_idempotency.go @@ -128,6 +128,15 @@ func fingerprintCore(title, body string, owner *string, labels []string, sortedL // at-or-after `since`. Returns nil when no match. Uses the partial index // idx_events_idempotency declared in 0001_init.sql. func (d *Store) LookupIdempotency(ctx context.Context, projectID int64, key string, since time.Time) (*db.IdempotencyMatch, error) { + return d.LookupIssueMutationIdempotency(ctx, projectID, "issue.created", key, since) +} + +// LookupIssueMutationIdempotency finds the newest recent issue event of the +// requested type carrying key. Event types are bound query values so callers +// can use the same retry contract for additional issue mutations. +func (d *Store) LookupIssueMutationIdempotency( + ctx context.Context, projectID int64, eventType, key string, since time.Time, +) (*db.IdempotencyMatch, error) { const q = ` SELECT e.id, e.uid, e.origin_instance_uid, e.project_id, p.uid, e.project_name, e.issue_id, e.issue_uid, @@ -136,13 +145,13 @@ func (d *Store) LookupIdempotency(ctx context.Context, projectID int64, key stri json_extract(e.payload, '$.idempotency_fingerprint') FROM events e JOIN projects p ON p.id = e.project_id - WHERE e.type = 'issue.created' + WHERE e.type = ? AND e.project_id = ? AND json_extract(e.payload, '$.idempotency_key') = ? AND e.created_at >= ? ORDER BY e.id DESC LIMIT 1` - row := d.QueryRowContext(ctx, q, projectID, key, since.UTC().Format(sqliteTimeFormat)) + row := d.QueryRowContext(ctx, q, eventType, projectID, key, since.UTC().Format(sqliteTimeFormat)) var ( evt db.Event @@ -155,30 +164,33 @@ func (d *Store) LookupIdempotency(ctx context.Context, projectID int64, key stri return nil, nil } if err != nil { - return nil, fmt.Errorf("lookup idempotency: %w", err) + return nil, fmt.Errorf("lookup issue mutation idempotency: %w", err) } if evt.IssueID == nil { // Defensive: an issue.created event without an issue_id is malformed. - return nil, fmt.Errorf("idempotency match has no issue_id") + return nil, fmt.Errorf("issue mutation idempotency match has no issue_id") } // Carveout (spec §6): idempotency-key collision detection sees the issue // even if it has been soft-deleted, so it can report the right mismatch. // IssueByID returns rows regardless of deleted_at, matching that intent. issue, err := d.IssueByID(ctx, *evt.IssueID) if err != nil { - return nil, fmt.Errorf("idempotency match issue: %w", err) + return nil, fmt.Errorf("issue mutation idempotency match issue: %w", err) } return &db.IdempotencyMatch{ IssueID: issue.ID, + IssueUID: issue.UID, IssueShortID: issue.ShortID, Fingerprint: fp.String, Event: evt, }, nil } -// LookupCommentIdempotency finds the newest recent issue.commented event -// carrying key and returns the comment that event created. -func (d *Store) LookupCommentIdempotency(ctx context.Context, projectID int64, key string, since time.Time) (*db.CommentIdempotencyMatch, error) { +// LookupCommentIdempotency finds the newest recent issue.commented event on +// issueUID carrying key and returns the comment that event created. +func (d *Store) LookupCommentIdempotency( + ctx context.Context, issueUID, key string, since time.Time, +) (*db.CommentIdempotencyMatch, error) { const q = ` SELECT e.id, e.uid, e.origin_instance_uid, e.project_id, p.uid, e.project_name, e.issue_id, e.issue_uid, @@ -187,12 +199,12 @@ func (d *Store) LookupCommentIdempotency(ctx context.Context, projectID int64, k FROM events e JOIN projects p ON p.id = e.project_id WHERE e.type = 'issue.commented' - AND e.project_id = ? + AND e.issue_uid = ? AND json_extract(e.payload, '$.idempotency_key') = ? AND e.created_at >= ? ORDER BY e.id DESC LIMIT 1` - row := d.QueryRowContext(ctx, q, projectID, key, since.UTC().Format(sqliteTimeFormat)) + row := d.QueryRowContext(ctx, q, issueUID, key, since.UTC().Format(sqliteTimeFormat)) var evt db.Event err := row.Scan(&evt.ID, &evt.UID, &evt.OriginInstanceUID, &evt.ProjectID, &evt.ProjectUID, &evt.ProjectName, @@ -204,8 +216,8 @@ func (d *Store) LookupCommentIdempotency(ctx context.Context, projectID int64, k if err != nil { return nil, fmt.Errorf("lookup comment idempotency: %w", err) } - if evt.IssueID == nil { - return nil, fmt.Errorf("comment idempotency match has no issue_id") + if evt.IssueID == nil || evt.IssueUID == nil { + return nil, fmt.Errorf("comment idempotency match has no issue identity") } var payload struct { CommentUID string `json:"comment_uid"` @@ -226,6 +238,7 @@ func (d *Store) LookupCommentIdempotency(ctx context.Context, projectID int64, k return nil, fmt.Errorf("comment idempotency match comment: %w", err) } return &db.CommentIdempotencyMatch{ - Comment: comment, Fingerprint: payload.Fingerprint, Event: evt, + Comment: comment, IssueUID: *evt.IssueUID, + Fingerprint: payload.Fingerprint, Event: evt, }, nil } diff --git a/internal/db/storage.go b/internal/db/storage.go index 7f6b4757..c5f6d921 100644 --- a/internal/db/storage.go +++ b/internal/db/storage.go @@ -76,6 +76,10 @@ type Storage interface { EditIssueAtomic(ctx context.Context, p EditIssueAtomicParams) (EditIssueAtomicResult, error) CloseIssue(ctx context.Context, issueID int64, reason, actor, message string, evidence []Evidence) (Issue, *Event, bool, error) CloseIssueWithEvents(ctx context.Context, issueID int64, reason, actor, message string, evidence []Evidence) (Issue, []Event, bool, error) + // CloseIssueGuarded preserves its attempted issue, event batch, and changed + // result when commit returns an ambiguous error. A caller holding an + // idempotency lock can compare those events with the persisted receipt. + CloseIssueGuarded(ctx context.Context, p CloseIssueParams) (Issue, []Event, bool, error) ReopenIssue(ctx context.Context, issueID int64, actor string) (Issue, *Event, bool, error) SoftDeleteIssue(ctx context.Context, issueID int64, actor string) (Issue, *Event, bool, error) RestoreIssue(ctx context.Context, issueID int64, actor string) (Issue, *Event, bool, error) @@ -142,12 +146,15 @@ type Storage interface { MaxEventID(ctx context.Context) (int64, error) MaxLocalOriginEventID(ctx context.Context, projectID int64) (int64, error) MaxFederationBaselineEventID(ctx context.Context, projectID, sinceEventID int64) (int64, error) - // AcquireIdempotencyLock serializes one project/key create decision until + // AcquireIdempotencyLock serializes one project/key mutation decision until // the returned release function runs. Implementations must coordinate every // daemon that can write the same backend, not only goroutines in one server. AcquireIdempotencyLock(ctx context.Context, projectID int64, key string) (release func() error, err error) LookupIdempotency(ctx context.Context, projectID int64, key string, since time.Time) (*IdempotencyMatch, error) - LookupCommentIdempotency(ctx context.Context, projectID int64, key string, since time.Time) (*CommentIdempotencyMatch, error) + LookupIssueMutationIdempotency(ctx context.Context, projectID int64, eventType, key string, since time.Time) (*IdempotencyMatch, error) + // LookupCommentIdempotency scopes comment keys by issue UID so a committed + // receipt survives a later project move without making keys global. + LookupCommentIdempotency(ctx context.Context, issueUID, key string, since time.Time) (*CommentIdempotencyMatch, error) InsertCloseThrottledEvent(ctx context.Context, issueID int64, actor string, payload CloseThrottledPayload) (Event, error) RecentSiblingCloses(ctx context.Context, parentIssueID, excludeIssueID int64, actor string, since time.Time) ([]Event, error) RecentSameMessageClose(ctx context.Context, parentIssueID, excludeIssueID int64, actor, normalizedMessage string, since time.Time) (*Event, error) diff --git a/internal/db/types.go b/internal/db/types.go index 1aa9e257..25008667 100644 --- a/internal/db/types.go +++ b/internal/db/types.go @@ -554,11 +554,12 @@ type SearchCandidate struct { MatchedIn []string `json:"matched_in"` } -// IdempotencyMatch is the payload returned by LookupIdempotency. The Event row -// is included so the handler can populate `original_event` in the reuse-case -// MutationResponse without a second query. +// IdempotencyMatch is a previously committed issue mutation matched through +// its event payload. The Event row lets handlers return the original mutation +// receipt without a second query. type IdempotencyMatch struct { IssueID int64 + IssueUID string IssueShortID string Fingerprint string Event Event @@ -568,6 +569,7 @@ type IdempotencyMatch struct { // through its issue.commented event payload. type CommentIdempotencyMatch struct { Comment Comment + IssueUID string Fingerprint string Event Event } diff --git a/internal/mcp/close_retry_test.go b/internal/mcp/close_retry_test.go new file mode 100644 index 00000000..5e135dd4 --- /dev/null +++ b/internal/mcp/close_retry_test.go @@ -0,0 +1,199 @@ +package mcpserver + +import ( + "encoding/json" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestCloseForwardsRetryGuardsAndReturnsOriginalReceipt(t *testing.T) { + var idempotencyKey, ifMatch, retryProtocol string + client := reviewClient(t, func(writer http.ResponseWriter, request *http.Request) { + switch { + case request.URL.Path == "/api/v1/projects": + writeJSON(writer, map[string]any{"projects": []any{ + projectJSON(1, "01HAAAAAAAAAAAAAAAAAAAAAAA", "spoke-project"), + }}) + case strings.HasSuffix(request.URL.Path, "/actions/close"): + idempotencyKey = request.Header.Get("Idempotency-Key") + ifMatch = request.Header.Get("If-Match") + var body map[string]any + require.NoError(t, json.NewDecoder(request.Body).Decode(&body)) + retryProtocol, _ = body["retry_protocol"].(string) + issue := issueJSON(1, "spoke-project", "abc1") + issue["status"] = "closed" + writeJSON(writer, map[string]any{ + "issue": issue, + "event": nil, + "original_event": closeRetryEventJSON(), + "changed": false, + "reused": true, + }) + default: + http.NotFound(writer, request) + } + }) + handlers := toolHandlers{options: Options{ + Client: client, Scope: NewAllScope(), Actor: "example-agent", + }} + revision := int64(7) + _, output, err := handlers.close(t.Context(), nil, CloseInput{ + Ref: "spoke-project#abc1", Reason: "wontfix", + Message: "Reviewed the request and recorded why the work should stop here.", + IdempotencyKey: "close-request-1", Revision: &revision, + }) + require.NoError(t, err) + require.Equal(t, "close-request-1", idempotencyKey) + require.Equal(t, `"rev-7"`, ifMatch) + require.Equal(t, "close-v1", retryProtocol) + require.NotNil(t, output.Reused) + require.True(t, *output.Reused) + require.NotNil(t, output.Event) + require.Equal(t, "01HCCCCCCCCCCCCCCCCCCCCCCC", output.Event.UID) +} + +func TestCloseRejectsReusedIssueMovedOutsideFixedScope(t *testing.T) { + const ( + allowedProjectUID = "01HAAAAAAAAAAAAAAAAAAAAAAA" + foreignProjectUID = "01HBBBBBBBBBBBBBBBBBBBBBBB" + ) + client := reviewClient(t, func(writer http.ResponseWriter, request *http.Request) { + switch { + case request.URL.Path == "/api/v1/projects": + writeJSON(writer, map[string]any{"projects": []any{ + projectJSON(1, allowedProjectUID, "spoke-project"), + projectJSON(2, foreignProjectUID, "other-project"), + }}) + case strings.HasSuffix(request.URL.Path, "/actions/close"): + issue := issueJSON(2, "other-project", "def1") + issue["project_uid"] = foreignProjectUID + issue["status"] = "closed" + writeJSON(writer, map[string]any{ + "issue": issue, "original_event": closeRetryEventJSON(), + "changed": false, "reused": true, + }) + default: + http.NotFound(writer, request) + } + }) + scope, err := NewAllowlistScope([]ProjectIdentity{{ + ID: 1, UID: allowedProjectUID, Name: "spoke-project", + }}) + require.NoError(t, err) + handlers := toolHandlers{options: Options{ + Client: client, Scope: scope, Actor: "example-agent", + }} + + _, _, err = handlers.close(t.Context(), nil, CloseInput{ + Ref: "spoke-project#abc1", Reason: "wontfix", + Message: "Reviewed the request and recorded why the work should stop here.", + IdempotencyKey: "close-request-1", + }) + require.ErrorContains(t, err, "outside the MCP startup scope") +} + +func TestCloseUsesCurrentProjectForReusedIssueMovedWithinScope(t *testing.T) { + const ( + sourceProjectUID = "01HAAAAAAAAAAAAAAAAAAAAAAA" + targetProjectUID = "01HBBBBBBBBBBBBBBBBBBBBBBB" + ) + client := reviewClient(t, func(writer http.ResponseWriter, request *http.Request) { + switch { + case request.URL.Path == "/api/v1/projects": + writeJSON(writer, map[string]any{"projects": []any{ + projectJSON(1, sourceProjectUID, "source-project"), + projectJSON(2, targetProjectUID, "target-project"), + }}) + case strings.HasSuffix(request.URL.Path, "/actions/close"): + issue := issueJSON(2, "target-project", "def1") + issue["project_uid"] = targetProjectUID + issue["status"] = "closed" + writeJSON(writer, map[string]any{ + "issue": issue, "original_event": closeRetryEventJSON(), + "changed": false, "reused": true, + }) + default: + http.NotFound(writer, request) + } + }) + scope, err := NewAllowlistScope([]ProjectIdentity{ + {ID: 1, UID: sourceProjectUID, Name: "source-project"}, + {ID: 2, UID: targetProjectUID, Name: "target-project"}, + }) + require.NoError(t, err) + handlers := toolHandlers{options: Options{ + Client: client, Scope: scope, Actor: "example-agent", + }} + + _, output, err := handlers.close(t.Context(), nil, CloseInput{ + Ref: "source-project#abc1", Reason: "wontfix", + Message: "Reviewed the request and recorded why the work should stop here.", + IdempotencyKey: "close-request-1", + }) + require.NoError(t, err) + require.Equal(t, ProjectIdentity{ID: 2, UID: targetProjectUID, Name: "target-project"}, output.Project) + require.Equal(t, "target-project#def1", output.Issue.QualifiedRef) +} + +func TestCloseRetryProtocolMakesLegacyDaemonRejectRequest(t *testing.T) { + var closeCalls, mutations int + var retryProtocol string + client := reviewClient(t, func(writer http.ResponseWriter, request *http.Request) { + switch { + case request.URL.Path == "/api/v1/projects": + writeJSON(writer, map[string]any{"projects": []any{ + projectJSON(1, "01HAAAAAAAAAAAAAAAAAAAAAAA", "spoke-project"), + }}) + case strings.HasSuffix(request.URL.Path, "/actions/close"): + closeCalls++ + var body map[string]any + require.NoError(t, json.NewDecoder(request.Body).Decode(&body)) + retryProtocol, _ = body["retry_protocol"].(string) + if retryProtocol != "" { + writer.Header().Set("Content-Type", "application/json") + writer.WriteHeader(http.StatusBadRequest) + writeJSON(writer, map[string]any{ + "status": http.StatusBadRequest, + "error": map[string]any{ + "code": "validation", "message": "retry_protocol: unexpected property", + }, + }) + return + } + mutations++ + writeJSON(writer, map[string]any{ + "issue": issueJSON(1, "spoke-project", "abc1"), "changed": true, + }) + default: + http.NotFound(writer, request) + } + }) + handlers := toolHandlers{options: Options{ + Client: client, Scope: NewAllScope(), Actor: "example-agent", + }} + _, _, err := handlers.close(t.Context(), nil, CloseInput{ + Ref: "spoke-project#abc1", Reason: "wontfix", + Message: "Reviewed the request and recorded why the work should stop here.", + IdempotencyKey: "close-request-1", + }) + require.Error(t, err) + require.Equal(t, "close-v1", retryProtocol) + require.Equal(t, 1, closeCalls) + require.Zero(t, mutations) +} + +func closeRetryEventJSON() map[string]any { + return map[string]any{ + "id": 9, "uid": "01HCCCCCCCCCCCCCCCCCCCCCCC", + "origin_instance_uid": "01HDDDDDDDDDDDDDDDDDDDDDDD", + "project_id": 1, "project_uid": "01HAAAAAAAAAAAAAAAAAAAAAAA", + "project_name": "spoke-project", "type": "issue.closed", + "actor": "example-agent", "payload": `{}`, + "hlc_physical_ms": 1, "hlc_counter": 0, + "content_hash": strings.Repeat("a", 64), + "created_at": "2026-08-11T00:00:00Z", + } +} diff --git a/internal/mcp/handlers.go b/internal/mcp/handlers.go index 88829c1c..e2699a73 100644 --- a/internal/mcp/handlers.go +++ b/internal/mcp/handlers.go @@ -757,8 +757,8 @@ func (h toolHandlers) close(ctx context.Context, _ *sdkmcp.CallToolRequest, inpu if err != nil { return nil, MutationOutput{}, err } - reason := generated.ActionRequestBodyReason(input.Reason) - if err := reason.Validate(); err != nil || reason == generated.Empty { + reason := generated.CloseActionRequestBodyReason(input.Reason) + if err := reason.Validate(); err != nil || reason == generated.CloseActionRequestBodyReasonEmpty { return nil, MutationOutput{}, fmt.Errorf("reason must be one of done, wontfix, duplicate, superseded, or audit-no-change") } if strings.TrimSpace(input.Message) == "" { @@ -772,20 +772,75 @@ func (h toolHandlers) close(ctx context.Context, _ *sdkmcp.CallToolRequest, inpu } evidence = append(evidence, converted) } + var headers *generated.CloseIssueHeaders + var retryProtocol *generated.CloseActionRequestBodyRetryProtocol + if input.IdempotencyKey != "" || input.Revision != nil { + headers = &generated.CloseIssueHeaders{} + protocol := generated.CloseV1 + retryProtocol = &protocol + if input.IdempotencyKey != "" { + headers.IdempotencyKey = &input.IdempotencyKey + } + if input.Revision != nil { + if *input.Revision < 0 { + return nil, MutationOutput{}, errors.New("revision must not be negative") + } + ifMatch := `"rev-` + strconv.FormatInt(*input.Revision, 10) + `"` + headers.IfMatch = &ifMatch + } + } response, err := h.options.Client.CloseIssue(ctx, &generated.CloseIssueRequestOptions{ PathParams: &generated.CloseIssuePath{ProjectID: project.ID, Ref: ref}, Body: &generated.CloseIssueBody{ - Actor: &h.options.Actor, - Reason: &reason, - Message: &input.Message, - Evidence: evidence, - DryRun: optionalTrue(input.DryRun), + Actor: &h.options.Actor, + Reason: &reason, + Message: &input.Message, + Evidence: evidence, + DryRun: optionalTrue(input.DryRun), + RetryProtocol: retryProtocol, }, + Header: headers, }) if err != nil { return nil, MutationOutput{}, h.scopedCloseError(err) } - return successResult(), h.mutation(project, response.Issue, response.Changed, response.Reused, &response.Event), nil + event := &response.Event + if response.OriginalEvent != nil { + event = response.OriginalEvent + } + project, err = h.closeMutationProject(ctx, project, response.Issue) + if err != nil { + return nil, MutationOutput{}, err + } + return successResult(), h.mutation(project, response.Issue, response.Changed, response.Reused, event), nil +} + +// closeMutationProject keeps replayed close receipts inside the immutable MCP +// startup scope. A retry can return an issue that moved after the original +// close, so the request project is only valid while its UID still matches. +func (h toolHandlers) closeMutationProject( + ctx context.Context, requestProject ProjectIdentity, issue generated.Issue, +) (ProjectIdentity, error) { + projectUID := "" + if issue.ProjectUID != nil { + projectUID = *issue.ProjectUID + } + if projectUID == "" { + return ProjectIdentity{}, errors.New("close response did not include the issue project UID") + } + if requestProject.UID == projectUID || requestProject.UID == "" && requestProject.ID == issue.ProjectID { + return requestProject, nil + } + projects, err := h.options.Scope.Projects(ctx, h.options.Client, false) + if err != nil { + return ProjectIdentity{}, err + } + for _, project := range projects { + if project.UID == projectUID { + return project, nil + } + } + return ProjectIdentity{}, errors.New("close response issue moved outside the MCP startup scope") } // Close-guard refusals render child, sibling-cohort, and prior-close diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 720e1c69..e067dc90 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -663,11 +663,13 @@ type Evidence struct { // CloseInput makes the completion assertion explicit. type CloseInput struct { - Ref string `json:"ref" jsonschema:"Issue reference; use project#ref in multi-project mode"` - Reason string `json:"reason" jsonschema:"Close reason: done, wontfix, duplicate, superseded, or audit-no-change"` - Message string `json:"message" jsonschema:"Substantive completion message"` - Evidence []Evidence `json:"evidence" jsonschema:"Typed evidence that supports the close"` - DryRun bool `json:"dry_run,omitempty" jsonschema:"Validate the close without changing the issue"` + Ref string `json:"ref" jsonschema:"Issue reference; use project#ref in multi-project mode"` + Reason string `json:"reason" jsonschema:"Close reason: done, wontfix, duplicate, superseded, or audit-no-change"` + Message string `json:"message" jsonschema:"Substantive completion message"` + Evidence []Evidence `json:"evidence" jsonschema:"Typed evidence that supports the close"` + DryRun bool `json:"dry_run,omitempty" jsonschema:"Validate the close without changing the issue"` + IdempotencyKey string `json:"idempotency_key,omitempty" jsonschema:"Stable unique key for safe retries"` + Revision *int64 `json:"revision,omitempty" jsonschema:"Expected current issue revision for a conditional close"` } // ReopenInput reactivates closed work. diff --git a/pkg/client/generated/client_options.go b/pkg/client/generated/client_options.go index 81a473c5..1c17ea8f 100644 --- a/pkg/client/generated/client_options.go +++ b/pkg/client/generated/client_options.go @@ -2446,6 +2446,7 @@ func (o *ClaimIssueRequestOptions) GetHeader() (map[string]string, error) { type CloseIssueRequestOptions struct { PathParams *CloseIssuePath Body *CloseIssueBody + Header *CloseIssueHeaders } // Validate validates all the fields in the options. @@ -2468,6 +2469,14 @@ func (o *CloseIssueRequestOptions) Validate() error { } } } + + if o.Header != nil { + if v, ok := any(o.Header).(runtime.Validator); ok { + if err := v.Validate(); err != nil { + errors = errors.Append("Header", err) + } + } + } if len(errors) == 0 { return nil } @@ -2492,7 +2501,7 @@ func (o *CloseIssueRequestOptions) GetBody() any { // GetHeader returns the headers as a map. func (o *CloseIssueRequestOptions) GetHeader() (map[string]string, error) { - return nil, nil + return runtime.AsMap[string](o.Header) } // DeleteIssueRequestOptions is the options needed to make a request to DeleteIssue. diff --git a/pkg/client/generated/enums.go b/pkg/client/generated/enums.go index c764bcc8..ec29b390 100644 --- a/pkg/client/generated/enums.go +++ b/pkg/client/generated/enums.go @@ -46,6 +46,61 @@ func (a ActionRequestBodySource) Validate() error { } } +type CloseActionRequestBodyReason string + +const ( + CloseActionRequestBodyReasonAuditNoChange CloseActionRequestBodyReason = "audit-no-change" + CloseActionRequestBodyReasonDone CloseActionRequestBodyReason = "done" + CloseActionRequestBodyReasonDuplicate CloseActionRequestBodyReason = "duplicate" + CloseActionRequestBodyReasonEmpty CloseActionRequestBodyReason = "" + CloseActionRequestBodyReasonSuperseded CloseActionRequestBodyReason = "superseded" + CloseActionRequestBodyReasonWontfix CloseActionRequestBodyReason = "wontfix" +) + +// Validate checks if the CloseActionRequestBodyReason value is valid +func (c CloseActionRequestBodyReason) Validate() error { + switch c { + case CloseActionRequestBodyReasonAuditNoChange, CloseActionRequestBodyReasonDone, CloseActionRequestBodyReasonDuplicate, CloseActionRequestBodyReasonEmpty, CloseActionRequestBodyReasonSuperseded, CloseActionRequestBodyReasonWontfix: + return nil + default: + return runtime.NewValidationErrorsFromString("Enum", fmt.Sprintf("must be a valid CloseActionRequestBodyReason value, got: %v", c)) + } +} + +type CloseActionRequestBodyRetryProtocol string + +const ( + CloseActionRequestBodyRetryProtocolEmpty CloseActionRequestBodyRetryProtocol = "" + CloseV1 CloseActionRequestBodyRetryProtocol = "close-v1" +) + +// Validate checks if the CloseActionRequestBodyRetryProtocol value is valid +func (c CloseActionRequestBodyRetryProtocol) Validate() error { + switch c { + case CloseActionRequestBodyRetryProtocolEmpty, CloseV1: + return nil + default: + return runtime.NewValidationErrorsFromString("Enum", fmt.Sprintf("must be a valid CloseActionRequestBodyRetryProtocol value, got: %v", c)) + } +} + +type CloseActionRequestBodySource string + +const ( + CloseActionRequestBodySourceEmpty CloseActionRequestBodySource = "" + CloseActionRequestBodySourceTui CloseActionRequestBodySource = "tui" +) + +// Validate checks if the CloseActionRequestBodySource value is valid +func (c CloseActionRequestBodySource) Validate() error { + switch c { + case CloseActionRequestBodySourceEmpty, CloseActionRequestBodySourceTui: + return nil + default: + return runtime.NewValidationErrorsFromString("Enum", fmt.Sprintf("must be a valid CloseActionRequestBodySource value, got: %v", c)) + } +} + type CreateInitialLinkBodyType string const ( diff --git a/pkg/client/generated/headers.go b/pkg/client/generated/headers.go index cbb7af0a..117b24d1 100644 --- a/pkg/client/generated/headers.go +++ b/pkg/client/generated/headers.go @@ -43,6 +43,11 @@ type CreateIssueHeaders struct { IdempotencyKey *string `json:"Idempotency-Key,omitempty"` } +type CloseIssueHeaders struct { + IdempotencyKey *string `json:"Idempotency-Key,omitempty"` + IfMatch *string `json:"If-Match,omitempty"` +} + type DeleteIssueHeaders struct { XKataConfirm *string `json:"X-Kata-Confirm,omitempty"` } diff --git a/pkg/client/generated/payloads.go b/pkg/client/generated/payloads.go index 7982769f..71a61745 100644 --- a/pkg/client/generated/payloads.go +++ b/pkg/client/generated/payloads.go @@ -48,7 +48,7 @@ type AssignIssueBody = AssignRequestBody type ClaimIssueBody = ClaimRequestBody -type CloseIssueBody = ActionRequestBody +type CloseIssueBody = CloseActionRequestBody type DeleteIssueBody = DestructiveActionRequestBody diff --git a/pkg/client/generated/types.go b/pkg/client/generated/types.go index 8e110e16..d24b9e26 100644 --- a/pkg/client/generated/types.go +++ b/pkg/client/generated/types.go @@ -461,6 +461,52 @@ func (c ClaimViolationOut) Validate() error { return runtime.ConvertValidatorError(typesValidator.Struct(c)) } +type CloseActionRequestBody struct { + Actor *string `json:"actor,omitempty"` + DryRun *bool `json:"dry_run,omitempty"` + Evidence []Evidence `json:"evidence,omitempty"` + Message *string `json:"message,omitempty"` + Reason *CloseActionRequestBodyReason `json:"reason,omitempty"` + RetryProtocol *CloseActionRequestBodyRetryProtocol `json:"retry_protocol,omitempty"` + Source *CloseActionRequestBodySource `json:"source,omitempty"` +} + +func (c CloseActionRequestBody) Validate() error { + var errors runtime.ValidationErrors + for i, item := range c.Evidence { + if v, ok := any(item).(runtime.Validator); ok { + if err := v.Validate(); err != nil { + errors = errors.Append(fmt.Sprintf("Evidence[%d]", i), err) + } + } + } + if c.Reason != nil { + if v, ok := any(c.Reason).(runtime.Validator); ok { + if err := v.Validate(); err != nil { + errors = errors.Append("Reason", err) + } + } + } + if c.RetryProtocol != nil { + if v, ok := any(c.RetryProtocol).(runtime.Validator); ok { + if err := v.Validate(); err != nil { + errors = errors.Append("RetryProtocol", err) + } + } + } + if c.Source != nil { + if v, ok := any(c.Source).(runtime.Validator); ok { + if err := v.Validate(); err != nil { + errors = errors.Append("Source", err) + } + } + } + if len(errors) == 0 { + return nil + } + return errors +} + type Comment struct { Author string `json:"author" validate:"required"` Body string `json:"body" validate:"required"` diff --git a/pkg/client/openapi.yaml b/pkg/client/openapi.yaml index 9a621078..9444bc8b 100644 --- a/pkg/client/openapi.yaml +++ b/pkg/client/openapi.yaml @@ -286,6 +286,40 @@ components: - issue_uid - at type: object + CloseActionRequestBody: + additionalProperties: false + properties: + actor: + type: string + dry_run: + type: boolean + evidence: + items: + $ref: "#/components/schemas/Evidence" + nullable: true + type: array + message: + type: string + reason: + enum: + - done + - wontfix + - duplicate + - superseded + - audit-no-change + - "" + type: string + retry_protocol: + enum: + - close-v1 + - "" + type: string + source: + enum: + - tui + - "" + type: string + type: object Comment: properties: author: @@ -4363,7 +4397,7 @@ components: type: object info: title: kata - version: 0.14.0 + version: 0.15.0 openapi: 3.0.3 paths: /api/v1/audit/closes: @@ -6035,11 +6069,19 @@ paths: required: true schema: type: string + - in: header + name: Idempotency-Key + schema: + type: string + - in: header + name: If-Match + schema: + type: string requestBody: content: application/json: schema: - $ref: "#/components/schemas/ActionRequestBody" + $ref: "#/components/schemas/CloseActionRequestBody" required: true responses: "200": diff --git a/web/src/lib/api/schema.d.ts b/web/src/lib/api/schema.d.ts index 2b8a9ee7..11321eee 100644 --- a/web/src/lib/api/schema.d.ts +++ b/web/src/lib/api/schema.d.ts @@ -1576,6 +1576,18 @@ export interface components { } & { [key: string]: unknown } + CloseActionRequestBody: { + actor?: string + dry_run?: boolean + evidence?: components['schemas']['Evidence'][] | null + message?: string + /** @enum {string} */ + reason?: 'done' | 'wontfix' | 'duplicate' | 'superseded' | 'audit-no-change' | '' + /** @enum {string} */ + retry_protocol?: 'close-v1' | '' + /** @enum {string} */ + source?: 'tui' | '' + } Comment: { author: string body: string @@ -5356,7 +5368,10 @@ export interface operations { closeIssue: { parameters: { query?: never - header?: never + header?: { + 'Idempotency-Key'?: string + 'If-Match'?: string + } path: { project_id: number ref: string @@ -5365,7 +5380,7 @@ export interface operations { } requestBody: { content: { - 'application/json': components['schemas']['ActionRequestBody'] + 'application/json': components['schemas']['CloseActionRequestBody'] } } responses: {