0b750d00dd
Watch is the agent noticing something later: you ask it to tell you when a condition holds, and it answers when it does — or when it can't any more. A watch is a **durable one-shot promise**. The condition is checked on a schedule by deterministic code (no LLM in the checks), the answer lands in the chat once, and then the watch is over. Ten kinds: three on a run, five on a queue, error recurrence, health recovery. ## Stack Stacked on **#4529** (UI), which is stacked on **#4418** (chat, reports, investigate). Merge those first. **#4516** (storybook gallery) sits on top of this branch. ## How to review [**GUIDEBOOK.md**](https://github.com/triggerdotdev/trigger.dev/blob/feat/dashboard-agent-flows-watch/internal-packages/dashboard-agent/GUIDEBOOK.md) on this branch is the behaviour reference — it states the conditions rather than the code, so you can predict what happens without running anything. "The ten watch kinds, and what makes each fire" and "Creating a watch" describe exactly this PR, and the tables there are the spec the code is written against. ## What's inside - **Ten watch kinds**, one deterministic check each (`dashboardAgentWatch*Checks.ts`), with the spec union in `dashboard-agent-contracts/src/watch.ts`. - **Scheduling** — each watch schedules its own next check; due watches of one `(environment, cadence)` group can be checked together in one batch pass, with a sweep as the backstop for expiry, redelivery and retention. - **Delivery** — the in-chat wake and card, an optional email alert (new `DASHBOARD_AGENT_WATCH` alert channel, so it shows on the project's Alerts page with one-click unsubscribe), and an optional investigation when the outcome needs attention. - **Submission ledger** — `watch_submissions`, keyed `(chat_id, client_request_id)`, so a retried card submission replays the recorded outcome instead of creating a second watch. - **Watch token** — a delayed-execution credential accepted only by the watch endpoints, re-checked against the user's live access on every tick. - **Unread work** — the panel polls for wakes that landed while it was closed, so a chat can go unread and light the launcher dot. ## Key decisions **A check result is a 4-way, and only two of them are verdicts.** `satisfied` / `terminal_unsatisfied` are answers; `pending` and `unavailable` are not. Any exception inside any check is caught in one place and becomes `unavailable` with an unverified observation — a check that failed is never evidence. **A completed window is an answer, and whether it is good or bad news is declared per kind, never inferred.** There is a table for that in the guidebook: `run_failed` completing its window is *good* news ("hasn't failed"), `backlog_drain` completing it is not. One rule overrides the table: a window that completed on an unverified observation is neutral and says only that the watch ended without a confirmed answer. **An unreadable source is never a negative answer** — and, because investigations only open on `attention`, it never starts one either. **Identity is `(chat, project, environment)` plus the condition,** enforced by a partial unique index over active rows (`watches_chat_active_identity_key`), not by the read-then-insert check. Cadence, window, note and `ticks` are deliberately not part of it. Two different chats may watch the same thing — a watch is a promise to a chat. **The server resolves the target's name, whatever the model calls it.** The model can't tell a task queue (`task/<id>`) from a custom queue, so both spellings are tried and the stored one wins — and the rewrite happens **before** identity and before the row is written, so the identity, the checks, the link and the wording all see one spelling. **Freshness fences.** Depth falls back from the live counter to the newest 60 s ClickHouse bucket, which only counts as current within 60 s of now. A non-current reading at or below the *quiet line* is refused as `unavailable` rather than believed, so a stale empty bucket is never read as "drained". The stall streak is the one piece of carried state: it lives in the previous check's facts and *freezes* on an unreadable reading rather than breaking. **Chain reliability.** There is no shared cron — each watch (or batch group) schedules its own next tick, so the failure mode to review is the chain dying. A failed batch check is caught, the next tick is scheduled anyway and the run resolves rather than failing, so the chain survives a check that couldn't run; the sweep re-arms groups and finalizes anything still active past its deadline, even when delivery isn't configured. Wake redelivery is id-deduped rather than conditional, because the sweep can't know whether the user was already told. Access is re-authorized on **every** check against the primary — replica lag would extend access the user has already lost. **Wording lives in one place.** `watch-wording.ts` is read by the card, banner, toast, email and the agent's own narration, and the numbers come from the frozen observation rather than a fresh read, so a retry produces the same sentence. Replay reproduces the **recorded** decision instead of deciding again — the transcript is append-once, so a second decision would contradict it forever. **Cancellation is the ending without an answer** — no resolution, no wake. One exception, decided during testing: a watch the *user* cancelled leaves a single neutral transcript line ("Stopped watching …"), keyed off the watch id so a retry can't repeat it. The other four reasons stay silent. **Email is opt-in and only a fired watch emails.** An expiry is narrated in the chat and nowhere else. Both gates (agent access, a configured email transport) are checked at subscribe time *and* again at delivery, and the subscription outcome is frozen on the ledger row so a retry replays it. Neither gate is a plan check. **One watch offer per turn.** The prompt and the renderer guard this independently — if the turn already proposed a watch card, the action button is dropped, because the card is the better affordance. Two eval cases pin the prompt side: exactly one offer with the line last and the button after it, and zero offers when the rendered card already carries one — deterministic assertions, over a real-model run. ## Testing Unit tests (vitest, testcontainers, no mocks) under `apps/webapp/test/dashboardAgentWatch*.test.ts` and `internal-packages/dashboard-agent/src/watch-*.test.ts` cover the invariants above: the 4-way check results and the freshness fences, identity/dedup and the submission ledger, queue-name resolution, the batch chain surviving a failed check, sweep boundaries and alert-once, tenancy and the watch token's scope, and the wording snapshot. The load-bearing ones were verified by control-breaking the guard first and checking the test goes red. Live-tested end to end against a local stack, following the guidebook: all ten watch kinds firing and expiring, cancellation, the email pair (a fired watch mails, an expired one does not), and watch recovery from a health report.
258 lines
10 KiB
TypeScript
258 lines
10 KiB
TypeScript
import { signUserActorToken } from "@trigger.dev/rbac";
|
|
import { TriggerClient } from "@trigger.dev/sdk";
|
|
import { chat } from "@trigger.dev/sdk/ai";
|
|
import { Counter } from "prom-client";
|
|
import { prisma } from "~/db.server";
|
|
import { env } from "~/env.server";
|
|
import { metricsRegister } from "~/metrics.server";
|
|
import { singleton } from "~/utils/singleton";
|
|
import { runStore } from "~/v3/runStore.server";
|
|
import { githubApp } from "./gitHub.server";
|
|
import { logger } from "./logger.server";
|
|
|
|
const TASK_ID = "dashboard-agent";
|
|
|
|
// The wake poll runs once a minute per visible tab. That trade-off is only defensible while it
|
|
// stays measurable, so count the requests. singleton: module-scope registration double-registers
|
|
// under dev HMR.
|
|
export const dashboardAgentWakeFeedCounter = singleton(
|
|
"dashboardAgentWakeFeedCounter",
|
|
() =>
|
|
new Counter({
|
|
name: "dashboard_agent_wake_feed_requests_total",
|
|
help: "Requests to the dashboard agent's wake feed",
|
|
registers: [metricsRegister],
|
|
})
|
|
);
|
|
|
|
// Read-only cap on the agent's delegated user-actor token. `read:apiKeys` is
|
|
// what lets it exchange the token for an env JWT (the gate on the exchange
|
|
// route); the rest scope the actual reads. No write/admin scopes, so even a
|
|
// leaked token can't mutate anything.
|
|
export const DASHBOARD_AGENT_UAT_CAP = [
|
|
"read:apiKeys",
|
|
"read:runs",
|
|
"read:deployments",
|
|
"read:environments",
|
|
"read:errors",
|
|
"read:query",
|
|
// Queue metrics ride on `read:query`, but a queue's own row — paused, depth, limit —
|
|
// is a `queues` read, and without it the agent can only see the metrics window.
|
|
"read:queues",
|
|
];
|
|
|
|
// Minted fresh on every turn (the `in` proxy injects it), so the lifetime only
|
|
// has to cover a single turn's tool calls. Short by design — a stale token in
|
|
// the agent's run payload expires quickly.
|
|
const DASHBOARD_AGENT_UAT_TTL_SECONDS = 10 * 60;
|
|
|
|
// The Trigger instance this webapp runs against — the same origin the agent
|
|
// task calls back to (as the user) for its read tools.
|
|
export function dashboardAgentApiOrigin(): string {
|
|
return env.API_ORIGIN ?? env.APP_ORIGIN;
|
|
}
|
|
|
|
// Mint a short-lived, read-only delegated token for the signed-in user. Self
|
|
// service from the dashboard session (never a PAT), so a user can only ever
|
|
// mint a token for themselves. The `in` proxy injects this into the turn's
|
|
// metadata so the token reaches the agent without ever touching the browser.
|
|
//
|
|
// Endpoints that bind something to one environment read `environmentId` off the token,
|
|
// so the agent can't name a different one in a request body.
|
|
export function mintDashboardAgentUserActorToken(
|
|
userId: string,
|
|
opts: { environmentId: string }
|
|
): Promise<string> {
|
|
return signUserActorToken(env.SESSION_SECRET, {
|
|
userId,
|
|
client: "dashboard-agent",
|
|
environmentId: opts.environmentId,
|
|
cap: DASHBOARD_AGENT_UAT_CAP,
|
|
expirationTime: Math.floor(Date.now() / 1000) + DASHBOARD_AGENT_UAT_TTL_SECONDS,
|
|
});
|
|
}
|
|
|
|
// The session is created in whatever env DASHBOARD_AGENT_SECRET_KEY belongs to.
|
|
// baseURL is the Trigger instance this webapp runs against (its own API origin).
|
|
function dashboardAgentConfig() {
|
|
const accessToken = env.DASHBOARD_AGENT_SECRET_KEY;
|
|
if (!accessToken) return null;
|
|
return { baseURL: dashboardAgentApiOrigin(), accessToken };
|
|
}
|
|
|
|
export function isDashboardAgentConfigured(): boolean {
|
|
return Boolean(env.DASHBOARD_AGENT_SECRET_KEY);
|
|
}
|
|
|
|
// Pins every agent session (and its continuation runs) to a deployed version
|
|
// when DASHBOARD_AGENT_VERSION is set; unset runs on the env's current version.
|
|
export function dashboardAgentTriggerConfig(): { lockToVersion: string } | undefined {
|
|
return env.DASHBOARD_AGENT_VERSION ? { lockToVersion: env.DASHBOARD_AGENT_VERSION } : undefined;
|
|
}
|
|
|
|
export async function startDashboardAgentSession(params: {
|
|
chatId: string;
|
|
clientData?: Record<string, unknown>;
|
|
}): Promise<{ publicAccessToken: string }> {
|
|
const config = dashboardAgentConfig();
|
|
if (!config) throw new Error("DASHBOARD_AGENT_SECRET_KEY is not set");
|
|
const startSession = chat.createStartSessionAction(TASK_ID, {
|
|
apiClient: config,
|
|
triggerConfig: dashboardAgentTriggerConfig(),
|
|
});
|
|
return startSession({ chatId: params.chatId, clientData: params.clientData });
|
|
}
|
|
|
|
export async function mintDashboardAgentToken(chatId: string): Promise<string> {
|
|
const config = dashboardAgentConfig();
|
|
if (!config) throw new Error("DASHBOARD_AGENT_SECRET_KEY is not set");
|
|
const client = new TriggerClient(config);
|
|
return client.auth.createPublicToken({
|
|
scopes: { read: { sessions: chatId }, write: { sessions: chatId } },
|
|
expirationTime: "1h",
|
|
});
|
|
}
|
|
|
|
// A signed, short-lived pointer to the project's connected repo at a commit. Only
|
|
// the URL crosses to the agent; the GitHub token stays here. The agent's code
|
|
// tools download + extract it on their own filesystem (see @internal/dashboard-agent).
|
|
export type DashboardAgentRepoSnapshot = {
|
|
tarballUrl: string;
|
|
owner: string;
|
|
repo: string;
|
|
sha: string;
|
|
defaultBranch?: string;
|
|
};
|
|
|
|
// The GitHub archive redirect URL is valid for a few minutes; cache the resolved
|
|
// pointer briefly so multi-turn chats don't re-mint a token + re-resolve on every
|
|
// message. Keyed by project + ref.
|
|
const repoSnapshotCache = new Map<
|
|
string,
|
|
{ snapshot: DashboardAgentRepoSnapshot; expiresAt: number }
|
|
>();
|
|
const REPO_SNAPSHOT_TTL_MS = 60_000;
|
|
const REPO_SNAPSHOT_MAX_ENTRIES = 1_000;
|
|
|
|
// Drop expired entries (key cardinality grows with each unique project + pinned
|
|
// SHA), then evict oldest-first if still over the cap, so the cache can't grow
|
|
// unbounded over a process lifetime.
|
|
function pruneRepoSnapshotCache(now = Date.now()) {
|
|
for (const [key, value] of repoSnapshotCache) {
|
|
if (value.expiresAt <= now) repoSnapshotCache.delete(key);
|
|
}
|
|
let overflow = repoSnapshotCache.size - REPO_SNAPSHOT_MAX_ENTRIES;
|
|
if (overflow <= 0) return;
|
|
for (const key of repoSnapshotCache.keys()) {
|
|
repoSnapshotCache.delete(key);
|
|
if (--overflow <= 0) break;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Resolve the code-mode repo snapshot for a project, or null when the GitHub App
|
|
* is disabled / no repo is connected (which keeps the agent in assistant mode).
|
|
*
|
|
* Mints a `contents:read` installation token scoped to the one repo, resolves the
|
|
* signed archive URL, and returns just that URL. The token never leaves the
|
|
* server. `opts.ref` pins a specific commit (run-SHA pinning); without it, the
|
|
* tracked prod branch (or the repo default) head is used.
|
|
*/
|
|
export async function resolveDashboardAgentRepoSnapshot(
|
|
projectId: string,
|
|
opts: { ref?: string } = {}
|
|
): Promise<DashboardAgentRepoSnapshot | null> {
|
|
if (!githubApp) return null;
|
|
|
|
// Cache per project + ref so HEAD and each pinned commit are cached separately.
|
|
const cacheKey = `${projectId}:${opts.ref ?? "HEAD"}`;
|
|
const cached = repoSnapshotCache.get(cacheKey);
|
|
if (cached && cached.expiresAt > Date.now()) return cached.snapshot;
|
|
if (cached) repoSnapshotCache.delete(cacheKey);
|
|
|
|
const connected = await prisma.connectedGithubRepository.findFirst({
|
|
where: { projectId },
|
|
select: {
|
|
branchTracking: true,
|
|
repository: {
|
|
select: {
|
|
fullName: true,
|
|
defaultBranch: true,
|
|
installation: { select: { appInstallationId: true } },
|
|
},
|
|
},
|
|
},
|
|
});
|
|
if (!connected) return null;
|
|
|
|
const [owner, repo] = connected.repository.fullName.split("/");
|
|
if (!owner || !repo) return null;
|
|
const installationId = Number(connected.repository.installation.appInstallationId);
|
|
const defaultBranch = connected.repository.defaultBranch;
|
|
const tracking = connected.branchTracking as { prod?: { branch?: string } } | null;
|
|
// An explicit 40-char commit SHA is used directly (run-SHA pinning); otherwise
|
|
// resolve the requested branch, the tracked prod branch, or the repo default.
|
|
const requested = opts.ref;
|
|
const isSha = !!requested && /^[0-9a-f]{40}$/i.test(requested);
|
|
const branchRef = requested && !isSha ? requested : tracking?.prod?.branch || defaultBranch;
|
|
|
|
try {
|
|
const octokit = await githubApp.getInstallationOctokit(installationId);
|
|
const sha = isSha
|
|
? requested!
|
|
: (await octokit.rest.repos.getBranch({ owner, repo, branch: branchRef })).data.commit.sha;
|
|
|
|
const token = await githubApp.octokit.rest.apps.createInstallationAccessToken({
|
|
installation_id: installationId,
|
|
repositories: [repo],
|
|
permissions: { contents: "read" },
|
|
});
|
|
|
|
// Resolve the signed archive URL without downloading the bytes server-side.
|
|
const redirect = await fetch(`https://api.github.com/repos/${owner}/${repo}/tarball/${sha}`, {
|
|
headers: {
|
|
Authorization: `Bearer ${token.data.token}`,
|
|
Accept: "application/vnd.github+json",
|
|
"User-Agent": "trigger-dashboard-agent",
|
|
},
|
|
redirect: "manual",
|
|
});
|
|
const tarballUrl = redirect.headers.get("location");
|
|
if (!tarballUrl) return null;
|
|
|
|
const snapshot: DashboardAgentRepoSnapshot = { tarballUrl, owner, repo, sha, defaultBranch };
|
|
pruneRepoSnapshotCache();
|
|
repoSnapshotCache.set(cacheKey, { snapshot, expiresAt: Date.now() + REPO_SNAPSHOT_TTL_MS });
|
|
return snapshot;
|
|
} catch (error) {
|
|
logger.error("Failed to resolve dashboard agent repo snapshot", { error, projectId });
|
|
return null;
|
|
}
|
|
}
|
|
|
|
// Map a run (by friendly id) to the commit its deployed version came from, for
|
|
// run-SHA pinning. A run locks to a BackgroundWorker (`lockedToVersionId`), whose
|
|
// WorkerDeployment carries the commit. Null for runs with no deployed version
|
|
// (e.g. dev runs), so the agent falls back to the branch head.
|
|
export async function resolveRunCommit(
|
|
environmentId: string,
|
|
runFriendlyId: string
|
|
): Promise<{ sha: string; version: string; dirty: boolean } | null> {
|
|
// Read-your-writes: a just-locked run's lockedToVersionId may not have replicated. Read the owning
|
|
// primary so a live, pinned run resolves its commit instead of silently falling back to branch head.
|
|
const run = await runStore.findRunOnPrimary(
|
|
{ friendlyId: runFriendlyId, runtimeEnvironmentId: environmentId },
|
|
{ select: { lockedToVersionId: true } }
|
|
);
|
|
if (!run?.lockedToVersionId) return null;
|
|
|
|
const deployment = await prisma.workerDeployment.findFirst({
|
|
where: { workerId: run.lockedToVersionId },
|
|
select: { commitSHA: true, version: true, git: true },
|
|
});
|
|
if (!deployment?.commitSHA) return null;
|
|
|
|
const dirty = (deployment.git as { dirty?: boolean } | null)?.dirty ?? false;
|
|
return { sha: deployment.commitSHA, version: deployment.version, dirty };
|
|
}
|