Files
multica/server/internal/handler/issue_trigger.go
J 451cb2b2a7 feat(attribution): split accountable_user_id from originator, close enqueue bypasses (MUL-4302)
Phase 1, per Bohan's decision on the MUL-4302 thread: audit and authorization
answer different questions and get different columns.

- Migration 151 adds agent_task_queue.accountable_user_id (no FK, no cascade).
  Authorization keeps reading ONLY originator_user_id (canInvokeAgent A2A gate,
  Composio overlay); audit/UI/usage read accountable_user_id + source + evidence.
- Invariant (finalizeAttribution, single chokepoint + §11 tests): originator
  non-null ⟹ accountable equals it. The two diverge only when originator is null
  (autopilot / degraded fallback), which is the deferred rule_owner/owner_fallback
  increment; this lands the column + mirror-write so that split has a home.

- Close the NULL-source enqueue bypasses Elon flagged: chat, quick-create,
  deferred-fallback and run_only-autopilot now stamp originator_source + evidence
  (+ accountable where a human exists). Autopilot stays unattributed until the
  rule-version snapshot table lands, but is no longer a silent NULL-source row.
  Retry inherits accountable_user_id like the rest of the attribution lineage.

- Fix assign/promote attribution (§4): a member who assigns/promotes an existing
  issue is now the accountable human (and, by the invariant, originator) ahead of
  the issue creator. Threaded as an OPTIONAL actor override, so comment/rerun/
  autopilot paths keep today's resolution and create-with-assignee (creator ==
  actor) is unchanged. The squad leader gate already judged the same member.

Also merges origin/main: renumbers the attribution migration 149→150 (main took
149 for issue_origin_agent_create) and folds agent_create into ClassifyDirect's
origin inheritance.

go build/vet/gofmt clean; attribution unit tests + service stamp/actor tests +
handler suite pass on a fresh DB migrated through 151.

Co-authored-by: multica-agent <github@multica.ai>
2026-07-09 17:38:19 +08:00

241 lines
9.2 KiB
Go

