mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-09 01:46:46 +08:00
* feat(daemon): worktree mode for local_directory resources (MUL-5707) Local directory resources serialise every task on a per-path mutex, so two tasks on one folder queue behind each other. This adds an opt-in worktree mode: each task gets its own git worktree of the user's repo inside the daemon's env root, so tasks run concurrently and deliver their work as a branch instead of editing the user's working copy. Opt in per resource with execution_mode=worktree (default stays in_place, so existing resources are unchanged). Requires a git repo with at least one commit; a non-git directory fails with an actionable message rather than silently falling back to serial execution. Design points that make the mode safe to use: - The agent starts from what the user sees, not HEAD. Tracked edits are captured with `git stash create` (which writes a commit object without touching the index, working tree, or stash list) and replayed into the worktree; untracked non-ignored files are copied, bounded at 2000 files / 200MB with the skipped count logged at warn level. - That replayed state is committed as a baseline, so "did this task change anything?" has an exact answer and `git diff <baseline>..<branch>` is precisely the agent's work. - Nothing is silently discarded: leftovers are committed to the branch before the worktree is removed, including when the task fails partway. - A task that changes nothing leaves no branch behind. - Sidecars stay out of the branch by running the existing CleanupRuntimeConfig + CleanupSidecars pass over the worktree before finalizing. The .git/info/exclude approach repocache uses is not usable here: a linked worktree resolves info/exclude to the user's own common git dir, so it would change what `git status` hides in their checkout. - Worktree-mode env roots do not get the local_directory GC exemption, so they age out normally instead of accumulating one env root per task. - `git worktree prune` on create self-heals registrations orphaned by a crashed daemon. Adds --execution-mode to `project resource add/update`, and updates the docs and the projects-and-resources built-in skill. Co-authored-by: multica-agent <[email protected]> * fix(daemon): close data-loss and isolation gaps in worktree mode (MUL-5707) Review follow-up on #6759. Six issues, all on failure or skew paths. 1. Preserve work when the final commit fails. Finalize logged a warning and then ran `git worktree remove --force`, destroying the agent's changes. commit.gpgSign with no key available to the daemon triggers this, and --no-verify does not disable signing. Finalize now returns an error and deliberately keeps the worktree — it stays registered, so `git worktree list` points at the work — and the daemon fails the task instead of reporting success with a branch that is missing the changes. 2. Fail closed on an unsupported execution_mode. Treating an unknown mode as in_place was wrong: execution_mode is how a user asks for ISOLATION, so the "safe" fallback would edit the working copy they asked us to stay out of. An unrecognised mode now fails the task naming the version skew. 3. Keep Codex conversations alive across turns. The per-issue session store was keyed on LocalWorkDir != "", so worktree tasks — which also never reuse an env root — got a fresh empty CODEX_HOME per turn and silently lost the thread. Covered by a two-turn regression test that fails without the fix. 4. Fail closed on an incomplete snapshot. A failed stash create/apply, a failed untracked copy, or hitting the copy bounds all degraded to "agent sees HEAD" with only a daemon log line. A half-replayed tree looks like a working checkout, so nothing downstream questions it while the agent reads different code than the user has. All of these now abort the prepare and tear the partial worktree back down. 5. Do not snapshot a directory mid-mutation. The same real path can be attached to another project as an in_place resource; snapshotting under a running in_place task captured a half-written tree plus that task's in-flight .agent_context/.multica sidecars, which then reached the branch and could feed the agent another issue's brief. Worktree prepare now takes the per-path mutex for the snapshot alone (released before the run, so tasks still overlap), and the untracked replay skips Multica's own sidecars. 6. Stop the test helper from masking regressions. gitRun skipped on any git failure, including post-finalize assertions, so a lost branch could report as a skip. Git availability is now checked once at repo setup; every later git failure is fatal. Also rebased onto main, keeping both MulticaConfigRoot and LocalWorktree in the Environment initialiser. Co-authored-by: multica-agent <[email protected]> * fix(daemon): close sidecar-commit and wait-visibility gaps in worktree mode; gate worktree saves on daemon version (MUL-5707) - runTask: register the sidecar-cleanup defer immediately after the Finalize defer. Previously two early returns (temp-dir setup, StartTask failure) sat between them, and on those paths Finalize auto-committed the sidecars Prepare had just written — delivering a branch whose only content is Multica's own runtime files (in_place: leaving them in the user's tree). - runTask: give the worktree snapshot wait the same plumbing as the in-place acquire — waiting_local_directory status, resourceWaitTasks accounting, and a cancellation poller. The holder can be an hours-long in_place task; the wait was invisible and a user cancel went unnoticed until the 5-minute prepare timeout. - isMulticaSidecarPath: match sidecar dirs as whole path segments at any depth, not only at the repo root. An in_place resource pointed at a SUBDIRECTORY leaves its sidecars below that subdirectory; replaying them committed another issue's brief into the delivered branch. - handler: refuse saving execution_mode=worktree while the owning daemon's registered cli_version is below MinLocalWorktreeCLIVersion (422, daemon_version_unsupported). An old daemon json-skips the field and runs tasks IN PLACE — the daemon-side unknown-mode refusal cannot protect a daemon that predates the field, so the server has to. Residual (documented): a daemon downgraded after the save needs a claim-time gate. - copyUntrackedFiles: read the -z listing raw instead of TrimSpace-ing it, which corrupted filenames beginning/ending with whitespace into paths that silently vanished from the replay. - docs + skill: describe the save-time version gate; drop the stale claim that old runtimes fail the task themselves. Co-authored-by: multica-agent <[email protected]> * feat(views): UI switch for local_directory execution mode (MUL-5707) Worktree mode shipped CLI/API-only, which for its target user — someone running several tasks at once from Desktop — means it does not exist. This adds the switch to the resources panel. The choice is framed by what the user gets back, not by the mode identifiers: "Edit this folder directly" (one task at a time, changes land in your working copy) vs "Run in parallel, isolated" (tasks run at the same time and each hands back an agent/… branch). `in_place` / `worktree` stay visible as a secondary hint for anyone cross-referencing the CLI or docs. Two entry points, matching where the decision is actually made: - The add-a-folder flow now asks for the mode after the folder is picked and validated, instead of creating the resource and leaving the mode to be discovered later. - Every local_directory row has a mode control. It is deliberately NOT gated on being the owning machine — switching the mode only rewrites a field, so it works from the web app too, unlike rename or the folder picker. Both blocked states are handled where the user chooses, not after they commit: - Non-git folder: the desktop validator now reports whether the path is inside a git working tree (walking up for `.git`, so a subdirectory of a repo counts), and the option is disabled with that reason. Unknown — an older desktop build, or an existing row — stays permissive; the daemon re-checks authoritatively. - Daemon too old: the version floor is mirrored from the server as MIN_LOCAL_WORKTREE_CLI_VERSION and checked against the runtime for the resource's OWN daemon, not the machine the browser happens to be on. A server rejection is rendered inline in the dialog rather than as a toast, so the dialog stays actionable. A "Parallel" badge on the row is always visible (not hover-only): without it there is no way to tell whether tasks on a folder edit it directly or hand back a branch, which is the first thing anyone asks when a task queues. Also adds `execution_mode` to LocalDirectoryResourceRef — the field existed server-side but had no TypeScript counterpart — and brings the zh/ja/ko project-resources docs in line with the English ones, which were the only locale updated when the mode landed. Co-authored-by: multica-agent <[email protected]> * fix(daemon,handler): close worktree isolation and delivery gaps (MUL-5707) Second review round. Five must-fix items, all on paths where the mode's promises could still break. 1. A failed sidecar cleanup no longer ships Multica's own files as the user's branch. The cleanup errors were logged and Finalize then ran `git add -A`, so a permission/disk/lock failure could deliver a branch whose entire content was the runtime brief and issue context. Worktree mode now escalates a cleanup failure to an abort: nothing is committed, the worktree is kept for inspection, and the task fails with the preserved path. In place it stays a warning — no commit, no branch, and the leftovers are visible in the user's own tree. 2. The untracked replay no longer drops files silently. `Lstat` failures and symlinks were `continue`d without counting, so the caller read a partial snapshot as complete. Both now count as skipped, which the existing fail-closed check turns into a refused prepare. Non-content entries (sockets, FIFOs, devices) stay uncounted deliberately. 3. Prepare rolls back a worktree it created. Any later failure — context files, provider home, MCP config — returned without an Environment, so nothing downstream knew a worktree existed, leaving a registration and a branch no task ever ran in. A deferred Discard now runs unless Prepare reached its successful return. 4. The daemon-version gate moved to where it can actually hold. Save-time validation cannot see a machine downgraded afterwards, and an old daemon json-skips execution_mode entirely — it would run the task in place, editing the working copy the user asked to isolate. Claims are now checked against the runtime that is actually claiming, and a blocked task is cancelled with a reason rather than left to be redelivered to the same too-old runtime forever. 5. branch_name is delivered end to end. The daemon sent it, but TaskCompleteRequest had no field so it vanished at decode, and the failure path never sent it at all. It now has a column, travels on both terminal callbacks, and renders as a copyable row in Run details. The failure path matters most: worktree mode commits the agent's leftovers before tearing the worktree down, so a run that died partway still has work worth finding. Docs (all four locales) and the projects-and-resources skill are updated for the claim-time gate and the branch surface. Co-authored-by: multica-agent <[email protected]> * fix(handler,daemon): migration renumber, cancel reasons, branch on every terminal path (MUL-5707) Third review round. 1. Rebased onto main and renumbered the migration to 285 — main had taken 273 and is now at 284. sqlc regenerated on top of the rebased queries; the conflicts were confined to generated files. 2. The claim gate now persists why it cancelled. It called CancelTask, which writes only status='cancelled', and returned the reason as a 4xx that reaches the daemon's log and nothing else — batch claim drops the task from its response entirely. A run refused for a durable reason the user has to act on cannot present as a bare "cancelled", so this adds CancelAgentTaskWithReason and records error + failure_reason on the row. 3. branch_name now survives every terminal path, not just the two direct callbacks. Three were dropping it: - the context-exhausted complete→fail reroute did not copy it; - the daemon's complete→fail fallback (server rejected the complete, agent had succeeded) did not set it; - both cancellation paths discarded the result wholesale, even though the worktree is finalized — and its work committed — before the daemon learns of the cancel. The branch now rides the cancel-ack and is recorded by SetAgentTaskBranchName, which never overwrites a name a terminal callback already wrote. 4. Rollout/rollback compatibility. Isolation is enforced by two server-side gates, so a daemon downgraded at any point after the resource was saved is caught. The gap they cannot close is rolling the SERVER back past this change while sub-floor daemons are still connected: an old server has no gate and an old daemon ignores execution_mode. That constraint is now documented where an operator will meet it (the migration and the docs), and the version floor is pinned across server and frontend by test so the two gates cannot silently drift apart. Co-authored-by: multica-agent <[email protected]> * fix(server,views): route claim-gate cancels through TaskService, make cancel reasons visible, harden cancel-ack branch delivery (MUL-5707) Closes the fourth-round review findings on worktree mode's cancellation chain: - Claim gate: cancel through a new TaskService.CancelTaskWithReason instead of a raw CancelAgentTaskWithReason query, so a gate-cancelled task gets the full cancellation side effects — audit capture, chat settle, agent status reconcile, task:cancelled broadcast, NotifyTaskFinished. Added REAL claim integration tests: singular claim (422 + row cancelled with reason), batch claim (task omitted + row cancelled with reason), and a current-runtime control. - Cancel reasons are now user-visible: execution log rows, agent activity rows, and the transcript dialog label a SERVER-cancelled task (persisted error/failure_reason) like a failed one, with the actionable text on hover and a Reason row in Run details. User-initiated cancels stay a plain "Cancelled". New cancelReasonLabel helper + local_directory_error label; details_reason locale key in en/zh-Hans/ja/ko; widened AgentTask.failure_reason to the open string the wire actually carries. - Cancel-ack is now a terminal delivery: the daemon retries it on transient errors (same schedule as complete/fail), the server fails LOUD (5xx) when persisting the branch/error fails instead of warn-and-200, and re-broadcasts task:cancelled after a successful write so clients that already refetched a branchless row pick up the delivery. Replays are idempotent (never overwrite). - Cancel + Finalize-abort no longer swallows the preserved-worktree pointer: the finalize error is wrapped in worktreePreservedError (joined onto an existing run error, never displacing it), extracted on the cancel path, and delivered via the ack into agent_task_queue.error through the new SetAgentTaskErrorIfEmpty query — never overwriting a reason recorded by a fail callback or the claim gate. Other run errors on a cancelled run stay discarded as before: they are expected noise. - Docs: the self-hosted rollback constraint callout now exists in zh/ja/ko, not just English. Co-authored-by: multica-agent <[email protected]> * fix(handler): guard cancel-ack writes to cancelled rows; requeue when the claim gate's cancel fails (MUL-5707) Fifth-round review fixes: - SetAgentTaskBranchName / SetAgentTaskErrorIfEmpty now carry a status='cancelled' CAS. The daemon acks on EVERY terminal status it observes, so a late or replayed ack from a stale run could previously backfill its branch or preserved-path error into a completed/failed row whose own callback — the authoritative channel — recorded nothing, without any event invalidating the change. A CAS-refused write is a deliberate no-op (still 200: nothing to retry), and the rebroadcast stays guarded by the same status check. Regression tests: a late ack against a completed row with empty fields and against a failed row leaves both untouched. - When the worktree claim gate's CancelTaskWithReason fails, the claimed row no longer strands in dispatched (daemon refused, nothing to release it but stale reclaim). The gate requeues via RequeueTaskAfterClaimFailure — the next claim re-runs the gate and retries the cancel — and reports 5xx so the singular daemon reads a transient server problem, not a claim refusal; the batch path omits the task as before, now with the row requeued. Fault injection tests (row-scoped trigger raising on the cancel UPDATE while letting the requeue through) cover singular and batch. Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: Bohan-J <[email protected]> Co-authored-by: multica-agent <[email protected]>
524 lines
19 KiB
Go
524 lines
19 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: dingtalk.sql
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const deleteDingTalkInstallationForReplacement = `-- name: DeleteDingTalkInstallationForReplacement :one
|
|
WITH retired AS (
|
|
DELETE FROM channel_installation ci
|
|
WHERE ci.id = $1
|
|
AND ci.workspace_id = $2
|
|
AND ci.agent_id = $3
|
|
AND ci.channel_type = 'dingtalk'
|
|
RETURNING ci.id
|
|
),
|
|
cleared_chat_sessions AS (
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
RETURNING chat_session_id
|
|
),
|
|
cleared_group_routes AS (
|
|
DELETE FROM dingtalk_group_route
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
),
|
|
cleared_outbound_cards AS (
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions)
|
|
),
|
|
cleared_binding_tokens AS (
|
|
DELETE FROM channel_binding_token
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
),
|
|
cleared_user_bindings AS (
|
|
DELETE FROM channel_user_binding
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
),
|
|
cleared_inbound_dedup AS (
|
|
DELETE FROM channel_inbound_message_dedup
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
),
|
|
detached_audit AS (
|
|
UPDATE channel_inbound_audit SET installation_id = NULL
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
),
|
|
detached_media_intents AS (
|
|
UPDATE channel_media_pending_object SET installation_id = NULL
|
|
WHERE installation_id IN (SELECT id FROM retired)
|
|
)
|
|
SELECT retired.id FROM retired
|
|
`
|
|
|
|
type DeleteDingTalkInstallationForReplacementParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
}
|
|
|
|
// Retires an installation when the same agent is connected with a DIFFERENT
|
|
// AppKey. A senderStaffId is scoped to one DingTalk organization, so none of the
|
|
// old installation's identity, token, session, dedup, or outbound state may
|
|
// cross into the new robot. The caller inserts a fresh installation in the same
|
|
// transaction, giving the replacement a new installation_id that also fences
|
|
// late writes and replies from the old connection.
|
|
//
|
|
// Chat sessions themselves remain as history, but their channel bindings and
|
|
// outbound cards are removed. Audit and media-intent rows remain useful for
|
|
// diagnostics / reconciliation, so their installation references are detached.
|
|
func (q *Queries) DeleteDingTalkInstallationForReplacement(ctx context.Context, arg DeleteDingTalkInstallationForReplacementParams) (pgtype.UUID, error) {
|
|
row := q.db.QueryRow(ctx, deleteDingTalkInstallationForReplacement, arg.InstallationID, arg.WorkspaceID, arg.AgentID)
|
|
var id pgtype.UUID
|
|
err := row.Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
const deleteDingTalkStaleGroupChatBinding = `-- name: DeleteDingTalkStaleGroupChatBinding :one
|
|
WITH cleared AS (
|
|
DELETE FROM channel_chat_session_binding b
|
|
WHERE b.installation_id = $1
|
|
AND b.channel_type = 'dingtalk'
|
|
AND b.channel_chat_id = $2::text
|
|
AND COALESCE(
|
|
NULLIF(b.config ->> 'agent_id', ''),
|
|
(
|
|
SELECT i.agent_id::text
|
|
FROM channel_installation i
|
|
WHERE i.id = b.installation_id
|
|
),
|
|
''
|
|
) <> $3::uuid::text
|
|
RETURNING b.chat_session_id
|
|
), cleared_outbound_cards AS (
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared)
|
|
)
|
|
SELECT count(*)::bigint AS cleared_count
|
|
FROM cleared
|
|
`
|
|
|
|
type DeleteDingTalkStaleGroupChatBindingParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ConversationID string `json:"conversation_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
}
|
|
|
|
// A route reassignment normally removes the group's binding in the same query.
|
|
// An inbound message that resolved immediately before the reassignment can,
|
|
// however, finish creating its old-agent binding immediately afterwards. Every
|
|
// subsequent group message runs this guard before EnsureSession. Legacy
|
|
// bindings have no agent stamp; preserve them when the effective route is still
|
|
// the installation's original/default agent, but retire them after a genuine
|
|
// reassignment so the new route can never inherit the old agent's transcript.
|
|
func (q *Queries) DeleteDingTalkStaleGroupChatBinding(ctx context.Context, arg DeleteDingTalkStaleGroupChatBindingParams) (int64, error) {
|
|
row := q.db.QueryRow(ctx, deleteDingTalkStaleGroupChatBinding, arg.InstallationID, arg.ConversationID, arg.AgentID)
|
|
var cleared_count int64
|
|
err := row.Scan(&cleared_count)
|
|
return cleared_count, err
|
|
}
|
|
|
|
const dingTalkGroupRouteMatchesAgent = `-- name: DingTalkGroupRouteMatchesAgent :one
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM dingtalk_group_route r
|
|
JOIN agent a ON a.id = r.agent_id
|
|
AND a.workspace_id = r.workspace_id
|
|
WHERE r.installation_id = $1
|
|
AND r.conversation_id = $2::text
|
|
AND r.agent_id = $3
|
|
AND r.revision = $4
|
|
AND a.kind = 'user'
|
|
AND a.archived_at IS NULL
|
|
) AS matches
|
|
`
|
|
|
|
type DingTalkGroupRouteMatchesAgentParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ConversationID string `json:"conversation_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
RouteRevision int64 `json:"route_revision"`
|
|
}
|
|
|
|
func (q *Queries) DingTalkGroupRouteMatchesAgent(ctx context.Context, arg DingTalkGroupRouteMatchesAgentParams) (bool, error) {
|
|
row := q.db.QueryRow(ctx, dingTalkGroupRouteMatchesAgent,
|
|
arg.InstallationID,
|
|
arg.ConversationID,
|
|
arg.AgentID,
|
|
arg.RouteRevision,
|
|
)
|
|
var matches bool
|
|
err := row.Scan(&matches)
|
|
return matches, err
|
|
}
|
|
|
|
const discoverDingTalkGroupRoute = `-- name: DiscoverDingTalkGroupRoute :one
|
|
|
|
WITH workspace_guard AS MATERIALIZED (
|
|
SELECT w.id
|
|
FROM workspace w
|
|
WHERE w.id = $1
|
|
FOR KEY SHARE
|
|
), installation AS MATERIALIZED (
|
|
SELECT i.id, i.workspace_id, i.agent_id, i.channel_type, i.config, i.status, i.ws_lease_token, i.ws_lease_expires_at, i.installer_user_id, i.installed_at, i.created_at, i.updated_at
|
|
FROM channel_installation i
|
|
JOIN workspace_guard w ON w.id = i.workspace_id
|
|
WHERE i.id = $2
|
|
AND i.workspace_id = $1
|
|
AND i.channel_type = 'dingtalk'
|
|
AND i.status = 'active'
|
|
FOR SHARE OF i
|
|
), group_route AS (
|
|
INSERT INTO dingtalk_group_route (
|
|
workspace_id, installation_id, conversation_id,
|
|
conversation_title, agent_id
|
|
)
|
|
SELECT
|
|
i.workspace_id, i.id, $3::text,
|
|
$4::text, i.agent_id
|
|
FROM installation i
|
|
ON CONFLICT (installation_id, conversation_id) DO UPDATE SET
|
|
conversation_title = CASE
|
|
WHEN EXCLUDED.conversation_title = '' THEN dingtalk_group_route.conversation_title
|
|
ELSE EXCLUDED.conversation_title
|
|
END,
|
|
updated_at = CASE
|
|
WHEN EXCLUDED.conversation_title <> ''
|
|
AND EXCLUDED.conversation_title IS DISTINCT FROM dingtalk_group_route.conversation_title
|
|
THEN now()
|
|
ELSE dingtalk_group_route.updated_at
|
|
END
|
|
RETURNING agent_id, workspace_id, revision
|
|
)
|
|
SELECT r.agent_id,
|
|
r.revision,
|
|
EXISTS (
|
|
SELECT 1 FROM agent a
|
|
WHERE a.id = r.agent_id
|
|
AND a.workspace_id = r.workspace_id
|
|
AND a.kind = 'user'
|
|
AND a.archived_at IS NULL
|
|
) AS agent_active
|
|
FROM group_route r
|
|
`
|
|
|
|
type DiscoverDingTalkGroupRouteParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ConversationID string `json:"conversation_id"`
|
|
ConversationTitle string `json:"conversation_title"`
|
|
}
|
|
|
|
type DiscoverDingTalkGroupRouteRow struct {
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
Revision int64 `json:"revision"`
|
|
AgentActive bool `json:"agent_active"`
|
|
}
|
|
|
|
// DingTalk-specific installation identity operations. The underlying channel_*
|
|
// tables are shared, but these replacement semantics belong to DingTalk's BYO
|
|
// AppKey model and deliberately stay out of the shared channel query surface.
|
|
// Persist a group only after the shared inbound router has accepted the @bot
|
|
// message and validated the sender's identity/workspace membership. The INSERT
|
|
// default is the installation's agent; a later admin reassignment survives
|
|
// subsequent inbound events because the conflict update refreshes only the
|
|
// human-readable title. The returned active bit revalidates the effective route
|
|
// target at runtime while preserving the route across archive/restore. The
|
|
// workspace/installation locks serialize discovery with teardown and revoke:
|
|
// revoke-first rechecks status after the wait and returns no row; discovery-
|
|
// first completes before revoke, matching the existing in-flight semantics.
|
|
func (q *Queries) DiscoverDingTalkGroupRoute(ctx context.Context, arg DiscoverDingTalkGroupRouteParams) (DiscoverDingTalkGroupRouteRow, error) {
|
|
row := q.db.QueryRow(ctx, discoverDingTalkGroupRoute,
|
|
arg.WorkspaceID,
|
|
arg.InstallationID,
|
|
arg.ConversationID,
|
|
arg.ConversationTitle,
|
|
)
|
|
var i DiscoverDingTalkGroupRouteRow
|
|
err := row.Scan(&i.AgentID, &i.Revision, &i.AgentActive)
|
|
return i, err
|
|
}
|
|
|
|
const getDingTalkGroupRouteInWorkspace = `-- name: GetDingTalkGroupRouteInWorkspace :one
|
|
SELECT r.id, r.workspace_id, r.installation_id, r.conversation_id, r.conversation_title, r.agent_id, r.revision, r.discovered_at, r.updated_at
|
|
FROM dingtalk_group_route r
|
|
JOIN channel_installation i ON i.id = r.installation_id
|
|
WHERE r.id = $1
|
|
AND r.workspace_id = $2
|
|
AND i.workspace_id = $2
|
|
AND i.channel_type = 'dingtalk'
|
|
AND i.status = 'active'
|
|
`
|
|
|
|
type GetDingTalkGroupRouteInWorkspaceParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
}
|
|
|
|
// Handler diagnosis deliberately uses the same active-installation boundary as
|
|
// reassignment. Retained routes stay available for reconnect internally, but a
|
|
// revoked installation's route is not a PATCH-visible resource.
|
|
func (q *Queries) GetDingTalkGroupRouteInWorkspace(ctx context.Context, arg GetDingTalkGroupRouteInWorkspaceParams) (DingtalkGroupRoute, error) {
|
|
row := q.db.QueryRow(ctx, getDingTalkGroupRouteInWorkspace, arg.ID, arg.WorkspaceID)
|
|
var i DingtalkGroupRoute
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.InstallationID,
|
|
&i.ConversationID,
|
|
&i.ConversationTitle,
|
|
&i.AgentID,
|
|
&i.Revision,
|
|
&i.DiscoveredAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getDingTalkInstallationOwnerForUpdate = `-- name: GetDingTalkInstallationOwnerForUpdate :one
|
|
SELECT id, COALESCE(config ->> 'app_id', '')::text AS app_id
|
|
FROM channel_installation
|
|
WHERE workspace_id = $1
|
|
AND agent_id = $2
|
|
AND channel_type = 'dingtalk'
|
|
FOR UPDATE
|
|
`
|
|
|
|
type GetDingTalkInstallationOwnerForUpdateParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
}
|
|
|
|
type GetDingTalkInstallationOwnerForUpdateRow struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
AppID string `json:"app_id"`
|
|
}
|
|
|
|
// Reads the current robot identity after LockDingTalkInstallationOwner has
|
|
// serialized the logical owner slot. app_id is non-null for every DingTalk
|
|
// installation; COALESCE treats malformed legacy config as a different robot,
|
|
// which safely replaces it instead of preserving unknown identity state.
|
|
func (q *Queries) GetDingTalkInstallationOwnerForUpdate(ctx context.Context, arg GetDingTalkInstallationOwnerForUpdateParams) (GetDingTalkInstallationOwnerForUpdateRow, error) {
|
|
row := q.db.QueryRow(ctx, getDingTalkInstallationOwnerForUpdate, arg.WorkspaceID, arg.AgentID)
|
|
var i GetDingTalkInstallationOwnerForUpdateRow
|
|
err := row.Scan(&i.ID, &i.AppID)
|
|
return i, err
|
|
}
|
|
|
|
const listDingTalkGroupRoutesByWorkspace = `-- name: ListDingTalkGroupRoutesByWorkspace :many
|
|
SELECT r.id, r.workspace_id, r.installation_id, r.conversation_id, r.conversation_title, r.agent_id, r.revision, r.discovered_at, r.updated_at
|
|
FROM dingtalk_group_route r
|
|
JOIN channel_installation i ON i.id = r.installation_id
|
|
WHERE r.workspace_id = $1
|
|
AND i.workspace_id = $1
|
|
AND i.channel_type = 'dingtalk'
|
|
AND i.status = 'active'
|
|
ORDER BY r.discovered_at ASC
|
|
`
|
|
|
|
func (q *Queries) ListDingTalkGroupRoutesByWorkspace(ctx context.Context, workspaceID pgtype.UUID) ([]DingtalkGroupRoute, error) {
|
|
rows, err := q.db.Query(ctx, listDingTalkGroupRoutesByWorkspace, workspaceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []DingtalkGroupRoute{}
|
|
for rows.Next() {
|
|
var i DingtalkGroupRoute
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.InstallationID,
|
|
&i.ConversationID,
|
|
&i.ConversationTitle,
|
|
&i.AgentID,
|
|
&i.Revision,
|
|
&i.DiscoveredAt,
|
|
&i.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const lockDingTalkGroupRouteForAppend = `-- name: LockDingTalkGroupRouteForAppend :one
|
|
WITH target_agent AS MATERIALIZED (
|
|
SELECT a.id
|
|
FROM agent a
|
|
WHERE a.id = $3
|
|
AND a.kind = 'user'
|
|
AND a.archived_at IS NULL
|
|
FOR SHARE
|
|
)
|
|
SELECT r.revision
|
|
FROM dingtalk_group_route r
|
|
WHERE r.installation_id = $1
|
|
AND r.conversation_id = $2::text
|
|
AND r.agent_id = $3
|
|
AND r.revision = $4
|
|
AND EXISTS (SELECT 1 FROM target_agent)
|
|
FOR SHARE OF r
|
|
`
|
|
|
|
type LockDingTalkGroupRouteForAppendParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ConversationID string `json:"conversation_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
RouteRevision int64 `json:"route_revision"`
|
|
}
|
|
|
|
// This is the durable cutover fence. The active target agent and route are
|
|
// locked in the same order as ReassignDingTalkGroupRoute, and the lock remains
|
|
// held by AppendUserMessage's transaction until the message and dedup mark
|
|
// commit. A PATCH that wins first changes revision and produces no row; an
|
|
// inbound append that wins first makes PATCH wait until that append commits.
|
|
func (q *Queries) LockDingTalkGroupRouteForAppend(ctx context.Context, arg LockDingTalkGroupRouteForAppendParams) (int64, error) {
|
|
row := q.db.QueryRow(ctx, lockDingTalkGroupRouteForAppend,
|
|
arg.InstallationID,
|
|
arg.ConversationID,
|
|
arg.AgentID,
|
|
arg.RouteRevision,
|
|
)
|
|
var revision int64
|
|
err := row.Scan(&revision)
|
|
return revision, err
|
|
}
|
|
|
|
const lockDingTalkInstallationOwner = `-- name: LockDingTalkInstallationOwner :exec
|
|
SELECT pg_advisory_xact_lock(
|
|
hashtextextended(
|
|
($1::uuid)::text || ':' ||
|
|
($2::uuid)::text || ':dingtalk',
|
|
0
|
|
)
|
|
)
|
|
`
|
|
|
|
type LockDingTalkInstallationOwnerParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
}
|
|
|
|
// Serializes install / replacement decisions for one logical
|
|
// (workspace, agent, channel) slot. A different-AppKey replacement deletes the
|
|
// old row and inserts a fresh installation id; the advisory lock closes the
|
|
// gap where two concurrent replacements could otherwise miss each other's new
|
|
// row and update it in place, carrying identity state across robot boundaries.
|
|
func (q *Queries) LockDingTalkInstallationOwner(ctx context.Context, arg LockDingTalkInstallationOwnerParams) error {
|
|
_, err := q.db.Exec(ctx, lockDingTalkInstallationOwner, arg.WorkspaceID, arg.AgentID)
|
|
return err
|
|
}
|
|
|
|
const reassignDingTalkGroupRoute = `-- name: ReassignDingTalkGroupRoute :one
|
|
WITH workspace_guard AS MATERIALIZED (
|
|
SELECT w.id
|
|
FROM workspace w
|
|
WHERE w.id = $1
|
|
FOR KEY SHARE
|
|
), target_agent AS MATERIALIZED (
|
|
SELECT a.id
|
|
FROM agent a
|
|
JOIN workspace_guard w ON w.id = a.workspace_id
|
|
WHERE a.id = $2
|
|
AND a.workspace_id = $1
|
|
AND a.kind = 'user'
|
|
AND a.archived_at IS NULL
|
|
FOR SHARE
|
|
), active_installation AS MATERIALIZED (
|
|
SELECT i.id
|
|
FROM channel_installation i
|
|
JOIN dingtalk_group_route r ON r.installation_id = i.id
|
|
WHERE r.id = $3
|
|
AND r.workspace_id = $1
|
|
AND i.workspace_id = $1
|
|
AND i.channel_type = 'dingtalk'
|
|
AND i.status = 'active'
|
|
AND EXISTS (SELECT 1 FROM target_agent)
|
|
FOR SHARE OF i
|
|
), target AS (
|
|
SELECT r.id, r.workspace_id, r.installation_id, r.conversation_id, r.conversation_title, r.agent_id, r.revision, r.discovered_at, r.updated_at, r.agent_id AS previous_agent_id
|
|
FROM dingtalk_group_route r
|
|
JOIN active_installation i ON i.id = r.installation_id
|
|
WHERE r.id = $3
|
|
AND r.workspace_id = $1
|
|
FOR UPDATE OF r
|
|
), updated AS (
|
|
UPDATE dingtalk_group_route r
|
|
SET agent_id = $2,
|
|
revision = r.revision + 1,
|
|
updated_at = now()
|
|
FROM target t
|
|
WHERE r.id = t.id
|
|
RETURNING r.id, r.workspace_id, r.installation_id, r.conversation_id, r.conversation_title, r.agent_id, r.revision, r.discovered_at, r.updated_at, t.previous_agent_id
|
|
), cleared_chat_sessions AS (
|
|
DELETE FROM channel_chat_session_binding b
|
|
USING updated u
|
|
WHERE u.previous_agent_id IS DISTINCT FROM u.agent_id
|
|
AND b.installation_id = u.installation_id
|
|
AND b.channel_chat_id = u.conversation_id
|
|
RETURNING b.chat_session_id
|
|
), cleared_outbound_cards AS (
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions)
|
|
)
|
|
SELECT id, workspace_id, installation_id, conversation_id,
|
|
conversation_title, agent_id, revision, discovered_at, updated_at
|
|
FROM updated
|
|
`
|
|
|
|
type ReassignDingTalkGroupRouteParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
ID pgtype.UUID `json:"id"`
|
|
}
|
|
|
|
type ReassignDingTalkGroupRouteRow struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ConversationID string `json:"conversation_id"`
|
|
ConversationTitle string `json:"conversation_title"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
Revision int64 `json:"revision"`
|
|
DiscoveredAt pgtype.Timestamptz `json:"discovered_at"`
|
|
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
|
|
}
|
|
|
|
// Reassigning a group must also sever its existing chat-session binding. The
|
|
// next message creates a fresh session for the new agent, so transcripts and
|
|
// pending outbound updates cannot cross agent boundaries. The old chat session
|
|
// remains as history; only its DingTalk route and outbound card state are
|
|
// removed. The target agent existence/workspace/archive check is repeated here
|
|
// so a concurrent archive or delete cannot create a dangling route. The active
|
|
// workspace, active installation, and route locks preserve the teardown order;
|
|
// the installation is share-locked before the route is update-locked. Revoke
|
|
// takes an exclusive lock on that same installation row, defining the race:
|
|
// PATCH-first may finish before revoke, while revoke-first makes PATCH wait,
|
|
// re-check status, and return no row without touching the route or binding.
|
|
func (q *Queries) ReassignDingTalkGroupRoute(ctx context.Context, arg ReassignDingTalkGroupRouteParams) (ReassignDingTalkGroupRouteRow, error) {
|
|
row := q.db.QueryRow(ctx, reassignDingTalkGroupRoute, arg.WorkspaceID, arg.AgentID, arg.ID)
|
|
var i ReassignDingTalkGroupRouteRow
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.InstallationID,
|
|
&i.ConversationID,
|
|
&i.ConversationTitle,
|
|
&i.AgentID,
|
|
&i.Revision,
|
|
&i.DiscoveredAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|