Files
16f0d307dd fix(realtime): announce comments folded into a queued run (#9702)
A reply posted while the agent already has a queued run in the same thread is
folded into that run on the server: the reply becomes the run's trigger and the
earlier comment joins its covered set. The run's status does not change, so no
task event fired and open issue pages kept the run under the earlier comment,
leaving the new reply with no visible queue row.

Publish task:coalesced after the merge (and after a delegated-failure merge).
Web and desktop already refresh the issue's runs on any task: event, which moves
the queued block under the newest comment it answers; mobile now listens for it
too.

Co-authored-by: Cursor <[email protected]>
Co-authored-by: multica-agent <[email protected]>
2026-10-30 17:45:54 +08:00

4141 lines
169 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package handler
import (
"bytes"
"context"
"encoding/json"
"errors"
"log/slog"
"net/http"
"sort"
"strconv"
"strings"
"time"
"unicode"
"github.com/go-chi/chi/v5"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/internal/logger"
obsmetrics "github.com/multica-ai/multica/server/internal/metrics"
"github.com/multica-ai/multica/server/internal/service"
"github.com/multica-ai/multica/server/internal/util"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/dbid"
"github.com/multica-ai/multica/server/pkg/protocol"
)
type CommentResponse struct {
ID string `json:"id"`
IssueID string `json:"issue_id"`
AuthorType string `json:"author_type"`
AuthorID string `json:"author_id"`
Content string `json:"content"`
Type string `json:"type"`
ParentID *string `json:"parent_id"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
Revision int64 `json:"revision"`
IssueRevision int64 `json:"issue_revision,omitempty"`
ResolvedAt *string `json:"resolved_at"`
ResolvedByType *string `json:"resolved_by_type"`
ResolvedByID *string `json:"resolved_by_id"`
SourceTaskID *string `json:"source_task_id,omitempty"`
// DeletedAt marks a tombstone: a comment deleted while it still had
// replies (#8296). Its content is empty and it carries no attachments or
// reactions; it stays only so the replies keep their direct parent.
// Omitted for every live comment.
DeletedAt *string `json:"deleted_at,omitempty"`
// QuickActionID marks a comment produced by a quick action run (MUL-5465).
// The timeline renders those as a collapsed one-line card instead of the
// raw prompt body. It is NOT settable through this endpoint — there is no
// request field for it — which is exactly why the card keys off this id
// rather than a `type` value the client controls.
QuickActionID *string `json:"quick_action_id,omitempty"`
Reactions []ReactionResponse `json:"reactions"`
Attachments []AttachmentResponse `json:"attachments"`
// Orientation stats — populated only on the roots_only path and omitted in
// every other mode, so the default response shape stays byte-identical for
// existing callers. ReplyCount is the number of descendants in the thread;
// LastActivityAt is the MAX(created_at) across the whole subtree. Together
// they let an agent triage which thread to drill into without fetching any
// replies.
ReplyCount *int `json:"reply_count,omitempty"`
LastActivityAt *string `json:"last_activity_at,omitempty"`
// ContentTruncated is set only under summary=true: true when Content was
// clipped to the summary budget, false when it fit. nil (omitted) means the
// caller did not request a summary projection, so Content is verbatim.
ContentTruncated *bool `json:"content_truncated,omitempty"`
// Fold projection fields — populated only under fold=true, and only on the
// thread ROOT of a resolved thread (mirrors the human timeline fold; see
// foldResolvedThreads). Both are nil/omitted everywhere else: every comment
// under an unresolved thread, and every non-root comment. ThreadResolved
// marks a thread whose discussion was collapsed to its conclusion;
// FoldedCount is how many comments in that thread were dropped from the
// response (0 when the thread was already at root + conclusion). The reader
// pulls the folded comments back with `comment list --full`.
ThreadResolved *bool `json:"thread_resolved,omitempty"`
FoldedCount *int `json:"folded_count,omitempty"`
// TriggerOutcomes is the per-target result of every EXPLICIT @agent / @squad
// mention in this comment (MUL-4525 §2). It is additive and populated only on
// create/edit responses: old clients ignore it. A saved comment whose mention
// was blocked (no invoke permission, target unavailable, runtime offline) now
// reports that here instead of silently dropping the trigger, so the client
// can show "comment posted, but N targets were not triggered".
TriggerOutcomes []CommentTriggerOutcome `json:"trigger_outcomes,omitempty"`
// Supplements lists every running turn this comment steered, one receipt
// per run. The single supplement_* fields mirror the first receipt for
// clients that predate multi-run steering.
Supplements []CommentSupplementResponse `json:"supplements,omitempty"`
SupplementTaskID string `json:"supplement_task_id,omitempty"`
SupplementStatus string `json:"supplement_status,omitempty"`
SupplementFailureReason *string `json:"supplement_failure_reason,omitempty"`
SupplementDeliveredAt *string `json:"supplement_delivered_at,omitempty"`
}
// CommentTriggerOutcome is the per-target result of an explicit @agent / @squad
// mention (MUL-4525 §2). target_id is the id the user mentioned — the agent id,
// or the SQUAD id for a squad mention — so the client correlates it back to the
// mention it rendered without the server echoing a private target's name/owner.
// reason_code is the stable, enumeration-safe admission reason.
type CommentTriggerOutcome struct {
TargetType string `json:"target_type"` // "agent" | "squad"
TargetID string `json:"target_id"`
Status DispatchStatus `json:"status"` // queued | coalesced | deferred | blocked
ReasonCode DispatchReasonCode `json:"reason_code"`
}
func commentToResponse(c db.Comment, reactions []ReactionResponse, attachments []AttachmentResponse) CommentResponse {
if reactions == nil {
reactions = []ReactionResponse{}
}
if attachments == nil {
attachments = []AttachmentResponse{}
}
return CommentResponse{
ID: uuidToString(c.ID),
IssueID: uuidToString(c.IssueID),
AuthorType: c.AuthorType,
AuthorID: uuidToString(c.AuthorID),
Content: c.Content,
Type: c.Type,
ParentID: uuidToPtr(c.ParentID),
CreatedAt: timestampToString(c.CreatedAt),
UpdatedAt: timestampToString(c.UpdatedAt),
Revision: c.Revision,
ResolvedAt: timestampToPtr(c.ResolvedAt),
ResolvedByType: textToPtr(c.ResolvedByType),
ResolvedByID: uuidToPtr(c.ResolvedByID),
SourceTaskID: uuidToPtr(c.SourceTaskID),
DeletedAt: timestampToPtr(c.DeletedAt),
QuickActionID: uuidToPtr(c.QuickActionID),
Reactions: reactions,
Attachments: attachments,
}
}
// Shared bound for plugin comments and input sent to a running agent.
const maxCommentContentBytes = 64 * 1024
// summaryContentRunes bounds comment content under summary=true. 200 runes is
// enough to tell what a comment is about (its opening) while cutting the bulk
// of a long body out of an agent's context budget. Counted in runes, not bytes,
// so multi-byte (e.g. CJK) content is clipped on a character boundary.
const summaryContentRunes = 200
// summarizeContent clips content to summaryContentRunes for the summary
// projection. Returns the (possibly clipped) content and whether it was
// truncated. An ellipsis marks a clip so the reader knows more text exists.
//
// It scans by rune and stops at the (budget+1)th rune rather than allocating a
// full []rune for the whole body — so a pathologically long comment costs only
// the budget, not its full length, under summary mode.
func summarizeContent(content string) (string, bool) {
count := 0
for byteOffset := range content { // range over a string yields rune start offsets
if count == summaryContentRunes {
return content[:byteOffset] + "…", true
}
count++
}
return content, false
}
// foldStat is the per-thread fold annotation attached to a resolved thread's
// root under fold=true. FoldedCount is how many comments in the thread were
// dropped from the response. See CommentResponse.ThreadResolved / FoldedCount.
type foldStat struct {
FoldedCount int
}
// foldResolvedThreads collapses every resolved thread in a COMPLETE-thread
// comment set down to the comments a reader actually needs, mirroring the human
// timeline fold (deriveThreadResolution in
// packages/views/issues/components/thread-utils.ts) so agents see what humans
// see:
//
// - unresolved thread → every comment kept, unchanged, no annotation.
// - reply-resolved thread → keep the thread root + the resolution reply (the
// conclusion), drop every other reply. Root + conclusion, never conclusion
// alone: a conclusion is often referential ("go with what we said above")
// and unreadable without the root question.
// - root-resolved thread → keep only the root, drop every reply (the whole
// discussion was a settled dead-end; the root states the topic).
//
// The thread root of each resolved thread is annotated (foldStat keyed by the
// root's id string) so the response can mark it thread_resolved and report how
// many comments were folded; a reader that needs the dropped discussion pulls
// it back with `comment list --full`.
//
// Resolution is derived exactly as the UI does: the root wins if resolved;
// otherwise the reply with the latest resolved_at is THE resolution. The single
// resolved-per-thread invariant (ClearOtherThreadResolutions) normally makes
// this unambiguous; the latest-wins tiebreak keeps the projection total if an
// older/concurrent write ever left two.
//
// Callers MUST invoke this only on a set containing COMPLETE threads (the
// default list, --recent, and untailed --thread reads). Partial-thread reads
// (--since, --tail) are rejected upstream because a fold computed over a partial
// thread could drop a resolution that was never fetched. As defense in depth,
// any comment whose thread root is absent from the set is treated as its own
// root and kept verbatim, so a partial thread degrades to "unchanged" rather
// than to silent data loss.
func foldResolvedThreads(comments []db.Comment) ([]db.Comment, map[string]foldStat) {
if len(comments) == 0 {
return comments, nil
}
byID := make(map[string]db.Comment, len(comments))
for _, c := range comments {
byID[uuidToString(c.ID)] = c
}
// rootOf walks parent_id up to the thread root within this set. It stops at
// the first comment whose parent is absent (in a complete-thread set that is
// the real root; in a partial set it is the highest ancestor we hold). The
// loop is bounded by len(comments) so an unexpected cycle cannot hang it —
// the PK forbids real cycles, but never trust a graph walk over stored data.
rootOf := func(c db.Comment) db.Comment {
cur := c
for i := 0; i < len(comments); i++ {
if !cur.ParentID.Valid {
return cur
}
parent, ok := byID[uuidToString(cur.ParentID)]
if !ok {
return cur
}
cur = parent
}
return cur
}
type thread struct {
root db.Comment
replies []db.Comment // non-root, in input (chronological) order
}
threads := map[string]*thread{}
for _, c := range comments {
root := rootOf(c)
rid := uuidToString(root.ID)
th := threads[rid]
if th == nil {
th = &thread{root: root}
threads[rid] = th
}
if uuidToString(c.ID) != rid {
th.replies = append(th.replies, c)
}
}
keep := make(map[string]bool, len(comments))
stats := map[string]foldStat{}
for rid, th := range threads {
// Root-resolved: keep only the root.
if th.root.ResolvedAt.Valid {
keep[rid] = true
stats[rid] = foldStat{FoldedCount: len(th.replies)}
continue
}
// Reply-resolved: the latest-resolved reply is the conclusion.
var resolution *db.Comment
for i := range th.replies {
r := &th.replies[i]
if !r.ResolvedAt.Valid {
continue
}
if resolution == nil || r.ResolvedAt.Time.After(resolution.ResolvedAt.Time) {
resolution = r
}
}
if resolution == nil {
// Unresolved thread: keep everything, no annotation.
keep[rid] = true
for _, r := range th.replies {
keep[uuidToString(r.ID)] = true
}
continue
}
keep[rid] = true
keep[uuidToString(resolution.ID)] = true
// Every reply except the conclusion is folded away.
stats[rid] = foldStat{FoldedCount: len(th.replies) - 1}
}
out := make([]db.Comment, 0, len(comments))
for _, c := range comments {
if keep[uuidToString(c.ID)] {
out = append(out, c)
}
}
return out, stats
}
// commentHardCap bounds the comments returned per issue. Sized as a defensive
// safety net rather than a UX paging window: prod p99 is ~30 comments and
// the all-time max observed is ~1.1k, so 2000 leaves ~2x headroom while still
// preventing a runaway response if some user manages to accumulate a wild
// number of rows on a single issue.
//
// This and the thread-completion budgets below are variables only so the cap
// tests can run the same windowing without seeding thousands of rows per
// scenario; nothing outside tests assigns them.
var commentHardCap = 2000
// HeaderCommentsTruncated tells browser and CLI callers that a defensive
// comment-list cap omitted rows. It is deliberately separate from the timeline
// header: /comments has several projections (roots, one thread, since) whose
// truncation cannot be described as a timeline activity kind.
const HeaderCommentsTruncated = "X-Comments-Truncated"
// commentProbeLimit reads one row past the cap so a truncated read can be told
// apart from an issue holding exactly commentHardCap comments. The difference
// matters: exact-cap reads are complete and must not advertise data loss.
func commentProbeLimit() int32 { return int32(commentHardCap) + 1 }
// Budgets for thread completion (completeCommentThreads).
//
// commentThreadContextBudget caps ALL context added outside the newest window:
// ancestors first, then missing siblings/descendants for any thread whose root
// had to be restored. A response is therefore bounded by commentHardCap +
// commentThreadContextBudget comments regardless of thread shape.
// commentThreadMaxDepth bounds each direction of the walk.
//
// Both are needed because reply depth is genuinely unbounded in stored data: the
// general write path saves the exact comment being replied to, so chains can run
// far deeper than the two levels the UI usually renders. Without a budget the
// completion pass would defeat the row cap it is meant to preserve.
var (
commentThreadContextBudget = 2000
commentThreadMaxDepth = 64
)
// ListComments returns chronological comments with bounded thread-aware reads.
// The default newest-comment cap restores complete threads within a context budget
// or drops each incomplete thread as a unit and sets HeaderCommentsTruncated.
//
// roots_only provides activity metadata; summary clips bodies. thread accepts a
// root or descendant anchor; tail limits replies while retaining the root. recent
// limits whole active threads, not individual comments. Resolved-thread folding
// requires complete threads and cannot combine with partial-thread projections.
//
// Cursors pair a timestamp with an ID for tie-breaking: recent pages by thread
// activity, while thread+tail pages by reply creation. Since filtering follows
// the thread window. The validation below enforces supported combinations.
func (h *Handler) ListComments(w http.ResponseWriter, r *http.Request) {
issueID := chi.URLParam(r, "id")
issue, ok := h.loadIssueForUser(w, r, issueID)
if !ok {
return
}
q := r.URL.Query()
var sinceTime pgtype.Timestamptz
if v := q.Get("since"); v != "" {
// RFC3339Nano parsing also accepts timestamps without fractional seconds.
t, err := time.Parse(time.RFC3339Nano, v)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid since parameter; expected RFC3339 format")
return
}
sinceTime = pgtype.Timestamptz{Time: t, Valid: true}
}
threadStr := q.Get("thread")
recentStr := q.Get("recent")
tailStr := q.Get("tail")
beforeTimeStr := q.Get("before")
beforeIDStr := q.Get("before_id")
if beforeIDStr == "" {
// Accept hyphenated alias to match CLI flag convention.
beforeIDStr = q.Get("before-id")
}
rootsOnlyStr := q.Get("roots_only")
if rootsOnlyStr == "" {
// Accept hyphenated alias to match CLI flag convention.
rootsOnlyStr = q.Get("roots-only")
}
rootsOnly := false
if rootsOnlyStr != "" {
switch rootsOnlyStr {
case "true":
rootsOnly = true
case "false":
default:
writeError(w, http.StatusBadRequest, "invalid roots_only parameter; expected boolean")
return
}
}
// summary=true is an orthogonal content projection: it clips each comment's
// content to a fixed budget so an agent can scan a list without pulling full
// bodies into context. It is intentionally NOT mutually exclusive with any
// mode — it composes with the default list, since, thread, recent, and
// roots_only alike.
summary := false
if summaryStr := q.Get("summary"); summaryStr != "" {
switch summaryStr {
case "true":
summary = true
case "false":
default:
writeError(w, http.StatusBadRequest, "invalid summary parameter; expected boolean")
return
}
}
// fold=true is a thread-level projection: it collapses every resolved thread
// in the result to root + conclusion (reply-resolved) or root only
// (root-resolved), reusing the human timeline's fold semantics so an agent
// reading a long issue doesn't pay tokens for settled discussion. Unlike
// summary (a per-comment content clip that composes with everything), fold
// needs WHOLE threads to compute a resolution, so it is rejected on the
// partial-thread reads (since / tail) and on roots_only (which carries no
// replies to fold). It composes with summary, the default list, recent, and
// untailed thread reads.
fold := false
if foldStr := q.Get("fold"); foldStr != "" {
switch foldStr {
case "true":
fold = true
case "false":
default:
writeError(w, http.StatusBadRequest, "invalid fold parameter; expected boolean")
return
}
}
// --- combination validation ----------------------------------------
if fold && sinceTime.Valid {
writeError(w, http.StatusBadRequest, "fold and since are mutually exclusive: since returns a partial thread, and a fold over a partial thread could hide a resolution that was not fetched")
return
}
if fold && tailStr != "" {
writeError(w, http.StatusBadRequest, "fold and tail are mutually exclusive: tail returns a partial thread, which cannot be folded safely")
return
}
if fold && rootsOnly {
writeError(w, http.StatusBadRequest, "fold and roots_only are mutually exclusive: roots_only returns no replies to fold")
return
}
if rootsOnly && threadStr != "" {
writeError(w, http.StatusBadRequest, "roots_only and thread are mutually exclusive")
return
}
if rootsOnly && recentStr != "" {
writeError(w, http.StatusBadRequest, "roots_only and recent are mutually exclusive")
return
}
if rootsOnly && tailStr != "" {
writeError(w, http.StatusBadRequest, "roots_only and tail are mutually exclusive")
return
}
if rootsOnly && (beforeTimeStr != "" || beforeIDStr != "") {
writeError(w, http.StatusBadRequest, "roots_only does not support before / before_id")
return
}
if threadStr != "" && recentStr != "" {
writeError(w, http.StatusBadRequest, "thread and recent are mutually exclusive")
return
}
if tailStr != "" && threadStr == "" {
writeError(w, http.StatusBadRequest, "tail requires thread (it is a thread-scoped limit)")
return
}
if (beforeTimeStr == "") != (beforeIDStr == "") {
writeError(w, http.StatusBadRequest, "before and before_id must be set together (composite cursor)")
return
}
// Cursor needs either a recent window (thread cursor) or a tailed thread
// (reply cursor). A bare cursor would otherwise fall through to the
// default / since path — returning a full timeline that the caller did
// not ask for. Reject loudly so the API surface matches the documented
// semantics.
if beforeTimeStr != "" && recentStr == "" && (threadStr == "" || tailStr == "") {
writeError(w, http.StatusBadRequest, "before / before_id require recent (thread cursor) or thread + tail (reply cursor)")
return
}
// --- parse cursor / recent ----------------------------------------
var beforeCursor pgtype.Timestamptz
var beforeUUID pgtype.UUID
hasCursor := false
if beforeTimeStr != "" {
t, err := time.Parse(time.RFC3339Nano, beforeTimeStr)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid before parameter; expected RFC3339 format")
return
}
beforeCursor = pgtype.Timestamptz{Time: t, Valid: true}
uuid, perr := util.ParseUUID(beforeIDStr)
if perr != nil {
writeError(w, http.StatusBadRequest, "invalid before_id parameter; expected UUID")
return
}
beforeUUID = uuid
hasCursor = true
}
recentN := 0
if recentStr != "" {
n, err := strconv.Atoi(recentStr)
if err != nil || n <= 0 {
writeError(w, http.StatusBadRequest, "invalid recent parameter; expected positive integer")
return
}
if n > commentHardCap {
n = commentHardCap
}
recentN = n
}
// tail=0 is allowed (returns root only — useful for "what is this thread
// about" lookups without dragging any replies into context). Negative
// values are rejected because they'd round-trip to LIMIT -N which
// PostgreSQL flags as a syntax error.
threadTail := -1
threadTailSet := false
if tailStr != "" {
n, err := strconv.Atoi(tailStr)
if err != nil || n < 0 {
writeError(w, http.StatusBadRequest, "invalid tail parameter; expected non-negative integer")
return
}
if n > commentHardCap {
n = commentHardCap
}
threadTail = n
threadTailSet = true
}
result, err := h.fetchCommentsForList(r.Context(), fetchCommentsArgs{
Issue: issue,
Since: sinceTime,
ThreadAnchor: threadStr,
ThreadTail: threadTail,
ThreadTailSet: threadTailSet,
RecentN: recentN,
HasCursor: hasCursor,
BeforeAt: beforeCursor,
BeforeID: beforeUUID,
RootsOnly: rootsOnly,
})
if err != nil {
switch err {
case errCommentThreadNotFound:
writeError(w, http.StatusNotFound, "thread anchor not found in this issue")
return
case errCommentThreadBadID:
writeError(w, http.StatusBadRequest, "invalid thread parameter; expected UUID")
return
default:
writeError(w, http.StatusInternalServerError, "failed to list comments")
return
}
}
// Apply the resolve-aware fold before anything keys off the comment set
// (reaction/attachment grouping, the response array): folding drops comments,
// and the dropped ones should not pay a reactions/attachments round-trip or
// appear in the response.
//
// fetchCommentsForList marks FoldUnsafe only when a returned thread is known
// to be partial. The default capped path is safe: completeCommentThreads
// either restores every affected thread or removes it as one unit. An
// oversized untailed --thread read, however, intentionally keeps only the
// newest replies, so folding that partial thread would invent resolution
// state and folded_count.
var foldInfo map[string]foldStat
if fold && !result.FoldUnsafe {
result.Comments, foldInfo = foldResolvedThreads(result.Comments)
}
commentIDs := make([]pgtype.UUID, len(result.Comments))
for i, c := range result.Comments {
commentIDs[i] = c.ID
}
grouped := h.groupReactions(r, commentIDs)
groupedAtt := h.groupAttachments(r, commentIDs)
resp := make([]CommentResponse, len(result.Comments))
for i, c := range result.Comments {
cid := uuidToString(c.ID)
resp[i] = commentToResponse(c, grouped[cid], groupedAtt[cid])
// Attach roots_only orientation stats when present (nil map elsewhere).
if st, ok := result.RootStats[cid]; ok {
rc := st.ReplyCount
resp[i].ReplyCount = &rc
if st.LastActivityAt.Valid {
la := timestampToString(st.LastActivityAt)
resp[i].LastActivityAt = &la
}
}
// Attach fold annotations on a resolved thread's root (nil map elsewhere;
// keyed by root id, so only root comments match).
if st, ok := foldInfo[cid]; ok {
resolved := true
resp[i].ThreadResolved = &resolved
fc := st.FoldedCount
resp[i].FoldedCount = &fc
}
// Apply the summary projection last so it clips whatever content the
// chosen read mode produced, uniformly across every mode.
if summary {
clipped, truncated := summarizeContent(resp[i].Content)
resp[i].Content = clipped
resp[i].ContentTruncated = &truncated
}
}
// Emit the next cursor as response headers when the page is likely not
// the last one. The cursor's meaning is context-dependent: under recent
// it points at the oldest thread in the page (next page = older threads);
// under thread + tail it points at the oldest reply in the page (next
// page = older replies in the same thread). Headers stay out of the JSON
// body so the default flat-array response shape — which the desktop UI
// and existing callers depend on — is unchanged.
if result.NextBefore != "" && result.NextBeforeID != "" {
w.Header().Set("X-Multica-Next-Before", result.NextBefore)
w.Header().Set("X-Multica-Next-Before-Id", result.NextBeforeID)
}
if result.CommentsTruncated {
w.Header().Set(HeaderCommentsTruncated, "true")
}
writeJSON(w, http.StatusOK, resp)
}
// fetchCommentsArgs bundles the parsed query params so fetchCommentsForList
// stays readable. Sentinel errors below let the caller turn DB-layer outcomes
// into the right HTTP status without leaking SQL details.
//
// ThreadTail is split into a value + a "set" flag because tail=0 is a
// meaningful caller intent (return just the root). A bare int would collapse
// "user did not pass --tail" and "user passed --tail 0" into the same state,
// which would silently downgrade the latter to the full-thread path.
type fetchCommentsArgs struct {
Issue db.Issue
Since pgtype.Timestamptz
RootsOnly bool
ThreadAnchor string
ThreadTail int
ThreadTailSet bool
RecentN int
HasCursor bool
BeforeAt pgtype.Timestamptz
BeforeID pgtype.UUID
}
// fetchCommentsResult carries both the materialised comments and (for the
// recent/thread-grouped path) the cursor to use for the next page. Cursor
// fields are empty strings when there is no next page or the path does not
// support cursors.
type fetchCommentsResult struct {
Comments []db.Comment
NextBefore string
NextBeforeID string
// RootStats carries per-root orientation stats keyed by comment id string.
// Populated only on the roots_only path; nil for every other mode.
RootStats map[string]rootStat
// CommentsTruncated reports that a defensive cap omitted comments. Derived
// from a probe read, never from the final response length.
CommentsTruncated bool
// FoldUnsafe reports that the response contains a known-partial thread.
// Truncation alone does not imply this: completeCommentThreads makes the
// default newest window safe to fold by restoring or dropping whole threads.
FoldUnsafe bool
}
// rootStat is the per-thread orientation metadata attached to each root comment
// on the roots_only path. See CommentResponse.ReplyCount / LastActivityAt.
type rootStat struct {
ReplyCount int
LastActivityAt pgtype.Timestamptz
}
var (
errCommentThreadNotFound = &commentFetchError{"thread anchor not found"}
errCommentThreadBadID = &commentFetchError{"invalid thread anchor id"}
)
type commentFetchError struct{ msg string }
func (e *commentFetchError) Error() string { return e.msg }
func (h *Handler) fetchCommentsForList(ctx context.Context, args fetchCommentsArgs) (fetchCommentsResult, error) {
issue := args.Issue
// Thread-scoped read. Server resolves the anchor → root via recursive
// CTE, so we don't have to assume two-layer flat threads here.
if args.ThreadAnchor != "" {
anchor, err := util.ParseUUID(args.ThreadAnchor)
if err != nil {
return fetchCommentsResult{}, errCommentThreadBadID
}
// Tailed path: paged query that returns root + the @reply_limit
// most recent replies (per (created_at, id)). The thread root is
// always returned, so a reader can land on a long thread without
// dragging hundreds of replies into context. The reply-internal
// cursor (--before / --before-id under --thread + --tail) scrolls
// to older replies inside the same thread.
if args.ThreadTailSet {
// Probe for has-more by asking the SQL for one extra reply
// beyond what the caller wants. If we get back >tail replies
// there is at least one older reply still on disk; if we get
// back ≤tail the page is the tail of the thread and there is
// nothing older to scroll to (so we must NOT emit a cursor —
// otherwise the next page is wasted round-trip that returns
// just the root). This is the exact-boundary fix called out
// in the MUL-2421 review.
rows, err := h.Queries.ListThreadCommentsForIssuePaged(ctx, db.ListThreadCommentsForIssuePagedParams{
AnchorID: anchor,
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
HasCursor: args.HasCursor,
BeforeAt: args.BeforeAt,
BeforeID: args.BeforeID,
ReplyLimit: int32(args.ThreadTail) + 1,
})
if err != nil {
return fetchCommentsResult{}, err
}
if len(rows) == 0 {
return fetchCommentsResult{}, errCommentThreadNotFound
}
// Split the result into root + replies (ASC order preserved).
// Root is identified by parent_id IS NULL and is always
// present in the SQL output; we keep it out of the cursor /
// tail-trim logic so the user always sees thread context.
var rootComment *db.Comment
replies := make([]db.Comment, 0, len(rows))
for _, r := range rows {
c := db.Comment{
ID: r.ID,
IssueID: r.IssueID,
AuthorType: r.AuthorType,
AuthorID: r.AuthorID,
Content: r.Content,
Type: r.Type,
CreatedAt: r.CreatedAt,
UpdatedAt: r.UpdatedAt,
ParentID: r.ParentID,
WorkspaceID: r.WorkspaceID,
ResolvedAt: r.ResolvedAt,
ResolvedByType: r.ResolvedByType,
ResolvedByID: r.ResolvedByID,
SourceTaskID: r.SourceTaskID,
QuickActionID: r.QuickActionID,
Revision: r.Revision,
DeletedAt: r.DeletedAt,
}
if !r.ParentID.Valid {
root := c
rootComment = &root
continue
}
replies = append(replies, c)
}
// Trim the probe overflow back to the caller's tail. The SQL
// emits ASC, so the extra row is the oldest reply — dropping
// it from the head is what aligns "newest N" with the user's
// request.
hasMore := len(replies) > args.ThreadTail
if hasMore {
replies = replies[1:]
}
out := make([]db.Comment, 0, len(replies)+1)
if rootComment != nil {
out = append(out, *rootComment)
}
for _, r := range replies {
// since drops stale rows AFTER the tail / cursor cut.
// The root is exempt (already appended above): a reader
// who set --since to skip already-seen replies still
// needs the root context if the page only contained
// the root.
if args.Since.Valid && !r.CreatedAt.Time.After(args.Since.Time) {
continue
}
out = append(out, r)
}
// Emit a reply cursor only when we proved an older reply
// exists (hasMore). On an exact-boundary page (replyCount
// == tail with no overflow) hasMore is false and the cursor
// stays empty.
//
// Additionally suppress the cursor when `since` is set and
// the oldest retained reply on this page is already <= since.
// The next page walks replies strictly older than that one,
// so every older reply has created_at strictly less — if the
// cursor target itself can't satisfy `> since`, no older
// reply can either, and continuing to paginate would only
// return root-only pages until the agent walks the entire
// pre-`since` history. This mirrors the head-thread guard on
// the recent + since path. Flagged by Elon's second review on
// MUL-2421.
res := fetchCommentsResult{Comments: out}
emitCursor := hasMore && len(replies) > 0
if emitCursor && args.Since.Valid && !replies[0].CreatedAt.Time.After(args.Since.Time) {
emitCursor = false
}
if emitCursor {
oldest := replies[0]
res.NextBefore = oldest.CreatedAt.Time.UTC().Format(time.RFC3339Nano)
res.NextBeforeID = uuidToString(oldest.ID)
}
return res, nil
}
// Untailed reads use the same newest-reply query as the paged path.
// Probe with commentHardCap replies: because the root is unconditional,
// 2000 replies proves the total thread exceeds the 2000-row response cap.
// We then drop the oldest reply and retain root + newest 1999 replies.
rows, err := h.Queries.ListThreadCommentsForIssuePaged(ctx, db.ListThreadCommentsForIssuePagedParams{
AnchorID: anchor,
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
ReplyLimit: int32(commentHardCap),
})
if err != nil {
return fetchCommentsResult{}, err
}
if len(rows) == 0 {
return fetchCommentsResult{}, errCommentThreadNotFound
}
var rootComment *db.Comment
replies := make([]db.Comment, 0, len(rows))
for _, r := range rows {
c := db.Comment{
ID: r.ID,
IssueID: r.IssueID,
AuthorType: r.AuthorType,
AuthorID: r.AuthorID,
Content: r.Content,
Type: r.Type,
CreatedAt: r.CreatedAt,
UpdatedAt: r.UpdatedAt,
ParentID: r.ParentID,
WorkspaceID: r.WorkspaceID,
ResolvedAt: r.ResolvedAt,
ResolvedByType: r.ResolvedByType,
ResolvedByID: r.ResolvedByID,
SourceTaskID: r.SourceTaskID,
QuickActionID: r.QuickActionID,
Revision: r.Revision,
DeletedAt: r.DeletedAt,
}
if !r.ParentID.Valid {
root := c
rootComment = &root
continue
}
replies = append(replies, c)
}
truncated := len(replies) >= commentHardCap
if truncated {
replies = replies[1:]
}
out := make([]db.Comment, 0, len(replies)+1)
if rootComment != nil && (!args.Since.Valid || rootComment.CreatedAt.Time.After(args.Since.Time)) {
out = append(out, *rootComment)
}
for _, reply := range replies {
if args.Since.Valid && !reply.CreatedAt.Time.After(args.Since.Time) {
continue
}
out = append(out, reply)
}
return fetchCommentsResult{
Comments: out,
CommentsTruncated: truncated,
FoldUnsafe: truncated,
}, nil
}
// Thread-grouped recent read: N most recently active threads.
if args.RecentN > 0 {
rows, err := h.Queries.ListRecentThreadCommentsForIssue(ctx, db.ListRecentThreadCommentsForIssueParams{
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
HasCursor: args.HasCursor,
BeforeAt: args.BeforeAt,
BeforeID: args.BeforeID,
ThreadLimit: int32(args.RecentN),
})
if err != nil {
return fetchCommentsResult{}, err
}
// The SQL already orders rows by (last_activity_at ASC, root_id ASC,
// created_at ASC, id ASC), so the OLDEST-active thread sits at the
// head and the FRESHEST thread at the tail. Walk the rows once to:
// 1. Strip the thread-metadata columns down to db.Comment for the
// caller (uniform shape across paths).
// 2. Count distinct threads in the page so we know whether a "next
// older page" is likely to exist.
// 3. Capture the head thread's (last_activity_at, root_id) — that
// is the cursor for the next page (next page = threads strictly
// less recent than this one).
comments := make([]db.Comment, 0, len(rows))
var headRoot pgtype.UUID
var headLast pgtype.Timestamptz
seenRoot := map[string]struct{}{}
for _, r := range rows {
if !headRoot.Valid {
headRoot = r.ThreadRootID
headLast = r.ThreadLastActivityAt
}
seenRoot[uuidToString(r.ThreadRootID)] = struct{}{}
// Since filter on the recent path: drop comments older than
// `since`. Done in-memory so we keep the thread-grouped
// semantics from the query (don't pre-filter rows before the
// MAX(created_at) ranking — that would silently downgrade a
// thread whose most recent activity falls inside the window).
if args.Since.Valid && !r.CreatedAt.Time.After(args.Since.Time) {
continue
}
comments = append(comments, db.Comment{
ID: r.ID,
IssueID: r.IssueID,
AuthorType: r.AuthorType,
AuthorID: r.AuthorID,
Content: r.Content,
Type: r.Type,
CreatedAt: r.CreatedAt,
UpdatedAt: r.UpdatedAt,
ParentID: r.ParentID,
WorkspaceID: r.WorkspaceID,
ResolvedAt: r.ResolvedAt,
ResolvedByType: r.ResolvedByType,
ResolvedByID: r.ResolvedByID,
SourceTaskID: r.SourceTaskID,
QuickActionID: r.QuickActionID,
Revision: r.Revision,
DeletedAt: r.DeletedAt,
})
}
// Only emit a cursor when the page is full. Fewer threads than
// requested ⇒ the SELECT exhausted matching threads, so there is
// no older page to scroll to.
//
// Additionally suppress the cursor when `since` is set and the head
// thread's last_activity_at is already <= since. The pagination
// walks threads in strictly decreasing last_activity_at, so every
// older page has last_activity_at strictly less than the head's —
// if the head itself can't satisfy `> since`, no older thread can
// either. Predicating on the head (not on whether `comments` is
// empty) also catches the mixed case where this page keeps rows
// from fresher threads but the head thread is already past `since`.
// Flagged by Elon in #2787's second review (MUL-2340 nit).
out := fetchCommentsResult{Comments: comments}
emitCursor := len(seenRoot) >= args.RecentN && headRoot.Valid && headLast.Valid
if emitCursor && args.Since.Valid && !headLast.Time.After(args.Since.Time) {
emitCursor = false
}
if emitCursor {
out.NextBefore = headLast.Time.UTC().Format(time.RFC3339Nano)
out.NextBeforeID = uuidToString(headRoot)
}
return out, nil
}
if args.RootsOnly {
// Root-only read for issue-level orientation. This intentionally
// stays separate from thread/recent modes: callers get the global
// top-level discussion first, then fetch a specific thread only when
// they need reply context. Each root carries reply_count +
// last_activity_at so the reader can triage which thread to drill into.
stats := map[string]rootStat{}
if args.Since.Valid {
rows, err := h.Queries.ListRootCommentsSinceForIssue(ctx, db.ListRootCommentsSinceForIssueParams{
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
Since: args.Since,
RowLimit: commentProbeLimit(),
})
if err != nil {
return fetchCommentsResult{}, err
}
truncated := len(rows) > commentHardCap
if truncated {
// Since is an incremental stream: keep the first page after the
// cursor so callers can advance without skipping a gap.
rows = rows[:commentHardCap]
}
comments := make([]db.Comment, len(rows))
for i, r := range rows {
comments[i] = db.Comment{
ID: r.ID, IssueID: r.IssueID, AuthorType: r.AuthorType, AuthorID: r.AuthorID,
Content: r.Content, Type: r.Type, CreatedAt: r.CreatedAt, UpdatedAt: r.UpdatedAt,
ParentID: r.ParentID, WorkspaceID: r.WorkspaceID, ResolvedAt: r.ResolvedAt,
ResolvedByType: r.ResolvedByType, ResolvedByID: r.ResolvedByID,
SourceTaskID: r.SourceTaskID, QuickActionID: r.QuickActionID, Revision: r.Revision,
DeletedAt: r.DeletedAt,
}
stats[uuidToString(r.ID)] = rootStat{ReplyCount: int(r.ReplyCount), LastActivityAt: r.LastActivityAt}
}
return fetchCommentsResult{
Comments: comments, RootStats: stats, CommentsTruncated: truncated,
}, nil
}
rows, err := h.Queries.ListRootCommentsForIssue(ctx, db.ListRootCommentsForIssueParams{
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
RowLimit: commentProbeLimit(),
})
if err != nil {
return fetchCommentsResult{}, err
}
truncated := len(rows) > commentHardCap
if truncated {
// The SQL returns selected roots chronologically after taking the
// newest probe window, so the overflow row is the oldest.
rows = rows[1:]
}
comments := make([]db.Comment, len(rows))
for i, r := range rows {
comments[i] = db.Comment{
ID: r.ID, IssueID: r.IssueID, AuthorType: r.AuthorType, AuthorID: r.AuthorID,
Content: r.Content, Type: r.Type, CreatedAt: r.CreatedAt, UpdatedAt: r.UpdatedAt,
ParentID: r.ParentID, WorkspaceID: r.WorkspaceID, ResolvedAt: r.ResolvedAt,
ResolvedByType: r.ResolvedByType, ResolvedByID: r.ResolvedByID,
SourceTaskID: r.SourceTaskID, QuickActionID: r.QuickActionID, Revision: r.Revision,
DeletedAt: r.DeletedAt,
}
stats[uuidToString(r.ID)] = rootStat{ReplyCount: int(r.ReplyCount), LastActivityAt: r.LastActivityAt}
}
return fetchCommentsResult{
Comments: comments, RootStats: stats, CommentsTruncated: truncated,
}, nil
}
// Since reads keep their chronological page shape. A probe makes the cap
// visible without skipping the next page of the incremental stream.
if args.Since.Valid {
comments, err := h.Queries.ListCommentsSinceForIssue(ctx, db.ListCommentsSinceForIssueParams{
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
CreatedAt: args.Since,
Limit: commentProbeLimit(),
})
if err != nil {
return fetchCommentsResult{}, err
}
truncated := len(comments) > commentHardCap
if truncated {
comments = comments[:commentHardCap]
}
return fetchCommentsResult{Comments: comments, CommentsTruncated: truncated}, nil
}
comments, err := h.Queries.ListCommentsForIssue(ctx, db.ListCommentsForIssueParams{
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
Limit: commentProbeLimit(),
})
if err != nil {
return fetchCommentsResult{}, err
}
// Probe read: an issue holding exactly commentHardCap comments is complete,
// and calling that truncated would needlessly suppress the thread fold below.
truncated := len(comments) > commentHardCap
if truncated {
comments = comments[len(comments)-commentHardCap:]
// The cap keeps the NEWEST comments (MUL-5492), so a thread can be cut
// in half: an old root or sibling outside the window, a fresh reply
// inside it. Complete every affected thread within the shared context
// budget. If a thread cannot be completed, it is dropped as one unit.
//
comments, err = h.completeCommentThreads(ctx, issue.ID, issue.WorkspaceID, comments)
if err != nil {
return fetchCommentsResult{}, err
}
}
return fetchCommentsResult{Comments: comments, CommentsTruncated: truncated}, nil
}
// completeCommentThreads restores complete renderable threads around a newest-N
// comment window.
//
// Why this is needed at all: capping with the OLDEST n comments could never
// orphan a reply, because a reply is always newer than its parent and so a
// prefix of the timeline is closed under "parent of". A newest-n window is a
// suffix and has no such property — an old thread root falls outside while a
// fresh reply to it stays inside (MUL-5492).
//
// Parent closure comes first because an orphaned reply is not merely mis-nested,
// it is invisible.
// The timeline builds its top level from "activities + comments with no
// parent_id" and renders replies by looking them up under their parent, so an
// orphan sits in the map with no card to render it — the exact shape of MUL-1847
// (1 root + 29 replies, root dropped, all 29 vanished from the UI while the API
// returned them).
//
// Parent closure alone is insufficient for the human timeline. Web and Desktop
// derive resolution bars and folded counts from the comments they receive, so a
// backfilled root plus only its newest replies would make a partial thread look
// complete. After reaching roots, this function walks DOWN from every restored
// root and adds all missing siblings/descendants. If the shared budget or depth
// bound cannot prove a whole affected thread complete, that thread is removed
// as one unit; old installed clients can therefore never fold partial data.
//
// Both walks share commentThreadContextBudget, so the returned set remains
// provably bounded by commentHardCap + commentThreadContextBudget. Descendants
// are fetched one level at a time for every affected root in one query, avoiding
// an N+1 query per thread. Each query carries issue + workspace predicates and a
// probe limit.
//
// When a budget is exhausted, a parent row is missing, or a parent belongs to
// another issue or workspace, keepRootConnected drops the affected replies. A
// descendant overflow drops every affected thread rather than guessing which
// root is complete. Returning fewer new replies is a conservative degradation
// that the caller already signals as truncated; leaking another tenant's
// comments, returning an unbounded response, or emitting partial threads are all
// worse.
func (h *Handler) completeCommentThreads(ctx context.Context, issueID, workspaceID pgtype.UUID, window []db.Comment) ([]db.Comment, error) {
if len(window) == 0 {
return window, nil
}
initialIDs := make(map[string]struct{}, len(window))
byID := make(map[string]db.Comment, len(window)+commentThreadContextBudget/8)
through := window[0]
for _, c := range window {
id := uuidToString(c.ID)
initialIDs[id] = struct{}{}
byID[id] = c
if c.CreatedAt.Time.After(through.CreatedAt.Time) ||
(c.CreatedAt.Time.Equal(through.CreatedAt.Time) &&
bytes.Compare(c.ID.Bytes[:], through.ID.Bytes[:]) > 0) {
through = c
}
}
added := 0
for depth := 0; depth < commentThreadMaxDepth; depth++ {
// Distinct parents referenced by the current set but not in it. Dedup is
// what keeps many replies sharing one ancestor from fetching it twice.
var missing []pgtype.UUID
seen := make(map[string]struct{})
for _, c := range byID {
if !c.ParentID.Valid {
continue
}
pid := uuidToString(c.ParentID)
if _, have := byID[pid]; have {
continue
}
if _, dup := seen[pid]; dup {
continue
}
seen[pid] = struct{}{}
missing = append(missing, c.ParentID)
}
if len(missing) == 0 {
break
}
if added+len(missing) > commentThreadContextBudget {
break
}
// Tenant-scoped: a parent in another issue or workspace simply does not
// come back, and keepRootConnected then prunes the reply that pointed at
// it. The filter is on every level, not just the first.
rows, err := h.Queries.ListCommentsByIDsForIssue(ctx, db.ListCommentsByIDsForIssueParams{
Ids: missing,
IssueID: issueID,
WorkspaceID: workspaceID,
})
if err != nil {
return nil, err
}
if len(rows) == 0 {
break
}
for _, r := range rows {
id := uuidToString(r.ID)
if _, have := byID[id]; have {
continue
}
byID[id] = r
added++
}
}
// Remove every chain that did not reach a same-tenant root within the
// ancestor budget/depth. Rebuild the map so the descendant phase cannot
// accidentally revive a disconnected chain.
connected := keepRootConnected(byID)
byID = make(map[string]db.Comment, len(connected)+commentThreadContextBudget-added)
for _, c := range connected {
byID[uuidToString(c.ID)] = c
}
// A restored root identifies a thread that the newest window cut. Threads
// whose roots were already in the suffix are complete under the normal
// parent-before-reply timestamp contract and need no extra query.
affectedRoots := make(map[string]struct{})
for id, c := range byID {
if c.ParentID.Valid {
continue
}
if _, wasInWindow := initialIDs[id]; !wasInWindow {
affectedRoots[id] = struct{}{}
}
}
if len(affectedRoots) > 0 {
complete, err := h.addMissingCommentDescendants(
ctx,
issueID,
workspaceID,
byID,
affectedRoots,
through,
&added,
)
if err != nil {
return nil, err
}
if !complete {
dropCommentThreads(byID, affectedRoots)
}
}
out := make([]db.Comment, 0, len(byID))
for _, c := range byID {
out = append(out, c)
}
sortCommentsChronologically(out)
return out, nil
}
// addMissingCommentDescendants walks every affected thread breadth-first. It
// returns false when the remaining context budget or depth bound cannot prove
// all affected threads complete; callers then drop those threads as one unit.
func (h *Handler) addMissingCommentDescendants(
ctx context.Context,
issueID, workspaceID pgtype.UUID,
byID map[string]db.Comment,
affectedRoots map[string]struct{},
through db.Comment,
added *int,
) (bool, error) {
frontier := make(map[string]string, len(affectedRoots)) // comment id -> root id
for rootID := range affectedRoots {
frontier[rootID] = rootID
}
// Every returned child is unique in a tree walk. The scan budget includes
// rows already present in the window plus the remaining add budget, so all
// descendant queries together inspect at most the response bound (+1 probe).
scanBudget := len(byID) + commentThreadContextBudget - *added
for depth := 0; depth <= commentThreadMaxDepth; depth++ {
parentKeys := make([]string, 0, len(frontier))
for id := range frontier {
parentKeys = append(parentKeys, id)
}
sort.Strings(parentKeys)
parentIDs := make([]pgtype.UUID, len(parentKeys))
for i, id := range parentKeys {
parentIDs[i] = byID[id].ID
}
rows, err := h.Queries.ListChildCommentsForParents(ctx, db.ListChildCommentsForParentsParams{
ParentIds: parentIDs,
IssueID: issueID,
WorkspaceID: workspaceID,
ThroughAt: through.CreatedAt,
ThroughID: through.ID,
RowLimit: int32(scanBudget + 1),
})
if err != nil {
return false, err
}
if len(rows) == 0 {
return true, nil
}
if len(rows) > scanBudget || depth == commentThreadMaxDepth {
return false, nil
}
scanBudget -= len(rows)
next := make(map[string]string, len(rows))
for _, row := range rows {
parentID := uuidToString(row.ParentID)
rootID, ok := frontier[parentID]
if !ok {
// The query is constrained to frontier parent ids, so this is
// defensive fail-closed handling for malformed query output.
return false, nil
}
id := uuidToString(row.ID)
next[id] = rootID
if _, exists := byID[id]; exists {
continue
}
if *added >= commentThreadContextBudget {
return false, nil
}
byID[id] = row
*added = *added + 1
}
frontier = next
}
return false, nil
}
// dropCommentThreads removes every comment whose root is in roots. It is used
// only after a bounded descendant walk could not prove the affected threads
// complete.
func dropCommentThreads(byID map[string]db.Comment, roots map[string]struct{}) {
for id := range byID {
rootID, ok := commentRootID(id, byID)
if !ok {
delete(byID, id)
continue
}
if _, drop := roots[rootID]; drop {
delete(byID, id)
}
}
}
func commentRootID(id string, byID map[string]db.Comment) (string, bool) {
seen := make(map[string]struct{})
for current := id; ; {
if _, loop := seen[current]; loop {
return "", false
}
seen[current] = struct{}{}
comment, ok := byID[current]
if !ok {
return "", false
}
if !comment.ParentID.Valid {
return current, true
}
current = uuidToString(comment.ParentID)
}
}
func sortCommentsChronologically(comments []db.Comment) {
sort.Slice(comments, func(i, j int) bool {
if !comments[i].CreatedAt.Time.Equal(comments[j].CreatedAt.Time) {
return comments[i].CreatedAt.Time.Before(comments[j].CreatedAt.Time)
}
return bytes.Compare(comments[i].ID.Bytes[:], comments[j].ID.Bytes[:]) < 0
})
}
// keepRootConnected returns only the comments whose parent chain terminates at a
// root (parent_id IS NULL) inside the set.
//
// A comment whose chain leaves the set, loops, or dead-ends is unrenderable: the
// UI needs a top-level card to hang the thread from. Dropping such a comment also
// drops its descendants, because their chains run through it and therefore fail
// the same test — no separate descendant pass is needed.
//
// The walk is bounded by the visited set, so a cycle in stored data terminates
// instead of hanging. Real cycles are impossible via the write path, but a graph
// walk over stored data should never assume that.
func keepRootConnected(byID map[string]db.Comment) []db.Comment {
connected := make(map[string]bool, len(byID))
for id := range byID {
if _, decided := connected[id]; decided {
continue
}
var path []string
visited := make(map[string]struct{})
verdict := false
for cur := id; ; {
if v, decided := connected[cur]; decided {
verdict = v
break
}
if _, loop := visited[cur]; loop {
break // cycle: nothing on this path reaches a root
}
c, present := byID[cur]
if !present {
break // chain leaves the set
}
visited[cur] = struct{}{}
path = append(path, cur)
if !c.ParentID.Valid {
verdict = true // a real root, inside the set
break
}
cur = uuidToString(c.ParentID)
}
for _, p := range path {
connected[p] = verdict
}
}
out := make([]db.Comment, 0, len(byID))
for id, c := range byID {
if connected[id] {
out = append(out, c)
}
}
return out
}
type CreateCommentRequest struct {
Content string `json:"content"`
Type string `json:"type"`
ParentID *string `json:"parent_id"`
AttachmentIDs []string `json:"attachment_ids"`
SuppressAgentIDs []string `json:"suppress_agent_ids"`
// SteerTaskIDs are the running turns the author chose to add this comment
// to, instead of starting a follow-up run for their agents. A turn that
// has ended, or cannot take additional input, is never swapped for another
// one: its agent keeps the normal trigger.
SteerTaskIDs []string `json:"steer_task_ids"`
// Steer adds the comment to the current running turn of every recipient
// that has no turn in SteerTaskIDs, with the same fallback.
Steer bool `json:"steer"`
}
type CommentTriggerPreviewRequest struct {
Content string `json:"content"`
ParentID *string `json:"parent_id"`
EditingCommentID *string `json:"editing_comment_id"`
}
type CommentTriggerPreviewResponse struct {
Agents []CommentTriggerAgentResponse `json:"agents"`
// Blocked lists explicit @agent / @squad mentions that will NOT trigger if
// this comment is posted as-is (MUL-4525 §2). Additive: old clients ignore
// it. It lets the composer warn before sending instead of the user only
// discovering the silent no-op afterwards.
Blocked []CommentTriggerOutcome `json:"blocked,omitempty"`
}
type CommentTriggerAgentResponse struct {
ID string `json:"id"`
Name string `json:"name"`
AvatarURL *string `json:"avatar_url,omitempty"`
Source string `json:"source"`
Reason string `json:"reason"`
}
type commentAgentTriggerSource string
const (
commentTriggerSourceIssueAssignee commentAgentTriggerSource = "issue_assignee"
commentTriggerSourceMentionAgent commentAgentTriggerSource = "mention_agent"
commentTriggerSourceMentionSquadLeader commentAgentTriggerSource = "mention_squad_leader"
commentTriggerSourceThreadParent commentAgentTriggerSource = "thread_parent"
commentTriggerSourceConversation commentAgentTriggerSource = "conversation_continuation"
)
type commentAgentTrigger struct {
Agent db.Agent
Source commentAgentTriggerSource
Squad *db.Squad
AlreadyPending bool
// NonLeaderAgentReply marks an agent comment routed back to a squad leader
// from worker context. The worker need not be a squad member. Completion may
// replay it only if creation recorded it as a planned input.
NonLeaderAgentReply bool
}
type commentTriggerComputeOptions struct {
ThreadCommentID pgtype.UUID
ExcludeTriggerCommentID pgtype.UUID
// AuthoringTaskID is the server-trusted task that wrote an agent comment.
// It lets a worker result continue the exact guest squad-leader delegation
// without guessing from squad membership or the issue assignee.
AuthoringTaskID pgtype.UUID
// OriginatorUserID is the top-of-chain human user id for this trigger
// (MUL-3963). Only consulted for AGENT actors — canInvokeAgent judges A2A
// by the originator, not the immediate agent principal. Members are their
// own originator so this may be empty for member-authored triggers.
OriginatorUserID string
}
func commentAgentTriggerReason(trigger commentAgentTrigger) string {
switch trigger.Source {
case commentTriggerSourceIssueAssignee:
return "Current issue assignment will trigger this agent."
case commentTriggerSourceMentionAgent:
return "This agent was mentioned in the comment."
case commentTriggerSourceMentionSquadLeader:
return "A mentioned squad will trigger its leader."
case commentTriggerSourceThreadParent:
return "This reply will trigger the parent comment's author."
case commentTriggerSourceConversation:
return "This follow-up will continue the recent agent conversation."
default:
return "This comment will trigger this agent."
}
}
func (h *Handler) commentAgentTriggerToResponse(trigger commentAgentTrigger) CommentTriggerAgentResponse {
return CommentTriggerAgentResponse{
ID: uuidToString(trigger.Agent.ID),
Name: trigger.Agent.Name,
AvatarURL: h.resolveAvatarURLPtr(textToPtr(trigger.Agent.AvatarUrl)),
Source: string(trigger.Source),
Reason: commentAgentTriggerReason(trigger),
}
}
func (h *Handler) PreviewCommentTriggers(w http.ResponseWriter, r *http.Request) {
issueID := chi.URLParam(r, "id")
issue, ok := h.loadIssueForUser(w, r, issueID)
if !ok {
return
}
userID, ok := requireUserID(w, r)
if !ok {
return
}
var req CommentTriggerPreviewRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
var editingComment *db.Comment
var opts commentTriggerComputeOptions
if req.EditingCommentID != nil {
editingID, ok := parseUUIDOrBadRequest(w, *req.EditingCommentID, "editing_comment_id")
if !ok {
return
}
comment, err := h.Queries.GetCommentInWorkspace(r.Context(), db.GetCommentInWorkspaceParams{
ID: editingID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil || uuidToString(comment.IssueID) != uuidToString(issue.ID) {
writeError(w, http.StatusBadRequest, "invalid editing comment")
return
}
editingComment = &comment
opts.ExcludeTriggerCommentID = editingID
}
var parentID pgtype.UUID
if req.ParentID != nil {
parentID, ok = parseUUIDOrBadRequest(w, *req.ParentID, "parent_id")
if !ok {
return
}
if editingComment != nil && uuidToString(parentID) != uuidToString(editingComment.ParentID) {
writeError(w, http.StatusBadRequest, "parent_id does not match editing comment")
return
}
} else if editingComment != nil && editingComment.ParentID.Valid {
parentID = editingComment.ParentID
}
var parentComment *db.Comment
if parentID.Valid {
parent, err := h.Queries.GetComment(r.Context(), parentID)
if err != nil || uuidToString(parent.IssueID) != uuidToString(issue.ID) {
writeError(w, http.StatusBadRequest, "invalid parent comment")
return
}
parentComment = &parent
}
// Normalize with the SAME entry point CreateComment/UpdateComment apply
// before persisting (sanitizeNullBytes), so the preview computes triggers on
// the exact content that will be stored and enqueued on submit. Otherwise a
// mention hidden behind a byte the DB strips (e.g. a NUL inside
// mention://agent/<uuid>) reads as inert in preview but enqueues the agent on
// submit — a preview/side-effect divergence (GH #5388 review).
content := sanitizeNullBytes(req.Content)
if content == "" {
writeJSON(w, http.StatusOK, CommentTriggerPreviewResponse{Agents: []CommentTriggerAgentResponse{}})
return
}
actorType, actorID := h.resolveActor(r, userID, uuidToString(issue.WorkspaceID))
opts.OriginatorUserID = h.invokeOriginatorFromRequest(r, actorType, actorID)
if actorType == "agent" {
opts.AuthoringTaskID = h.commentSourceTaskID(r)
}
triggers, targets := h.computeCommentAgentTriggers(r.Context(), issue, content, parentComment, actorType, actorID, opts)
resp := CommentTriggerPreviewResponse{
Agents: make([]CommentTriggerAgentResponse, 0, len(triggers)),
Blocked: commentBlockedTargetOutcomes(targets),
}
for _, trigger := range triggers {
resp.Agents = append(resp.Agents, h.commentAgentTriggerToResponse(trigger))
}
writeJSON(w, http.StatusOK, resp)
}
// taskCoversReplyParent reports whether parentID is a comment this task is
// authorized to reply under. A comment-triggered task may reply to its trigger
// comment OR to any earlier comment that was folded into the same run while it
// was still queued (coalesced_comment_ids). A coalesced run answers each root
// thread it covered inside that thread, so its replies legitimately target
// those threads' comments — not just the trigger (MUL-4348 per-thread fan-out).
//
// Every other parent on the task's own issue is still rejected: this is the
// defense against resumed-session --parent drift and cross-thread misplacement.
// The allow-list is exactly the set the run was given to answer, so it cannot
// reach arbitrary comments.
func taskCoversReplyParent(task db.AgentTaskQueue, parentID pgtype.UUID) bool {
if !parentID.Valid {
return false
}
target := uuidToString(parentID)
if task.TriggerCommentID.Valid && uuidToString(task.TriggerCommentID) == target {
return true
}
for _, id := range task.CoalescedCommentIds {
if id.Valid && uuidToString(id) == target {
return true
}
}
return false
}
func (h *Handler) CreateComment(w http.ResponseWriter, r *http.Request) {
r = h.withWakeupActor(r)
issueID := chi.URLParam(r, "id")
issue, ok := h.loadIssueForUser(w, r, issueID)
if !ok {
return
}
userID, ok := requireUserID(w, r)
if !ok {
return
}
var req CreateCommentRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
// Strip bytes PostgreSQL's TEXT column rejects before the empty check. The
// case reachable over the JSON API is an embedded NUL (0x00, SQLSTATE
// 22021) that survives a JSON round trip; sanitizeNullBytes also drops
// invalid UTF-8 as defense-in-depth. A stray such byte in agent-written
// content (notably via --content-file) otherwise fails the INSERT with an
// opaque 500 the CLI renders as "server unavailable" and retries forever —
// the plausible cause of GH #5388. Mirrors the skill-import sanitization;
// normalizing first means all-NUL content is correctly treated as empty.
req.Content = sanitizeNullBytes(req.Content)
if req.Content == "" {
writeError(w, http.StatusBadRequest, "content is required")
return
}
if req.Type == "" {
req.Type = "comment"
}
// Validate rather than letting the DB CHECK reject it: an unknown type
// previously surfaced as a 500 on a constraint violation, which reads as a
// server fault for what is plainly bad input. This is also the explicit
// refusal of a client trying to author a machine-written kind (MUL-5465
// review): a member must not be able to post a comment that renders as
// something the system generated.
if !isClientAuthorableCommentType(req.Type) {
writeError(w, http.StatusBadRequest, "invalid comment type")
return
}
var parentID pgtype.UUID
var parentComment *db.Comment
if req.ParentID != nil {
var parsed pgtype.UUID
parsed, ok = parseUUIDOrBadRequest(w, *req.ParentID, "parent_id")
if !ok {
return
}
parentID = parsed
parent, err := h.Queries.GetComment(r.Context(), parentID)
if err != nil || uuidToString(parent.IssueID) != uuidToString(issue.ID) {
writeError(w, http.StatusBadRequest, "invalid parent comment")
return
}
parentComment = &parent
}
attachmentIDs, ok := parseUUIDSliceOrBadRequest(w, req.AttachmentIDs, "attachment_ids")
if !ok {
return
}
suppressAgentIDs, ok := parseUUIDSliceOrBadRequest(w, req.SuppressAgentIDs, "suppress_agent_ids")
if !ok {
return
}
steerTaskIDs, ok := parseUUIDSliceOrBadRequest(w, req.SteerTaskIDs, "steer_task_ids")
if !ok {
return
}
// A running turn receives text only, so a comment that carries files is
// always a normal trigger.
steer := commentSteerRequest{TaskIDs: steerTaskIDs, All: req.Steer}
if len(attachmentIDs) > 0 {
steer = commentSteerRequest{}
}
// Determine author identity: agent (via X-Agent-ID header) or member.
authorType, authorID := h.resolveActor(r, userID, uuidToString(issue.WorkspaceID))
// sourceTaskID captures the agent's currently-executing task when it posts
// via the CLI (X-Task-ID header). Stamping it on the comment row keeps the
// originator inheritance chain (resolveOriginatorFromTriggerComment →
// comment.source_task_id → parent task's originator_user_id) intact across
// an agent→agent hop. Without this stamp, a private squad leader's
// worker-agent whose completion wakes the leader via
// routeAssignedSquadLeaderFallback can't pass canInvokeAgent — the worker's
// task originator is unattributed, effectiveUser resolves to "", and the
// private-agent gate denies the wake (MUL-4015).
//
// The header is NOT client-chosen: the auth middleware deletes any
// client-supplied X-Agent-ID / X-Task-ID and re-stamps both from the mat_
// token's own row (MUL-3428). So the stamp always records the authoring
// agent's own run.
//
// It is deliberately NOT scoped to that run's own issue (MUL-6490 / GH
// #7328). A run that legitimately comments on ANOTHER issue — the ordinary
// "agent creates issue Y, then coordinates there" flow — used to persist a
// NULL lineage, so the run that comment woke resolved as unattributed and
// every @mention / assign / sub-issue inside it hit invocation_not_allowed,
// even though the SAME delegation had just succeeded on the originating
// issue. Propagating the chain is monotonic: it can only carry the human
// this run ALREADY acts for, and every hop still re-runs canInvokeAgent
// against that same human, so the admissible set stays a subset of what that
// human could trigger directly. What must never happen is SUBSTITUTING a
// human — an owner fallback, or adopting the target issue's own originator —
// because that borrows a different person's authority; those stay rejected.
//
// Consumers that genuinely need "the authoring run is working on THIS issue"
// keep that check next to their own rule instead: the reply-parent /
// no_action guards below, and autopilotDelegationAuthority's lineage
// verification (MUL-4857).
var sourceTaskID pgtype.UUID
if authorType == "agent" {
if task, ok := h.taskFromRequestHeader(r); ok {
// Defense against resumed-session drift: when an agent posts from
// inside a comment-triggered task AND the comment is being posted on
// that same issue, the parent_id must exactly match the task's
// trigger comment. Resumed Claude sessions otherwise carry forward a
// previous turn's --parent UUID and silently misplace the reply.
//
// The task.IssueID scope is important: the CLI stamps X-Task-ID on
// every request, so an agent legitimately commenting on a different
// issue must not be blocked by its current task's trigger.
// Assignment-triggered tasks (no TriggerCommentID) are also
// unaffected.
if task.IssueID.Valid && uuidToString(task.IssueID) == uuidToString(issue.ID) {
if task.TriggerCommentID.Valid {
if !taskCoversReplyParent(task, parentID) {
// Keep this error actionable for agents (MUL-4417 / GH #5266).
// The two rejections need different copy. A resumed
// session carrying a previous turn's --parent forward
// (GH #6264) did NOT ask for a top-level comment, and
// telling it that it did sends it looking for a
// new-thread opt-in instead of simply correcting the
// parent it already passed.
fix := "set parent_id (--parent) to " + uuidToString(task.TriggerCommentID) + " or a coalesced comment id"
msg := "comment-triggered tasks cannot create top-level comments; " + fix
if parentID.Valid {
msg = "parent_id " + uuidToString(parentID) + " is not a comment this task may reply under; " + fix
}
writeError(w, http.StatusConflict, msg)
return
}
}
noAction, checkErr := service.HasSquadLeaderNoActionEvaluationForTask(r.Context(), h.Queries, task)
if checkErr != nil {
slog.Warn("checking squad leader no_action evaluation failed", append(logger.RequestAttrs(r),
"error", checkErr,
"task_id", uuidToString(task.ID),
"issue_id", issueID,
)...)
} else if noAction {
writeError(w, http.StatusConflict, "squad leader recorded no_action; comments are not allowed for this task")
return
}
}
sourceTaskID = task.ID
}
}
// NOTE: Comment content is stored as Markdown source. XSS is handled at the
// rendering layer (rehype-sanitize) and at the editor layer
// (@tiptap/markdown with html:false). Running an HTML sanitizer here would
// entity-encode Markdown syntax characters (>, ", &, <) and corrupt the
// source. See issue #1303 / discussion in MUL-1119, MUL-1125.
// parent_id stores the exact comment being replied to. Thread-level behavior
// (for example auto-unresolving a resolved thread) resolves the root
// separately so storing a reply-to-reply does not destroy the direct-parent
// signal used by trigger decisions.
var rootComment *db.Comment
if parentID.Valid {
if root, err := h.Queries.GetThreadRoot(r.Context(), db.GetThreadRootParams{
CommentID: parentID,
WorkspaceID: issue.WorkspaceID,
}); err == nil {
rootComment = &root
}
}
createParams := db.CreateCommentParams{
ID: dbid.NewV7(),
IssueID: issue.ID,
WorkspaceID: issue.WorkspaceID,
AuthorType: authorType,
AuthorID: parseUUID(authorID),
Content: req.Content,
Type: req.Type,
ParentID: parentID,
SourceTaskID: sourceTaskID,
// The author's "don't start" choices outlive this call: completion
// replay skips the agents it names.
SuppressedAgentIds: suppressAgentIDs,
}
var created db.CreateCommentRow
var err error
if len(attachmentIDs) > 0 {
// A comment and the attachments it was posted with are one visible
// change, and one database outcome. Committing the comment first leaves
// a window in which it can gain a reply and be deleted into a
// tombstone, and the link would then bind uploaded objects to that
// placeholder — invisible, and missed by the prune's storage cleanup
// (#8296 review). Inside this transaction the row is not yet visible to
// a delete, so there is no window at all. Mirrors the attachment-set
// edit in UpdateComment.
//
// Owner first: the CreateComment statement takes the issue row, and only
// then are the attachments locked — the issue -> comment -> child order
// LockIssueForDelete, UpdateComment, LockLiveComment and DeleteAttachment
// all share. Taking the attachments first inverts it against issue
// teardown, which holds the issue and then reaches the same rows through
// the issue_id cascade, and Postgres aborts one side. Locking them at all
// still pins the requested set: an attachment deleted while this waited
// is refused before the comment is committed, rather than the comment
// committing without it.
tx, beginErr := h.beginWakeupWrite(r.Context())
if beginErr != nil {
slog.Warn("create comment failed", append(logger.RequestAttrs(r), "error", beginErr, "issue_id", issueID)...)
writeError(w, http.StatusInternalServerError, "failed to create comment")
return
}
defer tx.Rollback(r.Context())
qtx := h.Queries.WithTx(tx)
created, err = qtx.CreateComment(r.Context(), createParams)
if err == nil {
var missing pgtype.UUID
missing, err = lockCommentAttachments(r.Context(), qtx, issue.WorkspaceID, issue.ID, attachmentIDs)
if err == nil && missing.Valid {
writeError(w, http.StatusConflict, "attachment "+uuidToString(missing)+" is no longer available")
return
}
}
if err == nil {
err = qtx.LinkAttachmentsToComment(r.Context(), db.LinkAttachmentsToCommentParams{
CommentID: created.ID,
IssueID: issue.ID,
Column3: attachmentIDs,
})
}
if err == nil {
err = tx.Commit(r.Context())
}
} else {
created, err = wakeupWrite(h, r, func(q *db.Queries) (db.CreateCommentRow, error) {
return q.CreateComment(r.Context(), createParams)
})
}
if errors.Is(err, pgx.ErrNoRows) {
// The issue was deleted, possibly while this waited for its row lock.
writeError(w, http.StatusNotFound, "issue not found")
return
}
if err != nil {
slog.Warn("create comment failed", append(logger.RequestAttrs(r), "error", err, "issue_id", issueID)...)
writeError(w, http.StatusInternalServerError, "failed to create comment")
return
}
comment := created.Comment()
// Fetch linked attachments so the response includes them.
groupedAtt := h.groupAttachments(r, []pgtype.UUID{comment.ID})
resp := commentToResponse(comment, nil, groupedAtt[uuidToString(comment.ID)])
resp.IssueRevision = created.IssueRevision
slog.Info("comment created", append(logger.RequestAttrs(r), "comment_id", uuidToString(comment.ID), "issue_id", issueID)...)
h.publish(protocol.EventCommentCreated, uuidToString(issue.WorkspaceID), authorType, authorID, map[string]any{
"comment": resp,
"issue_title": issue.Title,
"issue_assignee_type": textToPtr(issue.AssigneeType),
"issue_assignee_id": uuidToPtr(issue.AssigneeID),
"issue_status": issue.Status,
"issue_revision": created.IssueRevision,
})
// A reply in a resolved thread re-opens it. Done after CreateComment commits
// so the reply is visible regardless of the unresolve outcome. Shared with
// the agent task path (TaskService.createAgentComment) — both reply paths
// must keep the resolved root in sync.
h.TaskService.AutoUnresolveThreadOnReply(r.Context(), rootComment, uuidToString(issue.WorkspaceID), authorType, authorID, h.wakeupSourceTaskID(r))
originatorUserID := h.invokeOriginatorFromRequest(r, authorType, authorID)
// The comment is already saved; a blocked mention must not fail the whole
// request. Surface the per-target outcomes so the client can show partial
// success instead of a silent no-op (MUL-4525 §2).
resp.TriggerOutcomes = h.triggerTasksForComment(r.Context(), issue, comment, parentComment, authorType, authorID, originatorUserID, suppressAgentIDs, steer)
if !steer.empty() {
applyCommentSupplements(&resp, h.listCommentSupplements(r.Context(), issue.WorkspaceID, []pgtype.UUID{comment.ID})[uuidToString(comment.ID)])
}
writeJSON(w, http.StatusCreated, resp)
}
// clientAuthorableCommentTypes is what POST /comments accepts. `status_change`
// and `system` are platform-generated narration, so a client claiming them
// would be forging system output; they are deliberately absent.
var clientAuthorableCommentTypes = map[string]struct{}{
"comment": {},
"progress_update": {},
}
func isClientAuthorableCommentType(t string) bool {
_, ok := clientAuthorableCommentTypes[t]
return ok
}
// noteCommentPrefix marks a comment as a human-only note. A comment whose first
// whitespace-delimited token is this prefix (case-insensitive) is stored like
// any other comment but never triggers an agent.
const noteCommentPrefix = "/note"
// isNoteComment reports whether content opts out of agent triggering via the
// reserved /note prefix. The prefix must be the comment's first token, so
// "/note check expiry", " /NOTE", and "/note" all match, while "/notes",
// "/ note", and "see foo/note" do not.
func isNoteComment(content string) bool {
trimmed := strings.TrimLeft(content, " \t\r\n")
firstToken := trimmed
if i := strings.IndexFunc(trimmed, unicode.IsSpace); i >= 0 {
firstToken = trimmed[:i]
}
return strings.EqualFold(firstToken, noteCommentPrefix)
}
// triggerTasksForComment resolves and enqueues the comment's agent triggers and
// returns the per-target outcomes for explicit @agent / @squad mentions
// (MUL-4525 §2): blocked mentions from resolution plus queued / coalesced /
// deferred / blocked from enqueue. UI-suppressed triggers (the user unchecked
// them) are removed before enqueue and produce no outcome.
//
// steer names the running turns the author chose: a recipient whose chosen
// turn is still running receives this comment there instead of a follow-up run.
func (h *Handler) triggerTasksForComment(ctx context.Context, issue db.Issue, comment db.Comment, parentComment *db.Comment, actorType, actorID, originatorUserID string, suppressAgentIDs []pgtype.UUID, steer commentSteerRequest) []CommentTriggerOutcome {
if isNoteComment(comment.Content) {
return nil
}
triggers, targets := h.computeCommentAgentTriggers(ctx, issue, comment.Content, parentComment, actorType, actorID, commentTriggerComputeOptions{
ExcludeTriggerCommentID: comment.ID,
AuthoringTaskID: comment.SourceTaskID,
OriginatorUserID: originatorUserID,
})
triggers = filterSuppressedCommentAgentTriggers(triggers, suppressAgentIDs)
h.noteBlockedRuntimeTargets(ctx, issue, targets)
triggers, steered := h.steerCommentAgentTriggers(ctx, issue, comment, actorType, actorID, originatorUserID, triggers, steer)
enqueued := h.enqueueCommentAgentTriggers(ctx, issue, comment.ID, triggers)
for agentID, result := range steered {
enqueued[agentID] = result
}
outcomes := commentTriggerOutcomes(targets, enqueued)
h.noteCommentInvocationDenied(ctx, issue, comment, outcomes)
return outcomes
}
// noteBlockedRuntimeTargets leaves one system comment per agent refused for an
// unusable runtime, whoever authored the trigger — see noteRuntimeUnusable for
// why a human author gets one too. Deduped by agent so a comment naming both
// @Agent and the squad it leads produces a single notice.
//
// Called from the trigger path only. The resolver that produced these targets
// also runs for the composer preview, where writing a comment would be a side
// effect of typing.
func (h *Handler) noteBlockedRuntimeTargets(ctx context.Context, issue db.Issue, targets []commentMentionTarget) {
var noted map[string]struct{}
for _, t := range targets {
if t.unusable == nil {
continue
}
id := uuidToString(t.unusable.agent.ID)
if _, done := noted[id]; done {
continue
}
if noted == nil {
noted = make(map[string]struct{}, 1)
}
noted[id] = struct{}{}
h.noteRuntimeUnusable(ctx, issue, t.unusable.agent, t.unusable.verdict)
}
}
func filterSuppressedCommentAgentTriggers(triggers []commentAgentTrigger, suppressAgentIDs []pgtype.UUID) []commentAgentTrigger {
if len(triggers) == 0 || len(suppressAgentIDs) == 0 {
return triggers
}
suppressed := make(map[string]struct{}, len(suppressAgentIDs))
for _, id := range suppressAgentIDs {
if id.Valid {
suppressed[uuidToString(id)] = struct{}{}
}
}
if len(suppressed) == 0 {
return triggers
}
filtered := make([]commentAgentTrigger, 0, len(triggers))
for _, trigger := range triggers {
if _, ok := suppressed[uuidToString(trigger.Agent.ID)]; ok {
continue
}
filtered = append(filtered, trigger)
}
return filtered
}
// commentEnqueueResult is the domain outcome of enqueuing ONE executing agent.
// execSquadID is the squad whose leader context the run actually carries (set
// only for a squad-leader execution), so a DIFFERENT squad that shares this
// leader can be reported honestly as coalesced rather than as if its own leader
// context ran.
type commentEnqueueResult struct {
status DispatchStatus
reason DispatchReasonCode
execSquadID string
}
// enqueueCommentAgentTriggers enqueues each resolved trigger (already deduped by
// executing agent) and returns the result keyed by executing agent id
// (MUL-4525 §2). Outcomes are later fanned from these to every explicit mention
// target that resolved to the agent, so coalescing a run never drops a named
// target's outcome. queued / coalesced / deferred are success-shaped (the run
// was handled, no duplicate task); only a real enqueue failure is blocked.
func (h *Handler) enqueueCommentAgentTriggers(ctx context.Context, issue db.Issue, triggerCommentID pgtype.UUID, triggers []commentAgentTrigger) map[string]commentEnqueueResult {
results := make(map[string]commentEnqueueResult, len(triggers))
record := func(trigger commentAgentTrigger, status DispatchStatus, reason DispatchReasonCode) {
execSquadID := ""
if trigger.Squad != nil {
execSquadID = uuidToString(trigger.Squad.ID)
}
results[uuidToString(trigger.Agent.ID)] = commentEnqueueResult{status: status, reason: reason, execSquadID: execSquadID}
}
for _, trigger := range triggers {
status, reason := h.resolveCommentTriggerEnqueue(ctx, issue, trigger, triggerCommentID)
record(trigger, status, reason)
}
return results
}
// resolveCommentTriggerEnqueue resolves ONE trigger into its final dispatch
// outcome, folding the comment into an existing pending task rather than
// dropping it (MUL-4195). It handles the two ways the (issue, agent) pair can
// already hold a pending task:
//
// - resolution pre-flagged AlreadyPending, or
// - a fresh enqueue LOST the race to a concurrent sibling and the
// idx_one_pending_task_per_issue_agent unique index rejected the insert
// (service.ErrDuplicatePendingTask, #5914).
//
// Both mean "a task now exists". INVARIANT: a success-shaped outcome (coalesced /
// deferred / queued) is returned only when the comment is provably attached to a
// run that will address it:
//
// - coalesced — an atomic head-scoped merge folded it into a queued task;
// - queued — a fresh task was created for it;
// - deferred — an active task's completion reconcile will replay it, either
// because a planned-id write landed on a claim-receipt task
// (worker reply, or lost enqueue race where the comment may predate the task) or
// because the comment is newer than that task and therefore
// inside reconcile's `created_at > since` window (AlreadyPending
// path for member/explicit-mention routes, where the task existed first).
//
// The deferral is not a one-shot prediction: if a replay is later blocked by a
// task that cannot cover the comment, reconcileCommentsOnCompletion hands the
// obligation on (propagateUncoveredCommentObligation) instead of discarding it,
// and that hand-off retries across the blocker's state changes (#5914, Elon
// rounds 8–9). It is best-effort in exactly one shape — a DIFFERENT-head QUEUED
// blocker, which can neither cover the comment nor accept it without violating
// TEN-356 — and that case is logged at error rather than hidden; closing it needs
// a durable obligation record, which is deliberately out of scope here.
// It never defers off a mere classification of the current active tasks — a
// snapshot cannot prove anything about state after it is read (round 7).
// A duplicate that cannot yet be resolved re-loops (bounded by maxAttempts); on
// genuine non-convergence it returns a truthful internal_error, never a fabricated
// deferred that would silently drop the comment.
func (h *Handler) resolveCommentTriggerEnqueue(ctx context.Context, issue db.Issue, trigger commentAgentTrigger, triggerCommentID pgtype.UUID) (DispatchStatus, DispatchReasonCode) {
pending := trigger.AlreadyPending
lostRace := false
// Resolve the reviewed HEAD lazily and at most once — the common
// non-pending path enqueues without ever needing it, so it stays off the
// hot path. It keys both the head-scoped merge and the planned-comment
// registration, matching the AlreadyPending pre-check's head (TEN-356).
var headSha pgtype.Text
headShaLoaded := false
getHeadSha := func() pgtype.Text {
if !headShaLoaded {
headSha = h.TaskService.ResolveIssueReviewSHAParam(ctx, issue.ID)
headShaLoaded = true
}
return headSha
}
// Bounded retry (#5914, Elon round 3). INVARIANT: never return a
// success-shaped outcome (coalesced / deferred / queued) without a COMPLETED
// merge, planned-id registration, fresh enqueue, or a confirmed different-head
// deferral. A duplicate that cannot be durably resolved re-loops; on genuine
// non-convergence we return a truthful internal_error below, never a
// fabricated deferred that would silently drop the comment. maxAttempts is a
// concurrent-churn backstop — each duplicate is followed by a merge/register
// attempt, so the loop only spins while a sibling keeps flipping state.
const maxAttempts = 4
for attempt := 0; attempt < maxAttempts; attempt++ {
if pending {
// (a) Fold into a same-head QUEUED task. This is the ONLY path that
// ATOMICALLY re-attributes the run (trigger / originator / accountable
// / overlay / connected apps) to the folded comment, so a queued
// winner must always come through here, never a bare planned append
// (MUL-4302). HEAD-scoped so a DIFFERENT-head task is never folded
// (TEN-356) — the merge misses and this falls through. The merge
// reports HOW it resolved: coalesced on success; blocked on a
// fail-closed / failed merge — never mislabeled as success (MUL-4525 §2).
if status, reason, terminal := commentMergeTerminalOutcome(
h.mergeCommentIntoPendingTask(ctx, issue, trigger, triggerCommentID, getHeadSha()),
); terminal {
return status, reason
}
// No same-head QUEUED row to fold into (merge missed). The two paths
// resolve differently.
if !lostRace && !trigger.NonLeaderAgentReply {
// (b) AlreadyPending path: this comment arrived AFTER its task, so
// it is newer than the task and completion reconcile covers it by
// timestamp; MUL-4195 leaves the claimed task untouched. Defer to
// the active task, or enqueue fresh if it has since finished.
active, activeErr := h.hasActiveTaskForIssueAndAgent(ctx, issue.ID, trigger.Agent.ID, triggerCommentID)
if status, reason, enqueueFresh := decidePostMergeMiss(active, activeErr); !enqueueFresh {
return status, reason
}
// enqueueFresh → fall through to the enqueue below.
} else {
// (c) A worker reply needs a recorded obligation: unlike explicit
// mentions, timestamp-only agent replies are not replayed (loop
// safety). The same registration also covers a lost INSERT race,
// where the losing comment can PREDATE the winner,
// which completion reconcile's `created_at > since` window cannot
// see. Register it as a planned (undelivered) input on a same-head
// CLAIM-RECEIPT task (dispatched/running/waiting; queued is excluded
// — it must fold through the atomic merge above), turning the drop
// into a bounded follow-up (#5914, Elon round 2).
registered, err := h.registerPlannedCommentForActiveTask(ctx, issue, trigger.Agent.ID, triggerCommentID, getHeadSha())
if err != nil {
slog.Warn("register planned comment failed",
"issue_id", uuidToString(issue.ID), "agent_id", uuidToString(trigger.Agent.ID), "error", err)
return DispatchBlocked, ReasonInternalError
}
if registered {
return DispatchDeferred, ReasonDeferred
}
// (d) No same-head QUEUED task to fold (merge missed) and no
// same-head CLAIM-RECEIPT task to register on. Do NOT try to defer
// off a classification of the current active tasks: a snapshot only
// proves state NOW, and a DIFFERENT-head task's reconcile covering
// this comment can be defeated by a newer blocker that appears after
// the snapshot but before that task completes (its replay collides
// with the blocker and completion reconcile discards the failure —
// Elon round 7). Every success-shaped outcome must be backed by a
// COMPLETED write, never a prediction. So just fall through to a
// fresh enqueue: if the unique slot is still occupied — necessarily
// by a different-head task we can neither fold into nor durably cover
// — the insert re-raises ErrDuplicatePendingTask and we re-loop,
// converging to a truthful non-success; if the slot is free, the
// enqueue creates the covering task. (The occupant's own completion
// reconcile still replays this comment best-effort in the common
// no-blocker case; we simply never PROMISE it.)
}
}
if err := h.enqueueSingleCommentTrigger(ctx, issue, triggerCommentID, trigger); err != nil {
// Lost the enqueue race: a sibling task for this (issue, agent) now
// exists. Re-resolve as pending so the next attempt folds this
// comment into that sibling (queued) or durably registers it
// (dispatched) instead of dropping it (#5914).
if errors.Is(err, service.ErrDuplicatePendingTask) {
pending = true
lostRace = true
continue
}
return DispatchBlocked, commentEnqueueFailureReason(err)
}
return DispatchQueued, ReasonQueued
}
// Non-convergence after the bounded retries (only reachable under sustained
// concurrent churn where a sibling keeps flipping state under us): surface a
// truthful non-success — NEVER a fabricated deferred/coalesced that would
// report the comment covered when no task provably covers it (#5914, Elon
// round 3). The comment row persists; a later trigger/edit can still cover it.
slog.Warn("comment trigger enqueue did not converge; reporting non-success",
"issue_id", uuidToString(issue.ID), "agent_id", uuidToString(trigger.Agent.ID))
return DispatchBlocked, ReasonInternalError
}
// commentTriggerOutcomes maps each explicit mention target to its final outcome
// (MUL-4525 §2): a target that resolved to an executing agent takes that agent's
// enqueue status, so several mentions coalescing into one run each still get
// their own outcome; a terminal target (blocked / self-suppressed) carries its
// own status. A target whose executing agent has no enqueue result — because the
// composer suppressed (unchecked) it — yields no outcome, since the user opted
// out deliberately.
func commentTriggerOutcomes(targets []commentMentionTarget, enqueued map[string]commentEnqueueResult) []CommentTriggerOutcome {
if len(targets) == 0 {
return nil
}
outcomes := make([]CommentTriggerOutcome, 0, len(targets))
for _, t := range targets {
if t.ExecAgentID != "" {
res, ok := enqueued[t.ExecAgentID]
if !ok {
continue
}
status, reason := res.status, res.reason
// A @squad whose shared leader ran, but under a DIFFERENT squad's
// context, did not get its own leader briefing injected — the single
// leader run (one task per issue+agent) merely folds it in. Report
// coalesced, not queued, so we never claim this squad's leader
// context executed (MUL-4525, Elon round 3).
if t.TargetType == "squad" && res.execSquadID != "" && res.execSquadID != t.TargetID && status == DispatchQueued {
status, reason = DispatchCoalesced, ReasonCoalesced
}
outcomes = append(outcomes, CommentTriggerOutcome{TargetType: t.TargetType, TargetID: t.TargetID, Status: status, ReasonCode: reason})
continue
}
outcomes = append(outcomes, CommentTriggerOutcome{TargetType: t.TargetType, TargetID: t.TargetID, Status: t.Status, ReasonCode: t.ReasonCode})
}
return outcomes
}
// commentBlockedTargetOutcomes is the composer-preview projection: the explicit
// mentions that will NOT trigger if the comment is posted as-is (MUL-4525 §2). A
// resolvable/executing target instead appears in the preview `agents` list, so
// only terminal blocked targets surface here.
func commentBlockedTargetOutcomes(targets []commentMentionTarget) []CommentTriggerOutcome {
var blocked []CommentTriggerOutcome
for _, t := range targets {
if t.Status == DispatchBlocked {
blocked = append(blocked, CommentTriggerOutcome{TargetType: t.TargetType, TargetID: t.TargetID, Status: t.Status, ReasonCode: t.ReasonCode})
}
}
return blocked
}
// commentEnqueueFailureReason types an enqueue error that reached the response
// (MUL-4525 §2). The admission gate (canInvokeAgent / archived / runtime) already
// ran during resolution, so a failure here is either a fail-closed attribution
// refusal (attribution_blocked, typed via errors.Is) or a rare race /
// infrastructure error that stays an unclassified internal error rather than
// leaking the raw message.
func commentEnqueueFailureReason(err error) DispatchReasonCode {
if errors.Is(err, service.ErrAttributionFailClosed) {
return ReasonAttributionBlocked
}
return ReasonInternalError
}
// hasActiveTaskForIssueAndAgent reports whether the (issue, agent) pair has any
// non-terminal task whose completion will drive completion reconciliation. It
// returns the query error rather than swallowing it (MUL-4525, Elon round 4):
// callers must fail closed on error (never enqueue a possibly-colliding
// duplicate) AND must not report a success — "cannot confirm whether a run is
// active" is never the same as "a run is active". See decidePostMergeMiss /
// decideSuppressedLeaderOutcome for the two decisions.
func (h *Handler) hasActiveTaskForIssueAndAgent(ctx context.Context, issueID, agentID, threadCommentID pgtype.UUID) (bool, error) {
active, err := h.Queries.HasActiveTaskForIssueAndAgentInThread(ctx, db.HasActiveTaskForIssueAndAgentInThreadParams{
ThreadCommentID: threadCommentID,
IssueID: issueID,
AgentID: agentID,
})
if err != nil {
slog.Warn("has active task for issue+agent check failed",
"issue_id", uuidToString(issueID), "agent_id", uuidToString(agentID), "error", err)
return false, err
}
return active, nil
}
// decidePostMergeMiss decides what to do after a comment merge missed on a
// target that had a pending task (MUL-4525, Elon round 4). On a query failure
// (activeErr != nil) it FAILS CLOSED: never enqueue a fresh task — a duplicate
// concurrent run risk — and report a non-success internal_error, since we cannot
// confirm a run is active. A confirmed active task defers to that run's
// reconcile; only a confirmed-none enqueues a fresh follow-up.
func decidePostMergeMiss(active bool, activeErr error) (status DispatchStatus, reason DispatchReasonCode, enqueueFresh bool) {
switch {
case activeErr != nil:
return DispatchBlocked, ReasonInternalError, false
case active:
return DispatchDeferred, ReasonDeferred, false
default:
return "", "", true
}
}
// decideSuppressedLeaderOutcome maps the self-trigger-suppressed squad leader's
// active-task check to an honest outcome (MUL-4525, Elon round 4). A query
// failure is never success — it is internal_error, not a fabricated deferred.
// A confirmed active run defers (its reconcile covers the comment); otherwise
// nothing runs and the outcome is self_trigger_suppressed.
func decideSuppressedLeaderOutcome(active bool, activeErr error) (DispatchStatus, DispatchReasonCode) {
switch {
case activeErr != nil:
return DispatchBlocked, ReasonInternalError
case active:
return DispatchDeferred, ReasonAlreadyActive
default:
return DispatchBlocked, ReasonSelfTriggerSuppressed
}
}
// commentMergeResult distinguishes how a pending-task merge attempt resolved so
// the caller can report an HONEST outcome (MUL-4525, Elon round 5). A real merge
// is coalesced, but a REFUSED or FAILED merge — even when we correctly fail
// closed by keeping the original task and not enqueuing a duplicate — must NOT
// be reported as a success-shaped coalesced.
type commentMergeResult int
const (
// commentMergeSucceeded: the comment folded into the queued task → coalesced.
commentMergeSucceeded commentMergeResult = iota
// commentMergeNoPendingTask: no queued task to merge into anymore (it was
// claimed/started between the dedup check and now). The caller runs the
// active-task decision (defer vs fresh enqueue).
commentMergeNoPendingTask
// commentMergeAttributionBlocked: fail-closed attribution refused re-stamping
// the merge. The original task is kept and no fresh task is enqueued, but the
// re-attribution did NOT happen → outcome attribution_blocked, not success.
commentMergeAttributionBlocked
// commentMergeError: an unknown attribution/DB error. Fail closed (keep the
// task, no duplicate enqueue), but the merge did not complete → outcome
// internal_error, not success.
commentMergeError
)
// commentMergeTerminalOutcome maps a merge result that carries its own final
// outcome (everything except commentMergeNoPendingTask, which needs the
// active-task decision) to the reported (status, reason). terminal=false only
// for commentMergeNoPendingTask.
func commentMergeTerminalOutcome(result commentMergeResult) (status DispatchStatus, reason DispatchReasonCode, terminal bool) {
switch result {
case commentMergeSucceeded:
return DispatchCoalesced, ReasonCoalesced, true
case commentMergeAttributionBlocked:
return DispatchBlocked, ReasonAttributionBlocked, true
case commentMergeError:
return DispatchBlocked, ReasonInternalError, true
default: // commentMergeNoPendingTask
return "", "", false
}
}
// mergeCommentIntoPendingTask folds a newly-arrived comment into the existing
// pre-claim task for (issue, agent) instead of dropping it
// (MUL-4195). It reports HOW it resolved via commentMergeResult so the caller
// never mislabels a refused/failed merge as success (MUL-4525 §2). No path here
// enqueues a duplicate: on any failure the original task is kept intact, so the
// comment is still read by that run and its instruction is not lost — only the
// re-attribution / merge bookkeeping is declined, and that is surfaced honestly.
//
// Recompute-on-merge (MUL-4195 review must-fix #1): on success the run's
// originator_user_id, runtime_mcp_overlay and runtime_connected_apps are
// re-stamped to the NEW trigger comment's originator, and trigger_summary is
// refreshed — so a different member's comment safely folds into a task another
// member created, the coalescing run carrying the latest instruction's
// originator and matching connected-app overlay.
func (h *Handler) mergeCommentIntoPendingTask(ctx context.Context, issue db.Issue, trigger commentAgentTrigger, newTriggerCommentID pgtype.UUID, headSha pgtype.Text) commentMergeResult {
// Re-attribute the coalescing run to the new comment's human atomically: the
// whole attribution snapshot moves, not just the person columns (MUL-4302). An
// issue-assignee reaction is comment_source; a mention / thread-parent /
// conversation hop is delegation.
isMention := trigger.Source != commentTriggerSourceIssueAssignee
attr, err := h.TaskService.AttributionForMergedComment(ctx, issue.WorkspaceID, newTriggerCommentID, isMention, trigger.Agent)
if err != nil {
// The new comment cannot be re-attributed. REFUSE the merge — keep the
// existing queued task on its original (precise) snapshot rather than
// re-stamp it to a degraded owner_fallback, and never spawn a duplicate.
// A fail-closed refusal is a distinct, honest outcome (attribution_blocked);
// any other error is unclassified (internal_error) — neither is success.
slog.Warn("refused comment merge: attribution failed, keeping original task snapshot",
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(trigger.Agent.ID),
"new_trigger_comment_id", uuidToString(newTriggerCommentID),
"error", err)
if errors.Is(err, service.ErrAttributionFailClosed) {
return commentMergeAttributionBlocked
}
return commentMergeError
}
overlay, connectedApps := h.TaskService.BuildRuntimeMCPOverlayForMerge(ctx, attr.UserID, trigger.Agent)
row, err := h.Queries.MergeCommentIntoPendingTask(ctx, db.MergeCommentIntoPendingTaskParams{
IssueID: issue.ID,
AgentID: trigger.Agent.ID,
NewTriggerCommentID: newTriggerCommentID,
NewOriginatorUserID: attr.UserID,
NewAccountableUserID: attr.AccountableUserID,
NewOriginatorSource: pgtype.Text{String: attr.Source.String(), Valid: true},
NewDelegatedFromTaskID: attr.DelegatedFromTaskID,
NewRuleVersionID: attr.RuleVersionID,
NewTriggerEvidenceKind: pgtype.Text{String: string(attr.EvidenceKind), Valid: attr.EvidenceKind != ""},
NewTriggerEvidenceRefID: attr.EvidenceRefID,
NewTriggerSummary: h.TaskService.BuildCommentTriggerSummary(ctx, issue.WorkspaceID, newTriggerCommentID),
NewRuntimeMcpOverlay: overlay,
NewRuntimeConnectedApps: connectedApps,
HeadSha: headSha,
})
if err != nil {
if isNotFound(err) {
// No pre-claim (queued/deferred) task to merge into. The caller
// defers to completion reconcile when an active task exists, or
// enqueues fresh when none does.
return commentMergeNoPendingTask
}
// Unknown error: the pending task most likely still exists, so do NOT
// risk enqueuing a duplicate — but the merge did not happen, so this is
// not a success.
slog.Warn("merge comment into pending task failed",
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(trigger.Agent.ID),
"error", err)
return commentMergeError
}
slog.Info("merged comment into pending task",
"task_id", uuidToString(row.ID),
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(trigger.Agent.ID),
"new_trigger_comment_id", uuidToString(newTriggerCommentID),
"coalesced_count", len(row.CoalescedCommentIds))
// The fold re-attributes the queued run to this comment's author; record that
// human's action now, since the run row keeps no history of who acted.
h.TaskService.RecordHumanDelegation(ctx, attr)
// The run's status is unchanged, so no lifecycle event tells clients that it
// now answers this comment.
h.TaskService.BroadcastTaskCoalesced(ctx, row)
return commentMergeSucceeded
}
// registerPlannedCommentForActiveTask durably folds an accepted comment into the
// same-head active task's planned (coalesced) set when the queued merge could no
// longer target it (the winner was already claimed → dispatched/running). It
// returns true when a same-head active task absorbed the comment; (false, nil)
// means no same-head active task exists (a DIFFERENT-head task holds the slot, or
// the task just terminated), so the caller falls back to the active-task
// decision. Delivered ids are untouched, so completion reconcile replays the
// comment as a single bounded follow-up (#5914). See the query doc.
func (h *Handler) registerPlannedCommentForActiveTask(ctx context.Context, issue db.Issue, agentID, commentID pgtype.UUID, headSha pgtype.Text) (bool, error) {
row, err := h.Queries.RegisterPlannedCommentForActiveTask(ctx, db.RegisterPlannedCommentForActiveTaskParams{
CommentID: commentID,
IssueID: issue.ID,
AgentID: agentID,
HeadSha: headSha,
})
if err != nil {
if isNotFound(err) {
return false, nil
}
return false, err
}
slog.Info("registered comment as planned follow-up input",
"task_id", uuidToString(row.ID),
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(agentID),
"comment_id", uuidToString(commentID),
"coalesced_count", len(row.CoalescedCommentIds))
return true, nil
}
// propagateUncoveredCommentObligation hands an UNCOVERED comment's replay
// obligation to whichever active task now occupies the (issue, agent) slot, so a
// blocked completion replay is never silently dropped (#5914, Elon round 8).
//
// Without this, a deferral is only ever consumed ONCE: task A's completion
// reconcile replays comment C, the enqueue collides with a newer blocker B, the
// failure is discarded, and B never sees C (C predates B and is not in its
// planned ids) — so C is lost even though the original response promised it would
// be handled. Propagating turns "consumed once" into "handed along" until some
// task provably covers C.
//
// Each attempt tries, in order, the three writes that can durably cover the
// comment, and RETRIES when the slot's occupant changes state under us — the
// hand-off is not one atomic statement, so a blocker that completes or gets
// claimed between two steps must be re-observed rather than lost (Elon round 9):
//
// 1. a CLAIM-RECEIPT task (dispatched/running/waiting), ANY head: a planned id
// added after the claim is never delivered to that run, so it cannot consume
// the comment under the wrong head — it purely schedules the comment for that
// task's own completion reconcile, which re-enqueues at the head current then;
// 2. a SAME-HEAD queued task: the atomic merge, which re-stamps trigger /
// originator / overlay (MUL-4302). This is HEAD-SCOPED on purpose: merging
// into a different-head queued task would make the comment that run's trigger
// and mark it delivered at claim time, so an old-head run would consume a
// new-head request and no new-head follow-up would ever be created — the
// TEN-356 violation (Elon round 9, must-fix 2);
// 3. a fresh enqueue: correct precisely when the blocker has finished and freed
// the unique slot, which is the state a mid-hand-off completion leaves behind.
//
// A duplicate-key error from (3) means the slot is still held, so the loop
// re-observes and tries again. Returns false only when every attempt found the
// slot held by a task that can neither cover nor accept the comment (a
// different-head QUEUED blocker); the caller logs that loudly — it is the one
// shape today's schema has no safe place to park, and closing it needs a durable
// obligation record rather than an in-request hand-off.
func (h *Handler) propagateUncoveredCommentObligation(ctx context.Context, issue db.Issue, trigger commentAgentTrigger, commentID pgtype.UUID, headSha pgtype.Text) bool {
const maxAttempts = 3
for attempt := 0; attempt < maxAttempts; attempt++ {
registered, err := h.registerPlannedCommentForActiveTask(ctx, issue, trigger.Agent.ID, commentID, pgtype.Text{})
if err != nil {
slog.Warn("propagate comment obligation: register on active task failed",
"issue_id", uuidToString(issue.ID), "agent_id", uuidToString(trigger.Agent.ID),
"comment_id", uuidToString(commentID), "error", err)
} else if registered {
return true
}
if h.mergeCommentIntoPendingTask(ctx, issue, trigger, commentID, headSha) == commentMergeSucceeded {
return true
}
err = h.enqueueSingleCommentTrigger(ctx, issue, commentID, trigger)
if err == nil {
return true
}
if !errors.Is(err, service.ErrDuplicatePendingTask) {
return false
}
// The slot is still occupied; its occupant may have changed state between
// the steps above, so re-observe and try again.
}
return false
}
// logCommentEnqueueFailure logs a failed comment-trigger enqueue. A benign
// duplicate-pending-task race (service.ErrDuplicatePendingTask) is not a fault —
// resolveCommentTriggerEnqueue folds the comment into the winning task — so it
// is logged at debug and never surfaces as a warning, keeping the raw Postgres
// constraint name out of warn/error logs (#5914, Elon review). Any other error
// stays a warning with the error attached.
func logCommentEnqueueFailure(msg string, err error, attrs ...any) {
if errors.Is(err, service.ErrDuplicatePendingTask) {
slog.Debug(msg+": duplicate pending task, coalescing into the sibling run", attrs...)
return
}
slog.Warn(msg, append(attrs, "error", err)...)
}
// enqueueSingleCommentTrigger creates a fresh task for one computed trigger.
// Split out of enqueueCommentAgentTriggers so the merge-not-drop path
// (MUL-4195) can fall back to it when a pending task vanished mid-flight.
// enqueueSingleCommentTrigger enqueues one resolved trigger and returns the
// enqueue error (nil on success) so the caller can surface a
// trigger_outcome (MUL-4525 §2).
func (h *Handler) enqueueSingleCommentTrigger(ctx context.Context, issue db.Issue, triggerCommentID pgtype.UUID, trigger commentAgentTrigger) error {
switch trigger.Source {
case commentTriggerSourceIssueAssignee:
if trigger.Squad != nil {
if _, err := h.TaskService.EnqueueTaskForSquadLeader(ctx, issue, trigger.Agent.ID, trigger.Squad.ID, triggerCommentID, service.OriginDerived); err != nil {
logCommentEnqueueFailure("enqueue squad leader task failed", err,
"issue_id", uuidToString(issue.ID),
"squad_id", uuidToString(trigger.Squad.ID),
"leader_id", uuidToString(trigger.Agent.ID))
return err
}
return nil
}
if _, err := h.TaskService.EnqueueTaskForIssue(ctx, issue, triggerCommentID); err != nil {
// EnqueueTaskForIssue now returns ErrDuplicatePendingTask on the
// benign duplicate-pending-task race, so use the shared helper that
// downgrades that case to debug — matching the squad-leader and
// mention cases and letting resolveCommentTriggerEnqueue coalesce
// instead of surfacing a warning (#5914).
logCommentEnqueueFailure("enqueue agent task on comment failed", err,
"issue_id", uuidToString(issue.ID))
return err
}
case commentTriggerSourceMentionSquadLeader:
if _, err := h.TaskService.EnqueueTaskForSquadLeader(ctx, issue, trigger.Agent.ID, trigger.Squad.ID, triggerCommentID, service.OriginNamed); err != nil {
logCommentEnqueueFailure("enqueue squad leader mention task failed", err,
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(trigger.Agent.ID))
return err
}
case commentTriggerSourceMentionAgent:
if _, err := h.TaskService.EnqueueTaskForMention(ctx, issue, trigger.Agent.ID, triggerCommentID, service.OriginNamed); err != nil {
logCommentEnqueueFailure("enqueue mention agent task failed", err,
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(trigger.Agent.ID))
return err
}
case commentTriggerSourceThreadParent, commentTriggerSourceConversation:
var err error
// Squad is set on these two sources only when the routing already
// proved a leader role to continue: the thread root's prior task for
// the conversation path, the replied-to comment's own authoring task
// for the thread-parent path. Gating on the source as well would keep
// the thread-parent path demoted for no reason (MUL-7006).
if trigger.Squad != nil {
_, err = h.TaskService.EnqueueTaskForSquadLeader(ctx, issue, trigger.Agent.ID, trigger.Squad.ID, triggerCommentID, service.OriginNamed)
} else {
_, err = h.TaskService.EnqueueTaskForThreadParent(ctx, issue, trigger.Agent.ID, triggerCommentID)
}
if err != nil {
logCommentEnqueueFailure("enqueue routed comment agent task failed", err,
"issue_id", uuidToString(issue.ID),
"agent_id", uuidToString(trigger.Agent.ID),
"source", trigger.Source)
return err
}
}
return nil
}
// computeCommentAgentTriggers resolves which agents a comment triggers (deduped
// by executing agent), plus the per-target list for every EXPLICIT @agent /
// @squad mention (MUL-4525 §2). Targets come only from the explicit-mention path
// — the implicit routing fallbacks (assignee, thread parent, conversation) were
// never named by the user, so a no-route there is not a silent no-op.
func (h *Handler) computeCommentAgentTriggers(ctx context.Context, issue db.Issue, content string, parentComment *db.Comment, actorType, actorID string, opts commentTriggerComputeOptions) ([]commentAgentTrigger, []commentMentionTarget) {
// A persisted comment determines its thread; previews use the parent.
// A new top-level preview has no thread yet and cannot merge with a queue.
opts.ThreadCommentID = opts.ExcludeTriggerCommentID
if !opts.ThreadCommentID.Valid && parentComment != nil {
opts.ThreadCommentID = parentComment.ID
}
if isNoteComment(content) {
return nil, nil
}
// Autopilot delegation authority (MUL-4857) is applied by the gate via
// opts.OriginatorUserID: when a run carried no human originator, the gate
// falls back to opts.AutopilotDelegationAuthorityUserID, which the caller has
// already resolved from a server-trusted, lineage-verified speaking task (see
// autopilotDelegationAuthority). Nothing is re-derived from issue provenance
// here, so an unrelated unattributed run cannot borrow a stranger autopilot
// creator's authority by commenting on that autopilot's issue.
mentions := util.ParseMentions(content)
// EXPLICIT @agent / @squad mentions are a direct request and win over the
// @all broadcast (MUL-5411). @all only suppresses the IMPLICIT routing
// fallbacks (assignee / thread parent / conversation) below — it must not
// swallow a target the author named by hand. Before this ordering, a
// comment carrying both `@all` and `@Preflight` enqueued nothing at all.
// `all` is neither "agent" nor "squad", so it is skipped inside
// resolveMentionedAgentCommentTriggers and never enqueues a run of its own.
if hasAgentOrSquadMention(mentions) {
return h.resolveMentionedAgentCommentTriggers(ctx, issue, mentions, actorType, actorID, opts)
}
if util.HasMentionAll(mentions) {
return nil, nil
}
if hasMemberMention(mentions) {
return nil, nil
}
if actorType != "member" {
// Agent-authored comments do not participate in the member-driven
// conversation routing (parent-author / thread-root continuation) or
// the member assignee fallback. Worker-result comments retain the
// leader→worker→leader loop: the assigned squad is resolved from the
// issue (MUL-3879), while a guest squad is resolved from the exact
// covered leader delegation (GH-8301). Explicit @agent / @squad mentions
// are already handled above, so neither fallback double-enqueues a
// mentioned target.
if actorType == "agent" {
if triggers, handled := h.routeGuestSquadLeaderFallback(ctx, issue, parentComment, actorID, opts); handled {
return triggers, nil
}
}
if issue.AssigneeType.Valid && issue.AssigneeType.String == "squad" {
if trigger, ok := h.routeAssignedSquadLeaderFallback(ctx, issue, actorType, actorID, opts); ok {
trigger.NonLeaderAgentReply = actorType == "agent" && actorID != uuidToString(trigger.Agent.ID)
return []commentAgentTrigger{trigger}, nil
}
}
return nil, nil
}
// A deleted parent no longer speaks for its agent author (#8296).
if parentComment != nil && parentComment.AuthorType == "agent" && !parentComment.DeletedAt.Valid {
trigger, ok := h.routeReplyToParentAuthor(ctx, issue, parentComment, actorType, actorID, opts)
if !ok {
return nil, nil
}
return []commentAgentTrigger{trigger}, nil
}
if parentComment != nil {
triggers, handled := h.routeThreadRootOwners(ctx, issue, parentComment, actorID, opts)
if handled {
if len(triggers) == 0 {
return nil, nil
}
return triggers, nil
}
// A plain member-to-member reply must not start the issue assignee just
// because the thread has no agent owner. Explicit mentions and existing
// conversation owners were already resolved above. The same holds for a
// reply under a deleted comment: it is still a reply, not a new request.
if parentComment.AuthorType == "member" || parentComment.DeletedAt.Valid {
return nil, nil
}
}
if trigger, ok := h.routeAssigneeFallback(ctx, issue, actorType, actorID, opts); ok {
return []commentAgentTrigger{trigger}, nil
}
return nil, nil
}
func hasAgentOrSquadMention(mentions []util.Mention) bool {
for _, m := range mentions {
if m.Type == "agent" || m.Type == "squad" {
return true
}
}
return false
}
func hasMemberMention(mentions []util.Mention) bool {
for _, m := range mentions {
if m.Type == "member" {
return true
}
}
return false
}
func (h *Handler) routeReplyToParentAuthor(ctx context.Context, issue db.Issue, parent *db.Comment, authorType, authorID string, opts commentTriggerComputeOptions) (commentAgentTrigger, bool) {
if parent == nil || parent.AuthorType != "agent" || !parent.AuthorID.Valid {
return commentAgentTrigger{}, false
}
agent, err := h.Queries.GetAgentInWorkspace(ctx, db.GetAgentInWorkspaceParams{
ID: parent.AuthorID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil || !agent.RuntimeID.Valid || agent.ArchivedAt.Valid {
return commentAgentTrigger{}, false
}
if !h.canInvokeAgent(ctx, agent, authorType, authorID, opts.OriginatorUserID, uuidToString(issue.WorkspaceID)) {
return commentAgentTrigger{}, false
}
hasPending, err := h.hasPendingTaskForIssueAndAgent(ctx, issue.ID, parent.AuthorID, opts)
if err != nil {
return commentAgentTrigger{}, false
}
trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceThreadParent, AlreadyPending: hasPending}
if squad, ok := h.squadLeaderRoleOfAuthoringTask(ctx, issue, *parent, agent); ok {
trigger.Squad = squad
}
return trigger, true
}
// squadLeaderRoleOfAuthoringTask reports the squad a reply should continue
// under, when the comment being replied to was itself written by a run of that
// squad's leader acting IN the leader role.
//
// Without it, replying directly to a leader's comment demoted the next run to
// a generic direct-agent task (MUL-4024's thread-parent gap): no squad
// briefing at claim time, `multica squad activity` rejected, and — on a
// project with a local_directory resource — the coordinator dragged into the
// user's own directory, queued behind its path mutex and stripped of the
// prior session it was still holding a workdir for (MUL-7006). Replying to a
// MEMBER comment in the same thread already restores the role this way
// (routeConversationOwnersForRoot); this closes the asymmetry.
//
// The role is read from the replied-to comment's OWN authoring run
// (source_task_id), not from the agent's latest task on the issue. An agent
// that is both leader and worker of the same squad posts in both roles, and a
// reply continues the role of the comment it answers — the latest-task read
// would hand a reply to a worker comment the leader's coordinator role.
//
// Fails closed to today's direct-agent routing on every uncertainty: a comment
// with no recorded authoring run (pre-migration-120 rows, or an agent write
// without the X-Task-ID header), a task that did not run as leader, a squad
// since deleted, or leadership that has moved to another agent. The claim
// handler downgrades the wire role again if the briefing cannot be injected,
// so an is_leader_task row whose squad went missing later still cannot deliver
// a leader run.
//
// Deleting a squad ARCHIVES it (DeleteSquad soft-archives after transferring
// its issues to the leader agent), so its rows and the leader's old comments
// both outlive the deletion. GetSquadInWorkspace does not filter archived_at —
// reviving a deleted squad from a stale comment would be a role this issue no
// longer has any assignment for — so the archived check belongs here.
func (h *Handler) squadLeaderRoleOfAuthoringTask(ctx context.Context, issue db.Issue, parent db.Comment, agent db.Agent) (*db.Squad, bool) {
if !parent.SourceTaskID.Valid {
return nil, false
}
task, err := h.Queries.GetAgentTask(ctx, parent.SourceTaskID)
if err != nil || !task.IsLeaderTask || !task.SquadID.Valid {
return nil, false
}
squad, err := h.Queries.GetSquadInWorkspace(ctx, db.GetSquadInWorkspaceParams{
ID: task.SquadID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil || squad.ArchivedAt.Valid || uuidToString(squad.LeaderID) != uuidToString(agent.ID) {
return nil, false
}
return &squad, true
}
// routeGuestSquadLeaderFallback closes the leader→worker→leader loop when the
// squad was @mentioned as a guest instead of assigned to the issue. The
// authoring worker task must cover the replied-to delegation comment; that
// comment's own task then proves the exact squad and current leader role. Once
// that historical leader role is identified, a now-invalid squad or leader
// fails closed instead of falling through to an unrelated assigned squad.
func (h *Handler) routeGuestSquadLeaderFallback(ctx context.Context, issue db.Issue, parent *db.Comment, authorID string, opts commentTriggerComputeOptions) ([]commentAgentTrigger, bool) {
if parent == nil || !parent.ID.Valid {
return nil, false
}
// A tombstone preserves only the reply tree, not the guest delegation's
// routing authority (#8323). The issue's assignment remains an independent
// source of authority for the assigned-squad fallback.
if parent.DeletedAt.Valid {
return nil, false
}
if !opts.AuthoringTaskID.Valid {
return nil, false
}
workerTask, err := h.Queries.GetAgentTask(ctx, opts.AuthoringTaskID)
if err != nil || workerTask.IsLeaderTask || !workerTask.AgentID.Valid ||
uuidToString(workerTask.AgentID) != authorID || !workerTask.IssueID.Valid ||
uuidToString(workerTask.IssueID) != uuidToString(issue.ID) || !taskCoversReplyParent(workerTask, parent.ID) {
return nil, false
}
if !parent.SourceTaskID.Valid {
return nil, false
}
leaderTask, err := h.Queries.GetAgentTask(ctx, parent.SourceTaskID)
if err != nil || !leaderTask.IsLeaderTask || !leaderTask.SquadID.Valid ||
!leaderTask.AgentID.Valid || leaderTask.AgentID != parent.AuthorID {
return nil, false
}
// Preserve the established assigned-squad path when this delegation came
// from that same squad. This fallback is only for a distinct guest squad.
if issue.AssigneeType.Valid && issue.AssigneeType.String == "squad" &&
issue.AssigneeID.Valid && issue.AssigneeID == leaderTask.SquadID {
return nil, false
}
trigger, ok := h.routeReplyToParentAuthor(ctx, issue, parent, "agent", authorID, opts)
if !ok || trigger.Squad == nil {
return nil, true
}
trigger.NonLeaderAgentReply = true
return []commentAgentTrigger{trigger}, true
}
func (h *Handler) routeThreadRootOwners(ctx context.Context, issue db.Issue, parent *db.Comment, memberID string, opts commentTriggerComputeOptions) ([]commentAgentTrigger, bool) {
if parent == nil || !parent.ID.Valid {
return nil, false
}
root, err := h.Queries.GetThreadRoot(ctx, db.GetThreadRootParams{
CommentID: parent.ID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil || !root.ID.Valid || root.AuthorType != "member" {
return nil, false
}
return h.routeConversationOwnersForRoot(ctx, issue, root, memberID, opts)
}
func (h *Handler) routeConversationOwnersForRoot(ctx context.Context, issue db.Issue, root db.Comment, memberID string, opts commentTriggerComputeOptions) ([]commentAgentTrigger, bool) {
if !root.ID.Valid || root.AuthorType != "member" {
return nil, false
}
if trigger, hasExplicitOwner, ok := h.routeFirstExplicitRootMentionOwner(ctx, issue, root, memberID, opts); hasExplicitOwner {
if !ok {
return nil, true
}
return []commentAgentTrigger{trigger}, true
}
if opts.ExcludeTriggerCommentID == root.ID {
return nil, false
}
owners, err := h.Queries.ListConversationRootOwners(ctx, db.ListConversationRootOwnersParams{
IssueID: issue.ID, TriggerCommentID: root.ID,
})
if err != nil {
return nil, false
}
if len(owners) == 0 {
return nil, false
}
triggers := make([]commentAgentTrigger, 0, len(owners))
for _, owner := range owners {
trigger, ok := h.routeConversationContinuationToAgent(ctx, issue, owner.AgentID, owner.SquadID, memberID, opts)
if ok {
triggers = append(triggers, trigger)
}
}
return triggers, true
}
func (h *Handler) routeFirstExplicitRootMentionOwner(ctx context.Context, issue db.Issue, root db.Comment, memberID string, opts commentTriggerComputeOptions) (commentAgentTrigger, bool, bool) {
for _, mention := range util.ParseMentions(root.Content) {
switch mention.Type {
case "agent":
agentID, err := util.ParseUUID(mention.ID)
if err != nil {
return commentAgentTrigger{}, true, false
}
trigger, ok := h.routeConversationContinuationToAgent(ctx, issue, agentID, pgtype.UUID{}, memberID, opts)
return trigger, true, ok
case "squad":
squadID, err := util.ParseUUID(mention.ID)
if err != nil {
return commentAgentTrigger{}, true, false
}
squad, err := h.Queries.GetSquadInWorkspace(ctx, db.GetSquadInWorkspaceParams{
ID: squadID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil {
return commentAgentTrigger{}, true, false
}
trigger, ok := h.routeConversationContinuationToAgent(ctx, issue, squad.LeaderID, squadID, memberID, opts)
return trigger, true, ok
}
}
return commentAgentTrigger{}, false, false
}
func (h *Handler) routeConversationContinuationToAgent(ctx context.Context, issue db.Issue, agentID, squadID pgtype.UUID, memberID string, opts commentTriggerComputeOptions) (commentAgentTrigger, bool) {
if !agentID.Valid {
return commentAgentTrigger{}, false
}
agent, err := h.Queries.GetAgentInWorkspace(ctx, db.GetAgentInWorkspaceParams{
ID: agentID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil || !agent.RuntimeID.Valid || agent.ArchivedAt.Valid {
return commentAgentTrigger{}, false
}
if !h.canInvokeAgent(ctx, agent, "member", memberID, memberID, uuidToString(issue.WorkspaceID)) {
return commentAgentTrigger{}, false
}
hasPending, err := h.hasPendingTaskForIssueAndAgent(ctx, issue.ID, agentID, opts)
if err != nil {
return commentAgentTrigger{}, false
}
trigger := commentAgentTrigger{Agent: agent, Source: commentTriggerSourceConversation, AlreadyPending: hasPending}
if squadID.Valid {
if squad, err := h.Queries.GetSquadInWorkspace(ctx, db.GetSquadInWorkspaceParams{
ID: squadID,
WorkspaceID: issue.WorkspaceID,
}); err == nil {
trigger.Squad = &squad
}
}
return trigger, true
}
func (h *Handler) routeAssigneeFallback(ctx context.Context, issue db.Issue, authorType, authorID string, opts commentTriggerComputeOptions) (commentAgentTrigger, bool) {
if !issue.AssigneeType.Valid || !issue.AssigneeID.Valid {
return commentAgentTrigger{}, false
}
// In Triage the assignee is a proposal the triager is still making, not an
// owner, so there is nobody to fall back TO (MUL-7189 §2.3). This route and
// the squad-leader one below are the two that derive an executor from the
// issue; a comment naming an agent by hand is routed above and still runs.
//
// The queue door refuses these anyway. Stopping here is what keeps the
// server from attempting an enqueue it already knows will be refused, and
// logging a failure for every comment on a Triage entry.
if issue.TriageState.Valid {
return commentAgentTrigger{}, false
}
switch issue.AssigneeType.String {
case "agent":
agent, hasPending, ok := h.assigneeFallbackAgent(ctx, issue, authorType, authorID, opts)
if !ok {
return commentAgentTrigger{}, false
}
return commentAgentTrigger{Agent: agent, Source: commentTriggerSourceIssueAssignee, AlreadyPending: hasPending}, true
case "squad":
return h.routeAssignedSquadLeaderFallback(ctx, issue, authorType, authorID, opts)
default:
return commentAgentTrigger{}, false
}
}
func (h *Handler) routeAssignedSquadLeaderFallback(ctx context.Context, issue db.Issue, authorType, authorID string, opts commentTriggerComputeOptions) (commentAgentTrigger, bool) {
// Checked here as well as in routeAssigneeFallback: an agent-authored
// comment reaches this one directly, without passing through that caller.
if issue.TriageState.Valid {
return commentAgentTrigger{}, false
}
squad, err := h.Queries.GetSquadInWorkspace(ctx, db.GetSquadInWorkspaceParams{
ID: issue.AssigneeID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil {
return commentAgentTrigger{}, false
}
if authorType == "agent" && authorID == uuidToString(squad.LeaderID) &&
h.shouldSuppressSquadLeaderSelfTrigger(ctx, issue.ID, squad.LeaderID, squad.ID) {
return commentAgentTrigger{}, false
}
agent, err := h.Queries.GetAgentInWorkspace(ctx, db.GetAgentInWorkspaceParams{
ID: squad.LeaderID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil || !agent.RuntimeID.Valid || agent.ArchivedAt.Valid {
return commentAgentTrigger{}, false
}
if !h.canInvokeAgent(ctx, agent, authorType, authorID, opts.OriginatorUserID, uuidToString(issue.WorkspaceID)) {
return commentAgentTrigger{}, false
}
hasPending, err := h.hasPendingTaskForIssueAndAgent(ctx, issue.ID, squad.LeaderID, opts)
if err != nil {
return commentAgentTrigger{}, false
}
return commentAgentTrigger{Agent: agent, Source: commentTriggerSourceIssueAssignee, Squad: &squad, AlreadyPending: hasPending}, true
}
func (h *Handler) hasPendingTaskForIssueAndAgent(ctx context.Context, issueID, agentID pgtype.UUID, opts commentTriggerComputeOptions) (bool, error) {
if !opts.ThreadCommentID.Valid {
return false, nil
}
// Key dedup on the reviewed head so re-pushing to the PR mid-review
// invalidates dedup and a fresh run enqueues against the new HEAD (TEN-356).
headSha := h.TaskService.ResolveIssueReviewSHAParam(ctx, issueID)
if opts.ExcludeTriggerCommentID.Valid {
return h.Queries.HasPendingTaskForIssueAndAgentExcludingTriggerCommentInThread(ctx, db.HasPendingTaskForIssueAndAgentExcludingTriggerCommentInThreadParams{
ThreadCommentID: opts.ThreadCommentID,
IssueID: issueID,
AgentID: agentID,
ExcludeTriggerCommentID: opts.ExcludeTriggerCommentID,
HeadSha: headSha,
})
}
return h.Queries.HasPendingTaskForIssueAndAgentInThread(ctx, db.HasPendingTaskForIssueAndAgentInThreadParams{
ThreadCommentID: opts.ThreadCommentID,
IssueID: issueID,
AgentID: agentID,
HeadSha: headSha,
})
}
// resolveMentionedAgentCommentTriggers parses explicit @agent and @squad
// mentions from the current comment and returns the runnable agent recipients.
// Skips agents with on_mention trigger disabled, and private agents mentioned
// by non-owner members (only the agent owner or workspace admin/owner can
// mention a private agent). Self-mentions are intentionally allowed so an
// agent running in one issue can explicitly enqueue itself on another (e.g.
// a child-issue run notifying the parent issue whose assignee is the same
// agent); runaway loops are prevented by HasPendingTaskForIssueAndAgent
// dedupe and the natural queued/dispatched coalescing of the task queue.
// Note: no issue status gate here — @mention is an explicit action and should
// work even on done/cancelled issues (the agent can reopen the issue if needed).
// commentMentionTarget is one EXPLICIT @agent / @squad mention and how it
// resolved (MUL-4525 §2). This is tracked separately from the execution
// triggers: several mentions can resolve to the same executing agent (e.g.
// @Agent A and @Squad S whose leader is A), and each still needs its own
// outcome even though the run is coalesced. Exactly one of the resolution
// fields is set:
// - ExecAgentID non-empty → the mention runs via that executing agent; its
// outcome mirrors the agent's enqueue status (queued/coalesced/deferred).
// - Status set (with ReasonCode) → a terminal, non-executing outcome
// (blocked, or an A2A self-suppressed squad leader that is deferred).
type commentMentionTarget struct {
TargetType string // "agent" | "squad"
TargetID string
ExecAgentID string
Status DispatchStatus
ReasonCode DispatchReasonCode
// unusable carries the refused agent and its verdict for the reasons that
// need a durable trace (runtime_unusable or runtime_access_denied). Internal
// to the handler:
// the resolver runs for the composer PREVIEW as well, so it only records
// what happened — writing the notice is the trigger path's job.
unusable *blockedRuntimeNotice
}
// blockedRuntimeNotice is a refusal worth leaving on the issue: which agent, and
// the verdict that names the repair.
type blockedRuntimeNotice struct {
agent db.Agent
verdict service.AgentVerdict
}
func (h *Handler) resolveMentionedAgentCommentTriggers(ctx context.Context, issue db.Issue, mentions []util.Mention, authorType, authorID string, opts commentTriggerComputeOptions) ([]commentAgentTrigger, []commentMentionTarget) {
wsID := uuidToString(issue.WorkspaceID)
triggers := make([]commentAgentTrigger, 0, len(mentions))
// seen dedups EXECUTION by resolved agent id: two mentions resolving to the
// same agent enqueue only one task. Mapping to the trigger's index lets a
// squad-leader mention UPGRADE an already-added plain @agent trigger for the
// same agent — the leader task is a strict superset (it sets is_leader_task
// and squad_id so the daemon injects the squad briefing), so the merged run
// must carry that role regardless of mention order.
seen := make(map[string]int, len(mentions))
add := func(trigger commentAgentTrigger) {
id := uuidToString(trigger.Agent.ID)
if idx, ok := seen[id]; ok {
if triggers[idx].Source != commentTriggerSourceMentionSquadLeader &&
trigger.Source == commentTriggerSourceMentionSquadLeader {
triggers[idx] = trigger
}
return
}
seen[id] = len(triggers)
triggers = append(triggers, trigger)
}
// targets record one outcome per EXPLICIT mention — deduped by the target
// the user named (type:id), NOT by executing agent — so no explicitly-named
// target is silently dropped even when its run coalesces with another's.
var targets []commentMentionTarget
targetSeen := make(map[string]struct{}, len(mentions))
addTarget := func(t commentMentionTarget) {
key := t.TargetType + ":" + t.TargetID
if _, ok := targetSeen[key]; ok {
return
}
targetSeen[key] = struct{}{}
targets = append(targets, t)
}
// blockTarget records a mention that will not fire. The invoke gate is
// evaluated BEFORE any archived/runtime state is read, so a caller who
// cannot invoke a private target learns only the generic
// invocation_not_allowed and can never enumerate the target's existence or
// state from the reason code.
blockTarget := func(targetType, targetID string, reason DispatchReasonCode) {
addTarget(commentMentionTarget{TargetType: targetType, TargetID: targetID, Status: DispatchBlocked, ReasonCode: reason})
}
// blockUnusableTarget is blockTarget for the verdicts that also need a
// durable trace. Every author gets it, including a human: the chip and toast
// carry the reason code but not the repair command, and an agent-authored
// mention has nobody watching a response at all.
blockUnusableTarget := func(targetType, targetID string, agent db.Agent, verdict service.AgentVerdict) {
notice := &blockedRuntimeNotice{agent: agent, verdict: verdict}
if !service.RuntimeBlockedNeedsNotice(verdict.Reason) {
notice = nil
}
addTarget(commentMentionTarget{
TargetType: targetType, TargetID: targetID,
Status: DispatchBlocked, ReasonCode: verdict.Reason, unusable: notice,
})
}
for _, m := range mentions {
if m.Type == "squad" {
// @squad mention → trigger the squad's leader agent.
// The mention id comes from untrusted comment text and MentionRe
// accepts any hex/dash run (`mention://squad/-`), so it is parsed
// with the error-returning ParseUUID — the Must* variant would
// panic the request, and on the create path the comment row is
// already committed by then. A malformed id is indistinguishable
// from a well-formed id that owns no squad: same target_unavailable
// outcome, no run.
squadUUID, err := util.ParseUUID(m.ID)
if err != nil {
blockTarget("squad", m.ID, ReasonTargetUnavailable)
continue
}
squad, err := h.Queries.GetSquadInWorkspace(ctx, db.GetSquadInWorkspaceParams{
ID: squadUUID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil {
blockTarget("squad", m.ID, ReasonTargetUnavailable)
continue
}
leaderID := squad.LeaderID
// A2A self-suppression: the author IS this squad's leader and its
// most recent task on this issue was a leader/generic role (NOT a
// fresh same-squad worker→leader handoff), so we do not re-fire the
// leader from its own @mention. The outcome must reflect reality, not
// assume success (MUL-4525): `deferred` only when a real non-terminal
// task is still active (its reconcile covers this comment); a query
// failure is a non-success internal_error, never a fabricated
// deferred; otherwise nothing runs → self_trigger_suppressed.
if authorType == "agent" && authorID == uuidToString(leaderID) &&
h.shouldSuppressSquadLeaderSelfTrigger(ctx, issue.ID, leaderID, squad.ID) {
active, activeErr := h.hasActiveTaskForIssueAndAgent(ctx, issue.ID, leaderID, opts.ThreadCommentID)
status, reason := decideSuppressedLeaderOutcome(active, activeErr)
addTarget(commentMentionTarget{TargetType: "squad", TargetID: m.ID, Status: status, ReasonCode: reason})
continue
}
agent, err := h.Queries.GetAgentInWorkspace(ctx, db.GetAgentInWorkspaceParams{
ID: leaderID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil {
blockTarget("squad", m.ID, ReasonTargetUnavailable)
continue
}
// Private-leader gate first (enumeration-safe: a caller who cannot
// invoke the leader never learns its archived/runtime state).
if !h.canInvokeAgent(ctx, agent, authorType, authorID, opts.OriginatorUserID, wsID) {
blockTarget("squad", m.ID, ReasonInvocationNotAllowed)
continue
}
if agent.ArchivedAt.Valid {
blockTarget("squad", m.ID, ReasonTargetUnavailable)
continue
}
// Same shared verdict as the direct-agent branch below.
if verdict, err := service.AgentReadiness(ctx, h.runtimeLookup(obsmetrics.RuntimeLookupSourceComment), agent); err == nil && verdict.Blocked() {
blockUnusableTarget("squad", m.ID, agent, verdict)
continue
}
hasPending, err := h.hasPendingTaskForIssueAndAgent(ctx, issue.ID, leaderID, opts)
if err != nil {
blockTarget("squad", m.ID, ReasonInternalError)
continue
}
add(commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionSquadLeader, Squad: &squad, AlreadyPending: hasPending})
addTarget(commentMentionTarget{TargetType: "squad", TargetID: m.ID, ExecAgentID: uuidToString(leaderID)})
continue
}
if m.Type != "agent" {
continue
}
agentUUID, err := util.ParseUUID(m.ID)
if err != nil {
// Untrusted comment text: MentionRe matches any hex/dash run, so a
// malformed id must not reach the panicking Must* parser. A string
// that is not a UUID at all cannot name an entity in ANY workspace,
// so there is no existence to conceal here and the invoke-permission
// code would be a false cause (MUL-5548): report the same
// target_unavailable the squad path above already uses. Only the
// well-formed-but-unresolved case below stays enumeration-safe.
blockTarget("agent", m.ID, ReasonTargetUnavailable)
continue
}
// Load the agent scoped to the current issue's workspace. Using the
// bare GetAgent here would let a mention resolve to an agent in a
// different workspace, and the visibility check below would then be
// applied against the wrong workspace's roles (a workspace owner in
// THIS workspace would pass the gate for a private agent that lives
// in someone else's workspace).
agent, err := h.Queries.GetAgentInWorkspace(ctx, db.GetAgentInWorkspaceParams{
ID: agentUUID,
WorkspaceID: issue.WorkspaceID,
})
if err != nil {
// Do not reveal whether the id exists (it may be a private agent in
// another workspace): the generic invocation_not_allowed is returned.
blockTarget("agent", m.ID, ReasonInvocationNotAllowed)
continue
}
// Private-agent gate first, before any archived/runtime state is read.
if !h.canInvokeAgent(ctx, agent, authorType, authorID, opts.OriginatorUserID, wsID) {
blockTarget("agent", m.ID, ReasonInvocationNotAllowed)
continue
}
if agent.ArchivedAt.Valid {
blockTarget("agent", m.ID, ReasonTargetUnavailable)
continue
}
// One readiness verdict for every admission path (service.AgentReadiness).
// Only a BLOCKED verdict refuses the mention: an unbound agent has no
// machine to bring back (MUL-5559), and a machine whose CLI cannot run
// will not start doing so on its own (MUL-6164). A merely offline
// runtime keeps queueing — that wait ends by itself when the machine
// returns, and taking it away would remove a behaviour people rely on.
if verdict, err := service.AgentReadiness(ctx, h.runtimeLookup(obsmetrics.RuntimeLookupSourceComment), agent); err == nil && verdict.Blocked() {
blockUnusableTarget("agent", m.ID, agent, verdict)
continue
}
hasPending, err := h.hasPendingTaskForIssueAndAgent(ctx, issue.ID, agentUUID, opts)
if err != nil {
blockTarget("agent", m.ID, ReasonInternalError)
continue
}
add(commentAgentTrigger{Agent: agent, Source: commentTriggerSourceMentionAgent, AlreadyPending: hasPending})
addTarget(commentMentionTarget{TargetType: "agent", TargetID: m.ID, ExecAgentID: uuidToString(agentUUID)})
}
return triggers, targets
}
func (h *Handler) UpdateComment(w http.ResponseWriter, r *http.Request) {
r = h.withWakeupActor(r)
commentId := chi.URLParam(r, "commentId")
userID, ok := requireUserID(w, r)
if !ok {
return
}
commentUUID, ok := parseUUIDOrBadRequest(w, commentId, "comment id")
if !ok {
return
}
// Load comment scoped to current workspace.
workspaceID := h.resolveWorkspaceID(r)
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
if !ok {
return
}
existing, err := h.Queries.GetCommentInWorkspace(r.Context(), db.GetCommentInWorkspaceParams{
ID: commentUUID,
WorkspaceID: wsUUID,
})
// A deleted comment's tombstone has nothing left to edit.
if err != nil || existing.DeletedAt.Valid {
writeError(w, http.StatusNotFound, "comment not found")
return
}
if bound, err := h.Queries.CommentHasTaskSupplement(r.Context(), db.CommentHasTaskSupplementParams{
CommentID: existing.ID, WorkspaceID: wsUUID,
}); err != nil {
writeError(w, http.StatusInternalServerError, "failed to verify additional message")
return
} else if bound {
writeError(w, http.StatusConflict, "additional messages cannot be edited")
return
}
member, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
actorType, actorID := h.resolveActor(r, userID, workspaceID)
isAuthor := existing.AuthorType == actorType && uuidToString(existing.AuthorID) == actorID
isAdmin := roleAllowed(member.Role, "owner", "admin")
if !isAuthor && !isAdmin {
writeError(w, http.StatusForbidden, "only comment author or admin can edit")
return
}
var req struct {
Content string `json:"content"`
ContentBase *string `json:"content_base,omitempty"`
AttachmentIDs *[]string `json:"attachment_ids"`
SuppressAgentIDs []string `json:"suppress_agent_ids"`
ExpectedRevision *int64 `json:"expected_revision,omitempty"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
// See CreateComment: strip NUL / invalid-UTF-8 bytes PostgreSQL's TEXT column
// rejects before the empty check, so an edit that introduces such a byte
// can't 500 (GH #5388).
req.Content = sanitizeNullBytes(req.Content)
if req.ContentBase != nil {
sanitized := sanitizeNullBytes(*req.ContentBase)
req.ContentBase = &sanitized
}
if req.Content == "" {
writeError(w, http.StatusBadRequest, "content is required")
return
}
if req.ExpectedRevision != nil {
if *req.ExpectedRevision < 1 {
writeError(w, http.StatusBadRequest, "expected_revision must be a positive integer")
return
}
if existing.Revision != *req.ExpectedRevision {
writeRevisionConflict(w, "comment", existing.ID, *req.ExpectedRevision, existing.Revision)
return
}
}
strictContentEdit := req.ContentBase != nil || req.ExpectedRevision != nil
var attachmentIDs []pgtype.UUID
replaceAttachments := req.AttachmentIDs != nil
if replaceAttachments {
var ok bool
attachmentIDs, ok = parseUUIDSliceOrBadRequest(w, *req.AttachmentIDs, "attachment_ids")
if !ok {
return
}
}
suppressAgentIDs, ok := parseUUIDSliceOrBadRequest(w, req.SuppressAgentIDs, "suppress_agent_ids")
if !ok {
return
}
// NOTE: See CreateComment — Markdown is sanitized at render/edit time, not here.
oldContent := existing.Content
// Preserve the existing authority lineage by default — this path is taken only
// for an UNCHANGED edit (no re-trigger). When the content changes below, the
// lineage is re-derived from the EDIT action itself (MUL-4857), never carried
// over from the comment's original authoring run.
sourceTaskID := existing.SourceTaskID
var triggerIssue *db.Issue
var cancelled []db.AgentTaskQueue
if oldContent != req.Content {
issue, err := h.Queries.GetIssue(r.Context(), existing.IssueID)
if err != nil {
slog.Warn("load issue for edit post-processing failed", "issue_id", uuidToString(existing.IssueID), "error", err)
writeError(w, http.StatusInternalServerError, "failed to load issue")
return
}
triggerIssue = &issue
// A content edit is a NEW action, so its delegation lineage must key on THIS
// edit. Only the AGENT author re-editing its OWN comment carries the lineage
// forward (commentSourceTaskID re-stamps the current editing task) — so
// preview, save, and the deferred completion-reconcile all resolve the
// authority from this one action. Any OTHER editor — a workspace owner/admin
// editing an AGENT's comment, or a member editing their own — CLEARS the
// lineage so the deferred reconcile fails closed instead of resurrecting the
// original autopilot run's creator authority. An admin holds manage rights
// over the comment, not invoke rights over the author's private agents (Elon
// must-fix, round 3).
if actorType == "agent" && isAuthor {
sourceTaskID = h.commentSourceTaskID(r)
} else {
sourceTaskID = pgtype.UUID{}
}
// Legacy clients keep the existing cancel-before-update behavior. Strict
// revision writes defer cancellation until the conditional UPDATE wins,
// so a race that returns 409 cannot mutate the task queue.
if !strictContentEdit {
cancelled, err = h.TaskService.CancelTasksByTriggerComment(r.Context(), existing.ID)
if err != nil {
slog.Warn("cancel tasks for edited comment failed", "comment_id", uuidToString(existing.ID), "error", err)
writeError(w, http.StatusInternalServerError, "failed to prepare comment edit")
return
}
}
}
updateParams := db.UpdateCommentParams{
ID: commentUUID,
Content: req.Content,
SourceTaskID: sourceTaskID,
// New text replaces the "don't start" choices completion replay keeps.
SuppressedAgentIds: suppressAgentIDs,
}
if req.ContentBase != nil {
updateParams.ContentBase = pgtype.Text{String: *req.ContentBase, Valid: true}
}
if req.ExpectedRevision != nil {
updateParams.ExpectedRevision = pgtype.Int8{Int64: *req.ExpectedRevision, Valid: true}
}
var comment db.Comment
var issueRevision int64
transactionalEdit := replaceAttachments || (oldContent != req.Content && strictContentEdit)
if transactionalEdit {
// Strict body edits, attachment-set edits, and cancellation of tasks built
// from the old body are one database outcome. UpdateComment takes the row
// lock before the attachment replacement, so two modern editors cannot
// interleave their CAS check and attachment selection. A body + attachment
// edit is one visible mutation and therefore bumps revision exactly once.
tx, beginErr := h.beginWakeupWrite(r.Context())
if beginErr != nil {
writeError(w, http.StatusInternalServerError, "failed to prepare comment edit")
return
}
defer tx.Rollback(r.Context())
qtx := h.Queries.WithTx(tx)
var updated db.UpdateCommentRow
updated, err = qtx.UpdateComment(r.Context(), updateParams)
if err == nil {
comment = updated.Comment()
issueRevision = updated.IssueRevision
}
if err == nil && oldContent != req.Content && strictContentEdit {
cancelled, err = qtx.CancelAgentTasksByTriggerComment(r.Context(), existing.ID)
if err == nil {
err = service.SettleTerminalTaskState(r.Context(), qtx, cancelled...)
}
}
if err == nil && replaceAttachments {
var changed int64
changed, err = qtx.ReplaceCommentAttachments(r.Context(), db.ReplaceCommentAttachmentsParams{
CommentID: comment.ID,
IssueID: existing.IssueID,
AttachmentIds: attachmentIDs,
})
if err == nil && changed > 0 && oldContent == req.Content {
comment, err = qtx.BumpCommentRevision(r.Context(), db.BumpCommentRevisionParams{
ID: comment.ID,
WorkspaceID: existing.WorkspaceID,
})
}
}
if err == nil {
err = tx.Commit(r.Context())
}
if err == nil {
h.TaskService.BroadcastCancelledTasks(r.Context(), uuidToString(existing.WorkspaceID), cancelled)
}
} else {
var updated db.UpdateCommentRow
updated, err = wakeupWrite(h, r, func(q *db.Queries) (db.UpdateCommentRow, error) {
return q.UpdateComment(r.Context(), updateParams)
})
if err == nil {
comment = updated.Comment()
issueRevision = updated.IssueRevision
}
}
if err != nil {
slog.Warn("update comment failed", append(logger.RequestAttrs(r), "error", err, "comment_id", commentId)...)
if triggerIssue != nil && !strictContentEdit {
// Cancellation committed but the edit did not. Restore the complete
// original batch, including the still-valid unchanged comment.
h.retriggerCancelledTaskSurvivors(r.Context(), *triggerIssue, cancelled, pgtype.UUID{})
}
if errors.Is(err, pgx.ErrNoRows) {
current, reloadErr := h.Queries.GetCommentInWorkspace(r.Context(), db.GetCommentInWorkspaceParams{ID: commentUUID, WorkspaceID: wsUUID})
if errors.Is(reloadErr, pgx.ErrNoRows) || (reloadErr == nil && current.DeletedAt.Valid) {
// Deleted while the edit waited for it.
writeError(w, http.StatusNotFound, "comment not found")
return
}
if reloadErr == nil && strictContentEdit {
if req.ExpectedRevision != nil {
writeRevisionConflict(w, "comment", current.ID, *req.ExpectedRevision, current.Revision)
} else {
writeEditConflict(w, "comment", current.ID)
}
return
}
}
writeError(w, http.StatusInternalServerError, "failed to update comment")
return
}
retriggerEditedComment := func() []CommentTriggerOutcome {
if oldContent == comment.Content {
return nil
}
issue := *triggerIssue
var parentComment *db.Comment
if existing.ParentID.Valid {
parent, err := h.Queries.GetComment(r.Context(), existing.ParentID)
if err == nil {
parentComment = &parent
}
}
h.retriggerCancelledTaskSurvivors(r.Context(), issue, cancelled, existing.ID)
return h.triggerTasksForComment(r.Context(), issue, comment, parentComment, actorType, actorID, h.invokeOriginatorFromRequest(r, actorType, actorID), suppressAgentIDs, commentSteerRequest{})
}
// Fetch reactions and attachments for the updated comment.
grouped := h.groupReactions(r, []pgtype.UUID{comment.ID})
groupedAtt := h.groupAttachments(r, []pgtype.UUID{comment.ID})
cid := uuidToString(comment.ID)
resp := commentToResponse(comment, grouped[cid], groupedAtt[cid])
resp.IssueRevision = issueRevision
slog.Info("comment updated", append(logger.RequestAttrs(r), "comment_id", commentId)...)
eventPayload := map[string]any{"comment": resp}
if issueRevision > 0 {
eventPayload["issue_revision"] = issueRevision
}
h.publish(protocol.EventCommentUpdated, workspaceID, actorType, actorID, eventPayload)
// The broadcast above intentionally omits trigger_outcomes — it is the
// editor's private feedback, not shared timeline state (MUL-4525 §2).
resp.TriggerOutcomes = retriggerEditedComment()
writeJSON(w, http.StatusOK, resp)
}
func (h *Handler) DeleteComment(w http.ResponseWriter, r *http.Request) {
r = h.withWakeupActor(r)
commentId := chi.URLParam(r, "commentId")
userID, ok := requireUserID(w, r)
if !ok {
return
}
commentUUID, ok := parseUUIDOrBadRequest(w, commentId, "comment id")
if !ok {
return
}
// Load comment scoped to current workspace.
workspaceID := h.resolveWorkspaceID(r)
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
if !ok {
return
}
comment, err := h.Queries.GetCommentInWorkspace(r.Context(), db.GetCommentInWorkspaceParams{
ID: commentUUID,
WorkspaceID: wsUUID,
})
if err != nil {
writeError(w, http.StatusNotFound, "comment not found")
return
}
member, ok := h.workspaceMember(w, r, workspaceID)
if !ok {
return
}
actorType, actorID := h.resolveActor(r, userID, workspaceID)
isAuthor := comment.AuthorType == actorType && uuidToString(comment.AuthorID) == actorID
isAdmin := roleAllowed(member.Role, "owner", "admin")
if !isAuthor && !isAdmin {
writeError(w, http.StatusForbidden, "only comment author or admin can delete")
return
}
issue, err := h.Queries.GetIssue(r.Context(), comment.IssueID)
hasIssue := err == nil
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
slog.Warn("load issue for delete post-processing failed", "issue_id", uuidToString(comment.IssueID), "error", err)
writeError(w, http.StatusInternalServerError, "failed to load issue")
return
}
if errors.Is(err, pgx.ErrNoRows) {
// The issue delete may have won the race after GetCommentInWorkspace.
// Continue so DeleteComment can report whether the comment itself still
// existed; touching issue activity is intentionally best effort.
slog.Info("comment parent issue no longer exists", "issue_id", uuidToString(comment.IssueID), "comment_id", commentId)
}
if comment.DeletedAt.Valid {
writeError(w, http.StatusNotFound, "comment not found")
return
}
// Cancel any active task whose planned batch contains this comment so the
// agent does not run with the now-deleted content already embedded. Must
// run before the delete: a removed row would nullify trigger_comment_id
// through ON DELETE SET NULL and orphan those tasks in queued.
cancelled, cancelErr := h.TaskService.CancelTasksByTriggerComment(r.Context(), comment.ID)
if cancelErr != nil {
slog.Warn("cancel tasks for deleted trigger comment failed", append(logger.RequestAttrs(r), "error", cancelErr, "comment_id", commentId)...)
}
deleted, err := h.deleteComment(r.Context(), comment.ID, comment.WorkspaceID)
if err != nil {
slog.Warn("delete comment failed", append(logger.RequestAttrs(r), "error", err, "comment_id", commentId)...)
// Cancellation already committed but deletion did not. If the parent
// issue still exists, rebuild the complete cancelled batch (including
// this trigger) before reporting the storage error or concurrent no-op.
if hasIssue {
h.retriggerCancelledTaskSurvivors(r.Context(), issue, cancelled, pgtype.UUID{})
}
if errors.Is(err, errTaskSupplementInFlight) {
writeError(w, http.StatusConflict, "additional message is still being delivered")
} else if errors.Is(err, pgx.ErrNoRows) {
writeError(w, http.StatusNotFound, "comment not found")
} else {
writeError(w, http.StatusInternalServerError, "failed to delete comment")
}
return
}
h.deleteS3Objects(r.Context(), deleted.AttachmentURLs)
slog.Info("comment deleted", append(logger.RequestAttrs(r),
"comment_id", commentId,
"issue_id", uuidToString(comment.IssueID),
"tombstoned", deleted.Tombstone != nil,
"removed_count", len(deleted.RemovedIDs),
)...)
// A tombstone is still part of the thread, so clients replace it in place
// through the ordinary update event; removed rows get one delete event each.
if deleted.Tombstone != nil {
resp := commentToResponse(*deleted.Tombstone, nil, nil)
resp.IssueRevision = deleted.IssueRevision
eventPayload := map[string]any{"comment": resp}
if deleted.IssueRevision > 0 {
eventPayload["issue_revision"] = deleted.IssueRevision
}
h.publish(protocol.EventCommentUpdated, workspaceID, actorType, actorID, eventPayload)
}
for _, removedID := range deleted.RemovedIDs {
eventPayload := map[string]any{
"comment_id": uuidToString(removedID),
"issue_id": uuidToString(comment.IssueID),
}
if deleted.IssueRevision > 0 {
eventPayload["issue_revision"] = deleted.IssueRevision
}
h.publish(protocol.EventCommentDeleted, workspaceID, actorType, actorID, eventPayload)
}
if hasIssue {
h.retriggerCancelledTaskSurvivors(r.Context(), issue, cancelled, comment.ID)
}
w.WriteHeader(http.StatusNoContent)
}
// commentDeletion is the committed outcome of deleteComment.
type commentDeletion struct {
// Tombstone is the cleared row when the comment still had replies.
Tombstone *db.Comment
// RemovedIDs lists every row removed outright: the comment itself when it
// had no replies, followed by each tombstone ancestor that lost its last
// reply as a result, nearest first.
RemovedIDs []pgtype.UUID
// AttachmentURLs are the deleted comment's storage objects, for cleanup
// after commit.
AttachmentURLs []string
IssueRevision int64
}
// lockCommentAttachments locks the unbound issue attachments a comment is
// being created with and reports the first requested id that is not among
// them: never eligible, or deleted while the lock waited. The caller must
// already hold the issue's row lock — these are its children, and taking them
// first is the inversion CreateComment documents.
func lockCommentAttachments(ctx context.Context, qtx *db.Queries, workspaceID, issueID pgtype.UUID, ids []pgtype.UUID) (missing pgtype.UUID, err error) {
locked, err := qtx.LockAttachmentsForCommentLink(ctx, db.LockAttachmentsForCommentLinkParams{
WorkspaceID: workspaceID,
IssueID: issueID,
AttachmentIds: ids,
})
if err != nil {
return pgtype.UUID{}, err
}
have := make(map[pgtype.UUID]struct{}, len(locked))
for _, id := range locked {
have[id] = struct{}{}
}
for _, id := range ids {
if _, ok := have[id]; !ok {
return id, nil
}
}
return pgtype.UUID{}, nil
}
// withLiveCommentLock runs write in a transaction that first locks the
// comment's issue and then the comment — the order every comment mutation
// takes — and only while the comment is live. Writes to a comment's children
// (reactions, attachments) go through it so none can land on a tombstone, even
// when the delete commits while this waits. Returns pgx.ErrNoRows when the
// comment is gone or deleted.
func (h *Handler) withLiveCommentLock(ctx context.Context, commentID, workspaceID pgtype.UUID, write func(*db.Queries) error) error {
tx, err := h.beginWakeupWrite(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
qtx := h.Queries.WithTx(tx)
if _, err := qtx.LockLiveComment(ctx, db.LockLiveCommentParams{ID: commentID, WorkspaceID: workspaceID}); err != nil {
return err
}
if err := write(qtx); err != nil {
return err
}
return tx.Commit(ctx)
}
// commentTombstonePruneDepth bounds the upward walk over tombstone ancestors,
// matching the depth bound of the ancestor path queries.
const commentTombstonePruneDepth = 256
var errTaskSupplementInFlight = errors.New("additional message is still being delivered")
// deleteComment deletes exactly one comment (#8296). A comment that still has
// replies becomes a tombstone — content, attachments, reactions and resolution
// cleared, row kept — so each reply stays attached to its direct parent. A
// comment without replies is removed, along with every tombstone ancestor
// that is left without replies. Nothing here relies on the parent_id cascade.
//
// Returns pgx.ErrNoRows when the comment is already gone or already deleted.
func (h *Handler) deleteComment(ctx context.Context, commentID, workspaceID pgtype.UUID) (commentDeletion, error) {
var out commentDeletion
tx, err := h.beginWakeupWrite(ctx)
if err != nil {
return out, err
}
defer tx.Rollback(ctx)
qtx := h.Queries.WithTx(tx)
target, err := qtx.LockCommentForDelete(ctx, db.LockCommentForDeleteParams{
ID: commentID,
WorkspaceID: workspaceID,
})
if err != nil {
return out, err
}
// Keep receipt admission locked through deletion so a concurrent retry
// cannot resurrect or deliver the content being removed.
receipts, receiptErr := qtx.LockTaskSupplementsByComment(ctx, db.LockTaskSupplementsByCommentParams{
CommentID: commentID, WorkspaceID: workspaceID,
})
if receiptErr != nil {
return out, receiptErr
}
for _, receipt := range receipts {
if receipt.Status == "pending" || receipt.Status == "delivering" {
return out, errTaskSupplementInFlight
}
}
if len(receipts) > 0 {
if err := qtx.DeleteTaskSupplementByComment(ctx, db.DeleteTaskSupplementByCommentParams{
CommentID: commentID, WorkspaceID: workspaceID,
}); err != nil {
return out, err
}
}
// Separate statement on purpose: its snapshot postdates the locks above,
// so it sees every committed reply, and none can be added while they are
// held.
hasReplies, err := qtx.CommentHasReplies(ctx, db.CommentHasRepliesParams{
ID: target.ID,
WorkspaceID: target.WorkspaceID,
})
if err != nil {
return out, err
}
out.AttachmentURLs, err = qtx.DeleteCommentAttachments(ctx, db.DeleteCommentAttachmentsParams{
CommentID: target.ID,
WorkspaceID: target.WorkspaceID,
})
if err != nil {
return out, err
}
if err := qtx.DeleteCommentReactions(ctx, db.DeleteCommentReactionsParams{
CommentID: target.ID,
WorkspaceID: target.WorkspaceID,
}); err != nil {
return out, err
}
if hasReplies {
tombstone, err := qtx.TombstoneComment(ctx, db.TombstoneCommentParams{
ID: target.ID,
WorkspaceID: target.WorkspaceID,
})
if err != nil {
return out, err
}
// A deleted delegated-failure recovery signal is withdrawn: settle it so
// the recovery sweeper stops scanning its row. A no-op for every other
// comment.
if _, err := qtx.SettleDelegatedFailureRecoveryComment(ctx, target.ID); err != nil {
return out, err
}
out.Tombstone = &tombstone
} else {
removed, err := qtx.DeleteLeafComment(ctx, db.DeleteLeafCommentParams{
ID: target.ID,
WorkspaceID: target.WorkspaceID,
})
if err != nil {
return out, err
}
out.RemovedIDs = append(out.RemovedIDs, removed.ID)
parentID := removed.ParentID
for depth := 0; parentID.Valid && depth < commentTombstonePruneDepth; depth++ {
pruned, err := qtx.DeleteReplylessCommentTombstone(ctx, db.DeleteReplylessCommentTombstoneParams{
ID: parentID,
WorkspaceID: target.WorkspaceID,
})
if errors.Is(err, pgx.ErrNoRows) {
// A live comment, or a tombstone other replies still hold.
break
}
if err != nil {
return out, err
}
out.RemovedIDs = append(out.RemovedIDs, pruned.ID)
parentID = pruned.ParentID
}
}
out.IssueRevision, err = qtx.TouchIssueForCommentDelete(ctx, db.TouchIssueForCommentDeleteParams{
IssueID: target.IssueID,
WorkspaceID: target.WorkspaceID,
})
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
return out, err
}
if err := tx.Commit(ctx); err != nil {
return out, err
}
return out, nil
}
// retriggerCancelledTaskSurvivors repairs the surviving inputs from a
// batch containing a comment that was edited or deleted. Cancellation releases
// the one-pending-task constraint; every other comment is replayed through the
// normal authorization/routing path, scoped to the agent whose task carried
// it. Chronological replay makes those survivors coalesce back into one batch
// and lets the latest real comment restamp originator + connected-app context.
func (h *Handler) retriggerCancelledTaskSurvivors(ctx context.Context, issue db.Issue, cancelled []db.AgentTaskQueue, excludedCommentID pgtype.UUID) {
if len(cancelled) == 0 {
return
}
targetsByComment := make(map[string]map[string]pgtype.UUID)
for _, task := range cancelled {
if !task.AgentID.Valid {
continue
}
plannedCommentIDs := append([]pgtype.UUID{}, task.CoalescedCommentIds...)
if task.TriggerCommentID.Valid {
plannedCommentIDs = append(plannedCommentIDs, task.TriggerCommentID)
}
for _, commentID := range plannedCommentIDs {
if !commentID.Valid || (excludedCommentID.Valid && commentID == excludedCommentID) {
continue
}
commentKey := uuidToString(commentID)
if targetsByComment[commentKey] == nil {
targetsByComment[commentKey] = make(map[string]pgtype.UUID)
}
targetsByComment[commentKey][uuidToString(task.AgentID)] = task.AgentID
}
}
comments := make([]db.Comment, 0, len(targetsByComment))
for commentID := range targetsByComment {
comment, err := h.Queries.GetComment(ctx, parseUUID(commentID))
if err != nil {
slog.Warn("retrigger cancelled comment batch: load survivor failed",
"issue_id", uuidToString(issue.ID), "comment_id", commentID, "error", err)
continue
}
// A tombstone is a deleted input, not a survivor.
if comment.IssueID != issue.ID || comment.DeletedAt.Valid {
continue
}
comments = append(comments, comment)
}
sort.Slice(comments, func(i, j int) bool {
if !comments[i].CreatedAt.Time.Equal(comments[j].CreatedAt.Time) {
return comments[i].CreatedAt.Time.Before(comments[j].CreatedAt.Time)
}
return uuidToString(comments[i].ID) < uuidToString(comments[j].ID)
})
for i := range comments {
comment := comments[i]
if isNoteComment(comment.Content) {
continue
}
// Platform recovery comments bypass generic member/agent routing. If
// editing or deleting another comment cancelled the queued batch that
// carried one, replay it through the durable delegated-failure path so
// the cancelled, undelivered task cannot swallow the obligation.
if service.IsDelegatedFailureRecoveryComment(comment) {
if err := h.TaskService.DispatchDelegatedFailureRecoveryComment(ctx, comment, pgtype.UUID{}); err != nil {
slog.Warn("retrigger cancelled comment batch: delegated failure recovery replay failed",
"issue_id", uuidToString(issue.ID),
"comment_id", uuidToString(comment.ID),
"error", err,
)
}
continue
}
var parentComment *db.Comment
if comment.ParentID.Valid {
if parent, err := h.Queries.GetComment(ctx, comment.ParentID); err == nil {
parentComment = &parent
}
}
actorType := comment.AuthorType
actorID := uuidToString(comment.AuthorID)
originatorUserID := actorID
if actorType != "member" {
originatorUserID = uuidToString(h.TaskService.ResolveOriginatorFromTriggerComment(ctx, issue.WorkspaceID, comment.ID))
}
triggers, _ := h.computeCommentAgentTriggers(ctx, issue, comment.Content, parentComment, actorType, actorID, commentTriggerComputeOptions{
ExcludeTriggerCommentID: comment.ID,
AuthoringTaskID: comment.SourceTaskID,
OriginatorUserID: originatorUserID,
})
targets := targetsByComment[uuidToString(comment.ID)]
scoped := make([]commentAgentTrigger, 0, len(targets))
for _, trigger := range triggers {
if _, ok := targets[uuidToString(trigger.Agent.ID)]; ok {
scoped = append(scoped, trigger)
}
}
if len(scoped) > 0 {
h.enqueueCommentAgentTriggers(ctx, issue, comment.ID, scoped)
}
}
}
// loadCommentForActor resolves a {commentId} URL param to a comment in the
// caller's workspace. Returns the comment, the workspace UUID, the actor
// identity, and ok. Resolve / unresolve handlers share this scaffolding so the
// workspace membership + tenant guard stay identical. Any comment (root or
// reply) may be resolved: resolving a root collapses the whole thread; resolving
// a reply marks it as the thread's resolution. Which one is the thread's
// resolution is a pure frontend derivation, so the backend stays a plain setter.
func (h *Handler) loadCommentForActor(w http.ResponseWriter, r *http.Request) (db.Comment, string, string, string, bool) {
commentId := chi.URLParam(r, "commentId")
userID, ok := requireUserID(w, r)
if !ok {
return db.Comment{}, "", "", "", false
}
commentUUID, ok := parseUUIDOrBadRequest(w, commentId, "comment id")
if !ok {
return db.Comment{}, "", "", "", false
}
workspaceID := h.resolveWorkspaceID(r)
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
if !ok {
return db.Comment{}, "", "", "", false
}
if _, ok := h.workspaceMember(w, r, workspaceID); !ok {
return db.Comment{}, "", "", "", false
}
comment, err := h.Queries.GetCommentInWorkspace(r.Context(), db.GetCommentInWorkspaceParams{
ID: commentUUID,
WorkspaceID: wsUUID,
})
// A deleted comment's tombstone cannot be a thread's resolution.
if err != nil || comment.DeletedAt.Valid {
writeError(w, http.StatusNotFound, "comment not found")
return db.Comment{}, "", "", "", false
}
actorType, actorID := h.resolveActor(r, userID, workspaceID)
return comment, workspaceID, actorType, actorID, true
}
func (h *Handler) ResolveComment(w http.ResponseWriter, r *http.Request) {
r = h.withWakeupActor(r)
comment, workspaceID, actorType, actorID, ok := h.loadCommentForActor(w, r)
if !ok {
return
}
wasResolved := comment.ResolvedAt.Valid
actorUUID, err := util.ParseUUID(actorID)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid actor id")
return
}
// Single-resolution invariant: a thread has at most one resolved comment, so
// resolving this one must clear any other resolution in the same thread. Both
// writes run in one tx — clearing the old resolution and setting the new one
// is atomic, so a crash can never leave two resolutions (or none) visible.
tx, err := h.beginWakeupWrite(r.Context())
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to resolve comment")
return
}
defer tx.Rollback(r.Context())
qtx := h.Queries.WithTx(tx)
cleared, err := qtx.ClearOtherThreadResolutions(r.Context(), db.ClearOtherThreadResolutionsParams{
TargetID: comment.ID,
IssueID: comment.IssueID,
WorkspaceID: comment.WorkspaceID,
})
if err != nil {
slog.Warn("clear other thread resolutions failed", append(logger.RequestAttrs(r), "error", err, "comment_id", uuidToString(comment.ID))...)
writeError(w, http.StatusInternalServerError, "failed to resolve comment")
return
}
updated, err := qtx.ResolveComment(r.Context(), db.ResolveCommentParams{
ID: comment.ID,
ResolvedByType: pgtype.Text{String: actorType, Valid: true},
ResolvedByID: actorUUID,
})
if errors.Is(err, pgx.ErrNoRows) {
// Deleted after it was loaded.
writeError(w, http.StatusNotFound, "comment not found")
return
}
if err != nil {
slog.Warn("resolve comment failed", append(logger.RequestAttrs(r), "error", err, "comment_id", uuidToString(comment.ID))...)
writeError(w, http.StatusInternalServerError, "failed to resolve comment")
return
}
if err := tx.Commit(r.Context()); err != nil {
slog.Warn("resolve comment commit failed", append(logger.RequestAttrs(r), "error", err, "comment_id", uuidToString(comment.ID))...)
writeError(w, http.StatusInternalServerError, "failed to resolve comment")
return
}
// Emit a comment:unresolved per cleared sibling so granular realtime
// consumers (which patch a single comment in place) drop the stale
// resolution instead of showing two. Published after commit so no event ever
// describes an uncommitted state.
for _, c := range cleared {
clearedID := uuidToString(c.ID)
clearedReactions := h.groupReactions(r, []pgtype.UUID{c.ID})
clearedAtt := h.groupAttachments(r, []pgtype.UUID{c.ID})
clearedResp := commentToResponse(c, clearedReactions[clearedID], clearedAtt[clearedID])
slog.Info("comment unresolved (replaced)", append(logger.RequestAttrs(r), "comment_id", clearedID)...)
h.publish(protocol.EventCommentUnresolved, workspaceID, actorType, actorID, map[string]any{"comment": clearedResp})
}
grouped := h.groupReactions(r, []pgtype.UUID{updated.ID})
groupedAtt := h.groupAttachments(r, []pgtype.UUID{updated.ID})
cid := uuidToString(updated.ID)
resp := commentToResponse(updated, grouped[cid], groupedAtt[cid])
// Suppress the target event on a re-resolve no-op so consumers do not
// re-process an unchanged thread (notifications, log spam). Cleared siblings
// still get their own events above — those rows did change.
if !wasResolved {
slog.Info("comment resolved", append(logger.RequestAttrs(r), "comment_id", cid)...)
h.publish(protocol.EventCommentResolved, workspaceID, actorType, actorID, map[string]any{"comment": resp})
}
writeJSON(w, http.StatusOK, resp)
}
func (h *Handler) UnresolveComment(w http.ResponseWriter, r *http.Request) {
r = h.withWakeupActor(r)
comment, workspaceID, actorType, actorID, ok := h.loadCommentForActor(w, r)
if !ok {
return
}
wasResolved := comment.ResolvedAt.Valid
updated, err := wakeupWrite(h, r, func(q *db.Queries) (db.Comment, error) {
return q.UnresolveComment(r.Context(), comment.ID)
})
if err != nil {
slog.Warn("unresolve comment failed", append(logger.RequestAttrs(r), "error", err, "comment_id", uuidToString(comment.ID))...)
writeError(w, http.StatusInternalServerError, "failed to unresolve comment")
return
}
grouped := h.groupReactions(r, []pgtype.UUID{updated.ID})
groupedAtt := h.groupAttachments(r, []pgtype.UUID{updated.ID})
cid := uuidToString(updated.ID)
resp := commentToResponse(updated, grouped[cid], groupedAtt[cid])
if wasResolved {
slog.Info("comment unresolved", append(logger.RequestAttrs(r), "comment_id", cid)...)
h.publish(protocol.EventCommentUnresolved, workspaceID, actorType, actorID, map[string]any{"comment": resp})
}
writeJSON(w, http.StatusOK, resp)
}