package handler
import (
"context"
"encoding/json"
"net/http"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/internal/service"
"github.com/multica-ai/multica/server/internal/util"
agentver "github.com/multica-ai/multica/server/pkg/agent"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// maxPreviewTriggerIssues caps a single preview request so a pathological
// selection cannot fan out into thousands of readiness probes.
const maxPreviewTriggerIssues = 500
// issueTriggerWriteProbe builds the probe the write paths feed to
// WillEnqueueRun. The private-agent gate is already enforced at the HTTP
// boundary (validateAssigneePair on assign) and inside enqueueSquadLeaderTask
// (canEnqueueSquadLeader), so a write must NOT re-run or sink it — it passes
// allow-all. The self-loop check needs the request's X-Task-ID header.
func (h *Handler) issueTriggerWriteProbe(r *http.Request, actorType string, issue db.Issue) service.IssueTriggerProbe {
return service.IssueTriggerProbe{
CanAccessAgent: nil, // allow-all; gate lives at the write boundary
IsSelfLoop: func() bool {
return h.isAgentRunningOnIssue(r, actorType, issue)
},
}
}
// issueTriggerPreviewProbe mirrors the real write-time gates for the read-only
// preview: the private-agent gate (so preview never leaks a private agent's
// readiness to a member who cannot see it — matching validateAssigneePair /
// canEnqueueSquadLeader) and the same self-loop guard.
func (h *Handler) issueTriggerPreviewProbe(r *http.Request, actorType, actorID, workspaceID string, issue db.Issue) service.IssueTriggerProbe {
originatorUserID := h.invokeOriginatorFromRequest(r, actorType, actorID)
return service.IssueTriggerProbe{
CanAccessAgent: func(agent db.Agent) bool {
return h.canInvokeAgent(r.Context(), agent, actorType, actorID, originatorUserID, workspaceID)
},
IsSelfLoop: func() bool {
return h.isAgentRunningOnIssue(r, actorType, issue)
},
}
}
// dispatchIssueRun executes the enqueue side effect for a decision produced by
// WillEnqueueRun, carrying an optional handoff note into the run's opening
// context. The squad path still flows through enqueueSquadLeaderTask so the
// leader access gate and pending dedup stay in one place.
func (h *Handler) dispatchIssueRun(ctx context.Context, issue db.Issue, trigger service.IssueRunTrigger, actorType, actorID, handoffNote string) {
switch trigger.AssigneeType {
case "agent":
// The member who performed this assign/promote is the accountable human
// for the run (MUL-4302 §4). An agent actor is not a human, so only a
// member actor is threaded; otherwise attribution falls back to the chain.
_, _ = h.TaskService.EnqueueTaskForIssueWithHandoff(ctx, issue, handoffNote, memberActorUserID(actorType, actorID))
case "squad":
h.enqueueSquadLeaderTask(ctx, issue, pgtype.UUID{}, actorType, actorID, handoffNote)
}
}
// memberActorUserID returns the acting member's user id as a pgtype.UUID when the
// actor is a member, and an invalid UUID otherwise (an agent actor id is not a
// human and must never become an accountable human). Used to thread the
// assign/promote actor into the attribution resolver (MUL-4302 §4).
func memberActorUserID(actorType, actorID string) pgtype.UUID {
if actorType != "member" {
return pgtype.UUID{}
}
uid, err := util.ParseUUID(actorID)
if err != nil {
return pgtype.UUID{}
}
return uid
}
// IssueTriggerPreviewRequest asks "if I apply this assignee and/or status to
// these issues (or create one), which runs will start". All fields are
// optional; a nil prospective field means "leave unchanged".
type IssueTriggerPreviewRequest struct {
// IssueIDs are existing issues to evaluate (single assign, single status,
// or a batch). Empty with IsCreate=true evaluates a candidate new issue.
IssueIDs []string `json:"issue_ids"`
// IsCreate previews a not-yet-persisted issue from AssigneeType/ID/Status.
IsCreate bool `json:"is_create"`
AssigneeType *string `json:"assignee_type"`
AssigneeID *string `json:"assignee_id"`
Status *string `json:"status"`
}
// IssueTriggerPreviewItem is one issue that WILL start a run under the
// prospective write. AgentID is the runnable agent (squad leader for squads).
// HandoffSupported is the soft-gate signal: false when the target runtime's
// daemon is too old to render a handoff note, so the UI can gray out the note
// box rather than silently drop the text. The assignment itself still works.
type IssueTriggerPreviewItem struct {
IssueID string `json:"issue_id"`
AgentID string `json:"agent_id"`
Source string `json:"source"`
HandoffSupported bool `json:"handoff_supported"`
}
// IssueTriggerPreviewResponse lists every issue that will enqueue plus a total
// the UI can show directly ("将启动 N 个"). Issues that will NOT start a run are
// simply absent, so total_count == len(triggers).
type IssueTriggerPreviewResponse struct {
Triggers []IssueTriggerPreviewItem `json:"triggers"`
TotalCount int `json:"total_count"`
}
// PreviewIssueTrigger dry-runs WillEnqueueRun for a prospective issue write and
// returns the runs that would start, without any side effect. It is the single
// authority the four entry points (create / single assign / single status /
// batch) consult so the frontend never re-implements the enqueue rule
// (MUL-3375). Mirrors PreviewCommentTriggers.
func (h *Handler) PreviewIssueTrigger(w http.ResponseWriter, r *http.Request) {
userID, ok := requireUserID(w, r)
if !ok {
return
}
workspaceID := h.resolveWorkspaceID(r)
if workspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace is required")
return
}
var req IssueTriggerPreviewRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
if len(req.IssueIDs) > maxPreviewTriggerIssues {
writeError(w, http.StatusBadRequest, "too many issue_ids")
return
}
// Resolve the prospective assignee once — a malformed id is a deterministic
// 400, never a silent miscount.
var (
newAssigneeType pgtype.Text
newAssigneeID pgtype.UUID
hasNewAssignee bool
)
if req.AssigneeType != nil && *req.AssigneeType != "" && req.AssigneeID != nil && *req.AssigneeID != "" {
id, parseOK := parseUUIDOrBadRequest(w, *req.AssigneeID, "assignee_id")
if !parseOK {
return
}
newAssigneeType = pgtype.Text{String: *req.AssigneeType, Valid: true}
newAssigneeID = id
hasNewAssignee = true
}
actorType, actorID := h.resolveActor(r, userID, workspaceID)
resp := IssueTriggerPreviewResponse{Triggers: make([]IssueTriggerPreviewItem, 0)}
appendTrigger := func(issue db.Issue, in service.IssueTriggerInput) {
probe := h.issueTriggerPreviewProbe(r, actorType, actorID, workspaceID, issue)
if trigger, ok := h.IssueService.WillEnqueueRun(r.Context(), in, probe); ok {
resp.Triggers = append(resp.Triggers, IssueTriggerPreviewItem{
IssueID: uuidToString(trigger.IssueID),
AgentID: uuidToString(trigger.AgentID),
Source: string(trigger.Source),
HandoffSupported: h.runtimeSupportsHandoff(r.Context(), trigger.AgentID),
})
}
}
if req.IsCreate {
wsUUID, err := util.ParseUUID(workspaceID)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid workspace")
return
}
status := "todo"
if req.Status != nil && *req.Status != "" {
status = *req.Status
}
candidate := db.Issue{
WorkspaceID: wsUUID,
Status: status,
AssigneeType: newAssigneeType,
AssigneeID: newAssigneeID,
}
appendTrigger(candidate, service.IssueTriggerInput{Issue: candidate, IsCreate: true})
resp.TotalCount = len(resp.Triggers)
writeJSON(w, http.StatusOK, resp)
return
}
for _, rawID := range req.IssueIDs {
issueUUID, err := util.ParseUUID(rawID)
if err != nil {
continue // malformed id contributes no trigger; deterministic
}
loaded, err := h.Queries.GetIssueInWorkspace(r.Context(), db.GetIssueInWorkspaceParams{
ID: issueUUID,
WorkspaceID: parseUUID(workspaceID),
})
if err != nil {
continue // cross-workspace / unknown id contributes no trigger
}
post := loaded
in := service.IssueTriggerInput{PrevStatus: loaded.Status}
if hasNewAssignee {
post.AssigneeType = newAssigneeType
post.AssigneeID = newAssigneeID
in.AssigneeChanged = loaded.AssigneeType.String != newAssigneeType.String ||
uuidToString(loaded.AssigneeID) != uuidToString(newAssigneeID)
}
if req.Status != nil && *req.Status != "" {
post.Status = *req.Status
in.StatusChanged = loaded.Status != *req.Status
}
in.Issue = post
appendTrigger(post, in)
}
resp.TotalCount = len(resp.Triggers)
writeJSON(w, http.StatusOK, resp)
}
// runtimeSupportsHandoff reports whether the agent's bound runtime reports a
// CLI version new enough to render handoff notes. Drives the preview's
// handoff_supported soft-gate signal. Any resolution failure → false (degrade).
func (h *Handler) runtimeSupportsHandoff(ctx context.Context, agentID pgtype.UUID) bool {
agent, err := h.Queries.GetAgent(ctx, agentID)
if err != nil || !agent.RuntimeID.Valid {
return false
}
rt, err := h.Queries.GetAgentRuntime(ctx, agent.RuntimeID)
if err != nil {
return false
}
return agentver.HandoffSupported(readRuntimeCLIVersion(rt.Metadata))
}