mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
feat: persist agent-pushed workspace context snapshots in coderd (#26145)
Replaces the v2.10 `PushContextState` stub with a real coderd write path. Phase 1 of the chat-side persistence story; nothing reads these rows yet. Follows [#25983](https://github.com/coder/coder/pull/25983) and unblocks [CODAGT-569](https://linear.app/codercom/issue/CODAGT-569/enable-agent-api-v210-pushcontextstate-bump-currentminor-wire-coderd). ## What ships ### Schema (`000517_workspace_agent_context.{up,down}.sql`) Two new tables plus `api_key_scope` enum extensions: - `workspace_agent_context_snapshots` (PK `workspace_agent_id` to `workspace_agents(id) ON DELETE CASCADE`): one row per agent, overwritten per push. Holds `version`, `schema_version`, `aggregate_hash`, `snapshot_error`, `received_at`. - `workspace_agent_context_resources` (PK `(workspace_agent_id, source)`): per-resource state. `body_kind` and `status` are `TEXT` + `CHECK` so adding new wire kinds (the RFC's reserved PLUGIN/HOOK/SUBAGENT/COMMAND) is a one-line CHECK update plus a Go switch case. ### SQLC queries (`coderd/database/queries/workspaceagentcontext.sql`) - `UpsertWorkspaceAgentContextSnapshot` - `UpsertWorkspaceAgentContextResource` - `DeleteStaleWorkspaceAgentContextResources` (delete-where-source-not-in) - `GetLatestWorkspaceAgentContextSnapshot` - `ListWorkspaceAgentContextResources` ### Handler (`coderd/agentapi/context.go`) `ContextAPI` is a new sub-API. `PushContextState`: 1. Rejects `schema_version > 1` with a non-`Unimplemented` error so a forward-incompatible agent fails loudly during rollout instead of slipping into the permanent fallback path the `Unimplemented` translation reserves for old coderd deployments. 2. Validates resources: no empty/duplicate sources, every variant maps to a known body kind, every status maps to a known enum value, the `Body` oneof is set (even when status is non-OK, mirroring the wire guarantee so coderd can attribute failures to a known kind). 3. Inside `Database.InTx`, reads the existing snapshot. If the push is not `initial` and `version` is not strictly greater, returns `accepted = false` and leaves stored state untouched. Otherwise upserts the snapshot row, upserts each resource, then runs the stale-source prune so the snapshot and resource rows always agree. 4. Returns `accepted = true` on success. Resource bodies are stored as `protojson(body oneof variant)` in `body JSONB` with `body_kind` as the discriminator. Adding a new field to an existing variant is zero work since `protojson` tolerates new fields; adding a new variant is a CHECK + switch case. ### RBAC + dbauthz - New `ResourceWorkspaceAgentContext` (Create/Read/Update/Delete). - New `SubjectTypeAgentContext` plus `subjectAgentContext` system role and `dbauthz.AsAgentContext` helper. The push handler elevates to this subject; the agent's own role does not get direct write access to the table. - New `workspace_agent_context:*` API key scopes registered in the enum migration; internal-only (not added to `externalLowLevel`). ### Audit These rows are agent-pushed state, not user-authored. They are intentionally not added to `AuditActionMap` and not enumerated in `enterprise/audit/table.go`, matching `boundary_logs`, `workspace_agent_memory_resource_monitor`, etc. `enterprise/audit` tests pass unchanged. ## Tests - `coderd/agentapi/context_test.go`: 12 subtests covering accepts/rejects (schema version, empty/duplicate source, unknown status, missing body), version semantics (stale dropped, same-version replay dropped, `initial=true` overwrites lower version), variant coverage, non-OK status persistence, and the empty-active-set prune case. - `coderd/database/dbauthz/dbauthz_test.go`: 5 `MethodTestSuite` cases covering the new queries. - `coderd/rbac/roles_test.go`: `WorkspaceAgentContext` permission row asserting no human role currently has access. - `coderd/database/migrations/testdata/fixtures/000517_workspace_agent_context.up.sql`: one snapshot + one resource per known body kind plus a non-OK status, so the migration test suite never lands with these tables empty. ## Out of scope (later phases) - Chat hydration (`chats.context_aggregate_hash`, `last_injected_context`). - Dirty-bit fan-out and `PUT /chats/{id}/context`. - Agent-side `POST /api/v0/context/resync` barrier and the `coder exp chat context` CLI. - `codersdk` chat-context wire types and the dashboard Sources drawer. - Removal of the chatd per-turn pull fallback. ## Compat property This is a pure write path. If anything here returns errors the agent's `RunPush` loop backs off, no chat behavior changes, and the workspace keeps behaving exactly like it did before v2.10. <details> <summary>Implementation plan and decision log</summary> Key design calls: 1. **Concurrency**: Accept iff `req.Initial || req.Version > existing.Version`. The strict RFC reading ("version comparison is authoritative") locks restarted agents out because their per-process counter resets to 1; honoring `initial=true` reflects the real reboot reality while still rejecting steady-state replays/out-of-order pushes. 2. **Body encoding**: `protojson` over the oneof variant body proto, stored in JSONB with `body_kind` discriminator. Structured at the API/Go layer, schema-tolerant at the storage layer, and Phase 2 readers round-trip back via `protojson.Unmarshal`. 3. **Schema version rejection**: returns a normal error, not `Unimplemented`. The agent's `RunPush` loop only short-circuits on `Unimplemented`; that escape hatch is reserved for old coderd deployments. A forward-incompatible agent should retry-and-back-off, not flip the connection into permanent fallback. 4. **Validation strictness**: empty sources, duplicate sources, `STATUS_UNSPECIFIED`, and missing `Body` oneof variants are rejected before any write so a misbehaving agent cannot poison the snapshot table. Phase 2 readers can trust every row maps to a known proto variant. </details> _This PR was authored by Coder Agents on Kyle Carberry's behalf._
This commit is contained in:
Generated
+273
-27
@@ -31114,6 +31114,214 @@ func (q *sqlQuerier) ValidateUserIDs(ctx context.Context, userIds []uuid.UUID) (
|
||||
return i, err
|
||||
}
|
||||
|
||||
const deleteStaleWorkspaceAgentContextResources = `-- name: DeleteStaleWorkspaceAgentContextResources :exec
|
||||
DELETE FROM workspace_agent_context_resources
|
||||
WHERE workspace_agent_id = $1
|
||||
AND NOT (source = ANY($2 :: text[]))
|
||||
`
|
||||
|
||||
type DeleteStaleWorkspaceAgentContextResourcesParams struct {
|
||||
WorkspaceAgentID uuid.UUID `db:"workspace_agent_id" json:"workspace_agent_id"`
|
||||
ActiveSources []string `db:"active_sources" json:"active_sources"`
|
||||
}
|
||||
|
||||
// Deletes any resources for the agent whose source is not in the
|
||||
// supplied active set. Atomic alongside the snapshot upsert so the
|
||||
// stored snapshot and resource rows always agree.
|
||||
func (q *sqlQuerier) DeleteStaleWorkspaceAgentContextResources(ctx context.Context, arg DeleteStaleWorkspaceAgentContextResourcesParams) error {
|
||||
_, err := q.db.ExecContext(ctx, deleteStaleWorkspaceAgentContextResources, arg.WorkspaceAgentID, pq.Array(arg.ActiveSources))
|
||||
return err
|
||||
}
|
||||
|
||||
const getLatestWorkspaceAgentContextSnapshot = `-- name: GetLatestWorkspaceAgentContextSnapshot :one
|
||||
SELECT workspace_agent_id, version, aggregate_hash, snapshot_error, received_at FROM workspace_agent_context_snapshots
|
||||
WHERE workspace_agent_id = $1
|
||||
`
|
||||
|
||||
func (q *sqlQuerier) GetLatestWorkspaceAgentContextSnapshot(ctx context.Context, workspaceAgentID uuid.UUID) (WorkspaceAgentContextSnapshot, error) {
|
||||
row := q.db.QueryRowContext(ctx, getLatestWorkspaceAgentContextSnapshot, workspaceAgentID)
|
||||
var i WorkspaceAgentContextSnapshot
|
||||
err := row.Scan(
|
||||
&i.WorkspaceAgentID,
|
||||
&i.Version,
|
||||
&i.AggregateHash,
|
||||
&i.SnapshotError,
|
||||
&i.ReceivedAt,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const listWorkspaceAgentContextResources = `-- name: ListWorkspaceAgentContextResources :many
|
||||
SELECT workspace_agent_id, source, body_kind, body, content_hash, size_bytes, status, error, source_path, created_at, updated_at FROM workspace_agent_context_resources
|
||||
WHERE workspace_agent_id = $1
|
||||
ORDER BY source ASC
|
||||
`
|
||||
|
||||
func (q *sqlQuerier) ListWorkspaceAgentContextResources(ctx context.Context, workspaceAgentID uuid.UUID) ([]WorkspaceAgentContextResource, error) {
|
||||
rows, err := q.db.QueryContext(ctx, listWorkspaceAgentContextResources, workspaceAgentID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []WorkspaceAgentContextResource
|
||||
for rows.Next() {
|
||||
var i WorkspaceAgentContextResource
|
||||
if err := rows.Scan(
|
||||
&i.WorkspaceAgentID,
|
||||
&i.Source,
|
||||
&i.BodyKind,
|
||||
&i.Body,
|
||||
&i.ContentHash,
|
||||
&i.SizeBytes,
|
||||
&i.Status,
|
||||
&i.Error,
|
||||
&i.SourcePath,
|
||||
&i.CreatedAt,
|
||||
&i.UpdatedAt,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const upsertWorkspaceAgentContextResource = `-- name: UpsertWorkspaceAgentContextResource :one
|
||||
INSERT INTO workspace_agent_context_resources (
|
||||
workspace_agent_id,
|
||||
source,
|
||||
body_kind,
|
||||
body,
|
||||
content_hash,
|
||||
size_bytes,
|
||||
status,
|
||||
error,
|
||||
source_path,
|
||||
created_at,
|
||||
updated_at
|
||||
) VALUES (
|
||||
$1,
|
||||
$2,
|
||||
$3,
|
||||
$4,
|
||||
$5,
|
||||
$6,
|
||||
$7,
|
||||
$8,
|
||||
$9,
|
||||
$10,
|
||||
$10
|
||||
)
|
||||
ON CONFLICT (workspace_agent_id, source) DO UPDATE SET
|
||||
body_kind = EXCLUDED.body_kind,
|
||||
body = EXCLUDED.body,
|
||||
content_hash = EXCLUDED.content_hash,
|
||||
size_bytes = EXCLUDED.size_bytes,
|
||||
status = EXCLUDED.status,
|
||||
error = EXCLUDED.error,
|
||||
source_path = EXCLUDED.source_path,
|
||||
updated_at = EXCLUDED.updated_at
|
||||
RETURNING workspace_agent_id, source, body_kind, body, content_hash, size_bytes, status, error, source_path, created_at, updated_at
|
||||
`
|
||||
|
||||
type UpsertWorkspaceAgentContextResourceParams struct {
|
||||
WorkspaceAgentID uuid.UUID `db:"workspace_agent_id" json:"workspace_agent_id"`
|
||||
Source string `db:"source" json:"source"`
|
||||
BodyKind WorkspaceAgentContextBodyKind `db:"body_kind" json:"body_kind"`
|
||||
Body json.RawMessage `db:"body" json:"body"`
|
||||
ContentHash []byte `db:"content_hash" json:"content_hash"`
|
||||
SizeBytes int64 `db:"size_bytes" json:"size_bytes"`
|
||||
Status WorkspaceAgentContextResourceStatus `db:"status" json:"status"`
|
||||
Error string `db:"error" json:"error"`
|
||||
SourcePath string `db:"source_path" json:"source_path"`
|
||||
Now time.Time `db:"now" json:"now"`
|
||||
}
|
||||
|
||||
func (q *sqlQuerier) UpsertWorkspaceAgentContextResource(ctx context.Context, arg UpsertWorkspaceAgentContextResourceParams) (WorkspaceAgentContextResource, error) {
|
||||
row := q.db.QueryRowContext(ctx, upsertWorkspaceAgentContextResource,
|
||||
arg.WorkspaceAgentID,
|
||||
arg.Source,
|
||||
arg.BodyKind,
|
||||
arg.Body,
|
||||
arg.ContentHash,
|
||||
arg.SizeBytes,
|
||||
arg.Status,
|
||||
arg.Error,
|
||||
arg.SourcePath,
|
||||
arg.Now,
|
||||
)
|
||||
var i WorkspaceAgentContextResource
|
||||
err := row.Scan(
|
||||
&i.WorkspaceAgentID,
|
||||
&i.Source,
|
||||
&i.BodyKind,
|
||||
&i.Body,
|
||||
&i.ContentHash,
|
||||
&i.SizeBytes,
|
||||
&i.Status,
|
||||
&i.Error,
|
||||
&i.SourcePath,
|
||||
&i.CreatedAt,
|
||||
&i.UpdatedAt,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const upsertWorkspaceAgentContextSnapshot = `-- name: UpsertWorkspaceAgentContextSnapshot :one
|
||||
INSERT INTO workspace_agent_context_snapshots (
|
||||
workspace_agent_id,
|
||||
version,
|
||||
aggregate_hash,
|
||||
snapshot_error,
|
||||
received_at
|
||||
) VALUES (
|
||||
$1,
|
||||
$2,
|
||||
$3,
|
||||
$4,
|
||||
$5
|
||||
)
|
||||
ON CONFLICT (workspace_agent_id) DO UPDATE SET
|
||||
version = EXCLUDED.version,
|
||||
aggregate_hash = EXCLUDED.aggregate_hash,
|
||||
snapshot_error = EXCLUDED.snapshot_error,
|
||||
received_at = EXCLUDED.received_at
|
||||
RETURNING workspace_agent_id, version, aggregate_hash, snapshot_error, received_at
|
||||
`
|
||||
|
||||
type UpsertWorkspaceAgentContextSnapshotParams struct {
|
||||
WorkspaceAgentID uuid.UUID `db:"workspace_agent_id" json:"workspace_agent_id"`
|
||||
Version int64 `db:"version" json:"version"`
|
||||
AggregateHash []byte `db:"aggregate_hash" json:"aggregate_hash"`
|
||||
SnapshotError string `db:"snapshot_error" json:"snapshot_error"`
|
||||
ReceivedAt time.Time `db:"received_at" json:"received_at"`
|
||||
}
|
||||
|
||||
func (q *sqlQuerier) UpsertWorkspaceAgentContextSnapshot(ctx context.Context, arg UpsertWorkspaceAgentContextSnapshotParams) (WorkspaceAgentContextSnapshot, error) {
|
||||
row := q.db.QueryRowContext(ctx, upsertWorkspaceAgentContextSnapshot,
|
||||
arg.WorkspaceAgentID,
|
||||
arg.Version,
|
||||
arg.AggregateHash,
|
||||
arg.SnapshotError,
|
||||
arg.ReceivedAt,
|
||||
)
|
||||
var i WorkspaceAgentContextSnapshot
|
||||
err := row.Scan(
|
||||
&i.WorkspaceAgentID,
|
||||
&i.Version,
|
||||
&i.AggregateHash,
|
||||
&i.SnapshotError,
|
||||
&i.ReceivedAt,
|
||||
)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getWorkspaceAgentDevcontainersByAgentID = `-- name: GetWorkspaceAgentDevcontainersByAgentID :many
|
||||
SELECT
|
||||
id, workspace_agent_id, created_at, workspace_folder, config_path, name, subagent_id
|
||||
@@ -31800,16 +32008,30 @@ func (q *sqlQuerier) DeleteOldWorkspaceAgentLogs(ctx context.Context, threshold
|
||||
}
|
||||
|
||||
const deleteWorkspaceSubAgentByID = `-- name: DeleteWorkspaceSubAgentByID :exec
|
||||
UPDATE
|
||||
workspace_agents
|
||||
SET
|
||||
deleted = TRUE
|
||||
WHERE
|
||||
id = $1
|
||||
AND parent_id IS NOT NULL
|
||||
AND deleted = FALSE
|
||||
WITH soft_deleted_agents AS (
|
||||
UPDATE workspace_agents
|
||||
SET deleted = TRUE
|
||||
WHERE id = $1
|
||||
AND parent_id IS NOT NULL
|
||||
AND deleted = FALSE
|
||||
RETURNING id
|
||||
), purged_context_resources AS (
|
||||
DELETE FROM workspace_agent_context_resources
|
||||
WHERE workspace_agent_id IN (SELECT id FROM soft_deleted_agents)
|
||||
)
|
||||
DELETE FROM workspace_agent_context_snapshots
|
||||
WHERE workspace_agent_id IN (SELECT id FROM soft_deleted_agents)
|
||||
`
|
||||
|
||||
// Soft-deletes a single sub-agent (a child agent such as a devcontainer
|
||||
// agent). Called from the DeleteSubAgent RPC when a sub-agent is torn
|
||||
// down, which can happen mid-build without a full workspace rebuild.
|
||||
//
|
||||
// Agent context rows are hard-deleted for the same reason as in
|
||||
// SoftDeletePriorWorkspaceAgents: they only describe live agents, the
|
||||
// rebuild-time soft-delete queries skip already-deleted agents, and
|
||||
// agents are never hard-deleted, so the rows would otherwise orphan
|
||||
// forever.
|
||||
func (q *sqlQuerier) DeleteWorkspaceSubAgentByID(ctx context.Context, id uuid.UUID) error {
|
||||
_, err := q.db.ExecContext(ctx, deleteWorkspaceSubAgentByID, id)
|
||||
return err
|
||||
@@ -33391,17 +33613,25 @@ func (q *sqlQuerier) InsertWorkspaceAgentScriptTimings(ctx context.Context, arg
|
||||
}
|
||||
|
||||
const softDeletePriorWorkspaceAgents = `-- name: SoftDeletePriorWorkspaceAgents :exec
|
||||
UPDATE workspace_agents
|
||||
SET deleted = TRUE
|
||||
WHERE id IN (
|
||||
SELECT wa.id
|
||||
FROM workspace_agents wa
|
||||
JOIN workspace_resources wr ON wr.id = wa.resource_id
|
||||
JOIN workspace_builds wb ON wb.job_id = wr.job_id
|
||||
WHERE wb.workspace_id = $1
|
||||
AND wb.id <> $2
|
||||
AND wa.deleted = FALSE
|
||||
WITH soft_deleted_agents AS (
|
||||
UPDATE workspace_agents
|
||||
SET deleted = TRUE
|
||||
WHERE id IN (
|
||||
SELECT wa.id
|
||||
FROM workspace_agents wa
|
||||
JOIN workspace_resources wr ON wr.id = wa.resource_id
|
||||
JOIN workspace_builds wb ON wb.job_id = wr.job_id
|
||||
WHERE wb.workspace_id = $1
|
||||
AND wb.id <> $2
|
||||
AND wa.deleted = FALSE
|
||||
)
|
||||
RETURNING id
|
||||
), purged_context_resources AS (
|
||||
DELETE FROM workspace_agent_context_resources
|
||||
WHERE workspace_agent_id IN (SELECT id FROM soft_deleted_agents)
|
||||
)
|
||||
DELETE FROM workspace_agent_context_snapshots
|
||||
WHERE workspace_agent_id IN (SELECT id FROM soft_deleted_agents)
|
||||
`
|
||||
|
||||
type SoftDeletePriorWorkspaceAgentsParams struct {
|
||||
@@ -33414,22 +33644,35 @@ type SoftDeletePriorWorkspaceAgentsParams struct {
|
||||
// provisionerdserver when a workspace build completes, after the new
|
||||
// build's agents have been inserted, so running agents are not
|
||||
// deleted while a build is still queued or provisioning.
|
||||
//
|
||||
// Agent context rows (workspace_agent_context_snapshots and
|
||||
// workspace_agent_context_resources) only describe live agents, and
|
||||
// agents are never un-deleted, so they are hard-deleted here instead
|
||||
// of accumulating alongside the soft-deleted agent rows.
|
||||
func (q *sqlQuerier) SoftDeletePriorWorkspaceAgents(ctx context.Context, arg SoftDeletePriorWorkspaceAgentsParams) error {
|
||||
_, err := q.db.ExecContext(ctx, softDeletePriorWorkspaceAgents, arg.WorkspaceID, arg.CurrentBuildID)
|
||||
return err
|
||||
}
|
||||
|
||||
const softDeleteWorkspaceAgentsByWorkspaceID = `-- name: SoftDeleteWorkspaceAgentsByWorkspaceID :exec
|
||||
UPDATE workspace_agents
|
||||
SET deleted = TRUE
|
||||
WHERE id IN (
|
||||
SELECT wa.id
|
||||
FROM workspace_agents wa
|
||||
JOIN workspace_resources wr ON wr.id = wa.resource_id
|
||||
JOIN workspace_builds wb ON wb.job_id = wr.job_id
|
||||
WHERE wb.workspace_id = $1
|
||||
AND wa.deleted = FALSE
|
||||
WITH soft_deleted_agents AS (
|
||||
UPDATE workspace_agents
|
||||
SET deleted = TRUE
|
||||
WHERE id IN (
|
||||
SELECT wa.id
|
||||
FROM workspace_agents wa
|
||||
JOIN workspace_resources wr ON wr.id = wa.resource_id
|
||||
JOIN workspace_builds wb ON wb.job_id = wr.job_id
|
||||
WHERE wb.workspace_id = $1
|
||||
AND wa.deleted = FALSE
|
||||
)
|
||||
RETURNING id
|
||||
), purged_context_resources AS (
|
||||
DELETE FROM workspace_agent_context_resources
|
||||
WHERE workspace_agent_id IN (SELECT id FROM soft_deleted_agents)
|
||||
)
|
||||
DELETE FROM workspace_agent_context_snapshots
|
||||
WHERE workspace_agent_id IN (SELECT id FROM soft_deleted_agents)
|
||||
`
|
||||
|
||||
// Marks every non-deleted agent belonging to the given workspace as
|
||||
@@ -33437,6 +33680,9 @@ WHERE id IN (
|
||||
// itself is soft-deleted, so the agent instance-identity auth path
|
||||
// (which filters on workspace_agents.deleted) doesn't keep seeing
|
||||
// orphaned rows.
|
||||
//
|
||||
// Agent context rows are hard-deleted for the same reason as in
|
||||
// SoftDeletePriorWorkspaceAgents.
|
||||
func (q *sqlQuerier) SoftDeleteWorkspaceAgentsByWorkspaceID(ctx context.Context, workspaceID uuid.UUID) error {
|
||||
_, err := q.db.ExecContext(ctx, softDeleteWorkspaceAgentsByWorkspaceID, workspaceID)
|
||||
return err
|
||||
|
||||
Reference in New Issue
Block a user