fix(office): stop config import from clobbering concurrent row edits (#2826)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
nova28
2026-08-20 05:49:11 +08:00
committed by GitHub
parent 0630a02841
commit c24a16fcaf
13 changed files with 843 additions and 32 deletions
+32 -24
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// PreviewImport diffs a bundle against the current workspace state.
@@ -170,13 +171,15 @@ func (s *ConfigService) applyAgents(
}
for _, cfg := range incoming {
if agent, ok := byName[cfg.Name]; ok {
agent.Role = models.AgentRole(cfg.Role)
agent.Icon = cfg.Icon
agent.BudgetMonthlyCents = cfg.BudgetMonthlyCents
agent.MaxConcurrentSessions = cfg.MaxConcurrentSessions
agent.DesiredSkills = cfg.DesiredSkills
agent.ExecutorPreference = cfg.ExecutorPreference
if err := s.repo.UpdateAgentInstance(ctx, agent); err != nil {
fields := sqlite.AgentInstanceConfigFields{
Role: cfg.Role,
Icon: cfg.Icon,
BudgetMonthlyCents: cfg.BudgetMonthlyCents,
MaxConcurrentSessions: cfg.MaxConcurrentSessions,
DesiredSkills: cfg.DesiredSkills,
ExecutorPreference: cfg.ExecutorPreference,
}
if err := s.repo.UpdateAgentInstanceConfigFields(ctx, agent.ID, fields); err != nil {
return err
}
result.UpdatedCount++
@@ -257,8 +260,7 @@ func (s *ConfigService) applyAgentReportsTo(
if agent.ReportsTo == reportsTo {
continue
}
agent.ReportsTo = reportsTo
if err := s.repo.UpdateAgentInstance(ctx, agent); err != nil {
if err := s.repo.UpdateAgentReportsTo(ctx, agent.ID, reportsTo); err != nil {
return fmt.Errorf("resolve reports_to: update %q: %w", agent.Name, err)
}
}
@@ -353,11 +355,13 @@ func (s *ConfigService) applySkills(
}
for _, cfg := range incoming {
if skill, ok := bySlug[cfg.Slug]; ok {
skill.Name = cfg.Name
skill.Description = cfg.Description
skill.SourceType = models.SkillSourceType(cfg.SourceType)
skill.Content = cfg.Content
if err := s.repo.UpdateSkill(ctx, skill); err != nil {
fields := sqlite.SkillConfigFields{
Name: cfg.Name,
Description: cfg.Description,
SourceType: models.SkillSourceType(cfg.SourceType),
Content: cfg.Content,
}
if err := s.repo.UpdateSkillConfigFields(ctx, skill.ID, fields); err != nil {
return err
}
result.UpdatedCount++
@@ -392,10 +396,12 @@ func (s *ConfigService) applyRoutines(
}
for _, cfg := range incoming {
if routine, ok := byName[cfg.Name]; ok {
routine.Description = cfg.Description
routine.TaskTemplate = cfg.TaskTemplate
routine.ConcurrencyPolicy = models.RoutineConcurrencyPolicy(cfg.ConcurrencyPolicy)
if err := s.repo.UpdateRoutine(ctx, routine); err != nil {
fields := sqlite.RoutineConfigFields{
Description: cfg.Description,
TaskTemplate: cfg.TaskTemplate,
ConcurrencyPolicy: models.RoutineConcurrencyPolicy(cfg.ConcurrencyPolicy),
}
if err := s.repo.UpdateRoutineConfigFields(ctx, routine.ID, fields); err != nil {
return err
}
result.UpdatedCount++
@@ -430,12 +436,14 @@ func (s *ConfigService) applyProjects(
}
for _, cfg := range incoming {
if project, ok := byName[cfg.Name]; ok {
project.Description = cfg.Description
project.Color = cfg.Color
project.BudgetCents = cfg.BudgetCents
project.Repositories = cfg.Repositories
project.ExecutorConfig = cfg.ExecutorConfig
if err := s.repo.UpdateProject(ctx, project); err != nil {
fields := sqlite.ProjectConfigFields{
Description: cfg.Description,
Color: cfg.Color,
BudgetCents: cfg.BudgetCents,
Repositories: cfg.Repositories,
ExecutorConfig: cfg.ExecutorConfig,
}
if err := s.repo.UpdateProjectConfigFields(ctx, project.ID, fields); err != nil {
return err
}
result.UpdatedCount++
@@ -0,0 +1,239 @@
package config
import (
"context"
"fmt"
"strings"
"testing"
)
// These tests are the regression coverage for the full-row read-modify-write
// bug: applyAgents/applySkills/applyRoutines/applyProjects used to read a
// full row via List*, mutate only the fields the import owns on that struct,
// and hand the *whole* struct back to a full-row Update*. Any column that a
// genuinely concurrent writer (UpdateAgentStatusFields, the agent runtime,
// direct API edits) changed between the List read and the Update write was
// silently reverted to its stale snapshot value.
//
// Each apply* function calls List* exactly once, then loops over the
// incoming bundle entries and issues one Update* per matched row, in bundle
// order. That gives a real window between the single List call and the
// Update call for any row after the first: a write landing in that window is
// exactly what production concurrency looks like (a budget-cap pause, a
// direct API edit, another import) racing an in-flight import.
//
// Each test below drives that window through ConfigService.ApplyImport
// itself (not the repository methods directly - those already have their
// own coverage in internal/office/repository/sqlite). A bundle with two
// entities puts the first entity's Update* first in the loop; a SQL trigger
// fires when that first Update* statement runs and, in response, writes an
// unowned column on the *second* entity - landing strictly between the
// single List call (which read the second entity's pre-write state) and the
// second entity's own Update* call later in the same loop. That is the
// production race window, reproduced deterministically instead of with
// goroutines and a timing hope.
//
// Before the fix, the second entity's Update* call used the row List*
// captured before the trigger fired - the mutated-in-place stale struct -
// and a full-row Update* silently reverted the trigger's write. After the
// fix, Update*ConfigFields never references the unowned column at the SQL
// level, so staleness of the rest of the struct cannot matter.
// installRaceTrigger creates a same-table trigger that fires when the row
// identified by triggerID is updated, and in response writes value into
// column on the row identified by targetID. Used to land a "concurrent"
// write strictly inside an apply* function's List-once-then-loop window: the
// trigger fires off the *first* entity's Update* statement and mutates the
// *second* entity's unowned column before the loop reaches that row's own
// Update* call.
func installRaceTrigger(t *testing.T, e *testEnv, table, triggerID, targetID, column, value string) {
t.Helper()
stmt := fmt.Sprintf(`
CREATE TRIGGER race_%[1]s
AFTER UPDATE ON %[1]s
WHEN NEW.id = %[2]s
BEGIN
UPDATE %[1]s SET %[3]s = %[4]s WHERE id = %[5]s;
END;
`, table, sqlLit(triggerID), column, sqlLit(value), sqlLit(targetID))
if _, err := e.db.Exec(stmt); err != nil {
t.Fatalf("install race trigger on %s: %v", table, err)
}
}
// installReportsToRaceTrigger lands a status update after the reports_to
// resolver reads all agents but before it updates a later agent. The trigger
// does not fire during the first config-field pass because that pass does not
// include reports_to in its UPDATE statement.
func installReportsToRaceTrigger(t *testing.T, e *testEnv, triggerID, targetID string) {
t.Helper()
stmt := fmt.Sprintf(`
CREATE TRIGGER race_agent_reports_to
AFTER UPDATE OF reports_to ON agent_profiles
WHEN NEW.id = %[1]s
BEGIN
UPDATE agent_profiles SET status = 'paused' WHERE id = %[2]s;
END;
`, sqlLit(triggerID), sqlLit(targetID))
if _, err := e.db.Exec(stmt); err != nil {
t.Fatalf("install reports_to race trigger: %v", err)
}
}
// sqlLit renders s as a single-quoted SQL string literal. Trigger bodies
// cannot take bind parameters, so the id/value operands have to be spliced
// into the statement text; every value here comes from uuid.New().String()
// or a test-chosen literal, but the escaping is done properly anyway rather
// than assumed safe.
func sqlLit(s string) string {
return "'" + strings.ReplaceAll(s, "'", "''") + "'"
}
// TestApplyImport_Agents_PreservesConcurrentWriteBetweenListAndUpdate
// reproduces the triage receipt: a budget-cap pause (UpdateAgentStatusFields,
// the real concurrent writer used by internal/office/costs/budgets.go)
// landing between applyAgents' single List call and its Update call for a
// later row must survive, instead of being reverted to the stale "idle"
// status that List captured before the write happened.
func TestApplyImport_Agents_PreservesConcurrentWriteBetweenListAndUpdate(t *testing.T) {
e := newTestEnv(t)
ctx := context.Background()
first := seedAgent(t, e, testWorkspaceID, "alpha")
second := seedAgent(t, e, testWorkspaceID, "beta")
installRaceTrigger(t, e, "agent_profiles", first.ID, second.ID, "status", "paused")
bundle := &ConfigBundle{
Agents: []AgentConfig{
{Name: "alpha", Role: "cto", BudgetMonthlyCents: 100, MaxConcurrentSessions: 1},
{Name: "beta", Role: "cfo", Icon: "💰", BudgetMonthlyCents: 200, MaxConcurrentSessions: 2},
},
}
if _, err := e.svc.ApplyImport(ctx, testWorkspaceID, bundle); err != nil {
t.Fatalf("ApplyImport: %v", err)
}
got := agentByName(t, e, testWorkspaceID, "beta")
assertEqual(t, "status survives concurrent write during import", string(got.Status), "paused")
// The owned fields must still have landed - a scoped update that writes
// nothing would also make the assertion above pass for the wrong reason.
assertEqual(t, "role updated", string(got.Role), "cfo")
assertEqual(t, "icon updated", got.Icon, "💰")
assertEqual(t, "budget updated", got.BudgetMonthlyCents, 200)
assertEqual(t, "max sessions updated", got.MaxConcurrentSessions, 2)
}
func TestApplyImport_Agents_PreservesConcurrentWriteDuringReportsToUpdate(t *testing.T) {
e := newTestEnv(t)
ctx := context.Background()
first := seedAgent(t, e, testWorkspaceID, "alpha")
second := seedAgent(t, e, testWorkspaceID, "beta")
manager := seedAgent(t, e, testWorkspaceID, "lead")
installReportsToRaceTrigger(t, e, first.ID, second.ID)
bundle := &ConfigBundle{Agents: []AgentConfig{
{Name: "alpha", Role: "engineer", ReportsTo: "lead", MaxConcurrentSessions: 1},
{Name: "beta", Role: "engineer", ReportsTo: "lead", MaxConcurrentSessions: 1},
{Name: "lead", Role: "cto", MaxConcurrentSessions: 1},
}}
if _, err := e.svc.ApplyImport(ctx, testWorkspaceID, bundle); err != nil {
t.Fatalf("ApplyImport: %v", err)
}
gotFirst := agentByName(t, e, testWorkspaceID, "alpha")
gotSecond := agentByName(t, e, testWorkspaceID, "beta")
assertEqual(t, "first reports_to updated", gotFirst.ReportsTo, manager.ID)
assertEqual(t, "second reports_to updated", gotSecond.ReportsTo, manager.ID)
assertEqual(t, "status survives concurrent write during reports_to update", string(gotSecond.Status), "paused")
}
// TestApplyImport_Skills_PreservesConcurrentWriteBetweenListAndUpdate mirrors
// the agent case for skills: approval_state is not a column applySkills
// owns, so a write landing between applyImport's List call and a later
// row's Update call must survive.
func TestApplyImport_Skills_PreservesConcurrentWriteBetweenListAndUpdate(t *testing.T) {
e := newTestEnv(t)
ctx := context.Background()
first := seedSkill(t, e, testWorkspaceID, "alpha")
second := seedSkill(t, e, testWorkspaceID, "beta")
installRaceTrigger(t, e, "office_skills", first.ID, second.ID, "approval_state", "approved")
bundle := &ConfigBundle{
Skills: []SkillConfig{
{Name: "Alpha", Slug: "alpha", Description: "first", SourceType: "inline", Content: "# Alpha\n"},
{Name: "Beta", Slug: "beta", Description: "second", SourceType: "inline", Content: "# Beta\n"},
},
}
if _, err := e.svc.ApplyImport(ctx, testWorkspaceID, bundle); err != nil {
t.Fatalf("ApplyImport: %v", err)
}
got := skillBySlug(t, e, testWorkspaceID, "beta")
assertEqual(t, "approval_state survives concurrent write during import", string(got.ApprovalState), "approved")
assertEqual(t, "name updated", got.Name, "Beta")
assertEqual(t, "description updated", got.Description, "second")
assertEqual(t, "content updated", got.Content, "# Beta\n")
}
// TestApplyImport_Routines_PreservesConcurrentWriteBetweenListAndUpdate
// mirrors the agent case for routines: status is not a column applyRoutines
// owns.
func TestApplyImport_Routines_PreservesConcurrentWriteBetweenListAndUpdate(t *testing.T) {
e := newTestEnv(t)
ctx := context.Background()
first := seedRoutine(t, e, testWorkspaceID, "alpha")
second := seedRoutine(t, e, testWorkspaceID, "beta")
installRaceTrigger(t, e, "office_routines", first.ID, second.ID, "status", "paused")
bundle := &ConfigBundle{
Routines: []RoutineConfig{
{Name: "alpha", Description: "first", TaskTemplate: "do the first thing", ConcurrencyPolicy: "skip"},
{Name: "beta", Description: "second", TaskTemplate: "do the second thing", ConcurrencyPolicy: "always_create"},
},
}
if _, err := e.svc.ApplyImport(ctx, testWorkspaceID, bundle); err != nil {
t.Fatalf("ApplyImport: %v", err)
}
got := routineByName(t, e, testWorkspaceID, "beta")
assertEqual(t, "status survives concurrent write during import", got.Status, "paused")
assertEqual(t, "description updated", got.Description, "second")
assertEqual(t, "task template updated", got.TaskTemplate, "do the second thing")
assertEqual(t, "concurrency policy updated", string(got.ConcurrencyPolicy), "always_create")
}
// TestApplyImport_Projects_PreservesConcurrentWriteBetweenListAndUpdate
// mirrors the agent case for projects: status is not a column applyProjects
// owns.
func TestApplyImport_Projects_PreservesConcurrentWriteBetweenListAndUpdate(t *testing.T) {
e := newTestEnv(t)
ctx := context.Background()
first := seedProject(t, e, testWorkspaceID, "alpha")
second := seedProject(t, e, testWorkspaceID, "beta")
installRaceTrigger(t, e, "office_projects", first.ID, second.ID, "status", "on_hold")
bundle := &ConfigBundle{
Projects: []ProjectConfig{
{Name: "alpha", Description: "first", Color: "#ff0000", BudgetCents: 100},
{Name: "beta", Description: "second", Color: "#00ff00", BudgetCents: 200, Repositories: `["kdlbs/other"]`},
},
}
if _, err := e.svc.ApplyImport(ctx, testWorkspaceID, bundle); err != nil {
t.Fatalf("ApplyImport: %v", err)
}
got := projectByName(t, e, testWorkspaceID, "beta")
assertEqual(t, "status survives concurrent write during import", string(got.Status), "on_hold")
assertEqual(t, "description updated", got.Description, "second")
assertEqual(t, "color updated", got.Color, "#00ff00")
assertEqual(t, "budget updated", got.BudgetCents, 200)
assertEqual(t, "repositories updated", got.Repositories, `["kdlbs/other"]`)
}
@@ -0,0 +1,13 @@
package models
import (
"crypto/sha256"
"encoding/hex"
)
// SkillPackageContentHash returns the identity of the content and package
// metadata that determine a materialized skill package.
func SkillPackageContentHash(content, fileInventory, sourceLocator string) string {
sum := sha256.Sum256([]byte(content + "\x00" + fileInventory + "\x00" + sourceLocator))
return hex.EncodeToString(sum[:])
}
@@ -269,6 +269,55 @@ func (r *Repository) UpdateAgentInstance(ctx context.Context, agent *models.Agen
return err
}
// AgentInstanceConfigFields is the subset of agent_profiles columns a
// config import owns (see UpdateAgentInstanceConfigFields).
type AgentInstanceConfigFields struct {
Role string
Icon string
BudgetMonthlyCents int
MaxConcurrentSessions int
DesiredSkills string
ExecutorPreference string
}
// UpdateAgentInstanceConfigFields updates only the columns a config import
// owns (role, icon, budget, max concurrent sessions, desired skills,
// executor preference). Unlike UpdateAgentInstance, it leaves status,
// pause_reason, settings, last_run_finished_at, and every other column at
// whatever a concurrent writer (UpdateAgentStatusFields, UpdateAgentSettings,
// the agent runtime, direct API edits) set them to, instead of reverting
// them to a stale read-then-write snapshot.
func (r *Repository) UpdateAgentInstanceConfigFields(
ctx context.Context, id string, fields AgentInstanceConfigFields,
) error {
desiredSkills := normalizeAgentJSONArray(fields.DesiredSkills)
now := time.Now().UTC()
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE agent_profiles SET
role = ?, icon = ?, budget_monthly_cents = ?,
max_concurrent_sessions = ?, desired_skills = ?,
executor_preference = ?, updated_at = ?
WHERE id = ? AND `+agentInstanceFilter+`
`), fields.Role, fields.Icon, fields.BudgetMonthlyCents,
fields.MaxConcurrentSessions, desiredSkills,
fields.ExecutorPreference, now, id)
return err
}
// UpdateAgentReportsTo updates only an agent's reporting relationship. Config
// import resolves this field in a second pass, after all agent IDs are known.
// A narrow update prevents that pass from reverting runtime-owned fields from
// the stale agent snapshot used for name resolution.
func (r *Repository) UpdateAgentReportsTo(ctx context.Context, id, reportsTo string) error {
now := time.Now().UTC()
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE agent_profiles
SET reports_to = ?, updated_at = ?
WHERE id = ? AND `+agentInstanceFilter+`
`), reportsTo, now, id)
return err
}
// UpdateAgentSettings persists the agent_profiles.settings JSON blob for
// an agent. Used by onboarding to seed routing.tier_source / .provider_
// order_source markers on the freshly created CEO agent so the routing
@@ -0,0 +1,137 @@
package sqlite_test
import (
"context"
"testing"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// TestUpdateAgentInstanceConfigFields_UpdatesOnlyOwnedColumns proves the
// column-scoped writer touches only the six columns a config import owns
// and leaves every other column - in particular status and pause_reason,
// which a concurrent writer like UpdateAgentStatusFields owns - exactly as
// they were.
func TestUpdateAgentInstanceConfigFields_UpdatesOnlyOwnedColumns(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
agent := fullAgentInstance("agent-1", "ws-1", "ada")
if err := repo.CreateAgentInstance(ctx, agent); err != nil {
t.Fatalf("CreateAgentInstance: %v", err)
}
// CreateAgentInstance always writes settings='{}'; set a real value
// through the column's real owner (UpdateAgentSettings) so this test
// can prove UpdateAgentInstanceConfigFields leaves it alone.
const wantSettings = `{"routing":{"tier_source":"explicit"}}`
if err := repo.UpdateAgentSettings(ctx, agent.ID, wantSettings); err != nil {
t.Fatalf("UpdateAgentSettings: %v", err)
}
fields := sqlite.AgentInstanceConfigFields{
Role: "cfo",
Icon: "💼",
BudgetMonthlyCents: 9900,
MaxConcurrentSessions: 5,
DesiredSkills: `["finance"]`,
ExecutorPreference: "sprites",
}
if err := repo.UpdateAgentInstanceConfigFields(ctx, agent.ID, fields); err != nil {
t.Fatalf("UpdateAgentInstanceConfigFields: %v", err)
}
rows, err := repo.ListAgentInstances(ctx, "ws-1")
if err != nil {
t.Fatalf("ListAgentInstances: %v", err)
}
if len(rows) != 1 {
t.Fatalf("rows = %d, want 1", len(rows))
}
got := rows[0]
// Owned columns landed.
if string(got.Role) != fields.Role {
t.Errorf("Role = %q, want %q", got.Role, fields.Role)
}
if got.Icon != fields.Icon {
t.Errorf("Icon = %q, want %q", got.Icon, fields.Icon)
}
if got.BudgetMonthlyCents != fields.BudgetMonthlyCents {
t.Errorf("BudgetMonthlyCents = %d, want %d", got.BudgetMonthlyCents, fields.BudgetMonthlyCents)
}
if got.MaxConcurrentSessions != fields.MaxConcurrentSessions {
t.Errorf("MaxConcurrentSessions = %d, want %d", got.MaxConcurrentSessions, fields.MaxConcurrentSessions)
}
if got.DesiredSkills != fields.DesiredSkills {
t.Errorf("DesiredSkills = %q, want %q", got.DesiredSkills, fields.DesiredSkills)
}
if got.ExecutorPreference != fields.ExecutorPreference {
t.Errorf("ExecutorPreference = %q, want %q", got.ExecutorPreference, fields.ExecutorPreference)
}
// Everything else stayed exactly as seeded.
if string(got.Status) != string(agent.Status) {
t.Errorf("Status changed: got %q, want %q", got.Status, agent.Status)
}
if got.PauseReason != agent.PauseReason {
t.Errorf("PauseReason changed: got %q, want %q", got.PauseReason, agent.PauseReason)
}
if got.Settings != wantSettings {
t.Errorf("Settings changed: got %q, want %q", got.Settings, wantSettings)
}
if got.Model != agent.Model {
t.Errorf("Model changed: got %q, want %q", got.Model, agent.Model)
}
if got.ReportsTo != agent.ReportsTo {
t.Errorf("ReportsTo changed: got %q, want %q", got.ReportsTo, agent.ReportsTo)
}
if got.ConsecutiveFailures != agent.ConsecutiveFailures {
t.Errorf("ConsecutiveFailures changed: got %d, want %d", got.ConsecutiveFailures, agent.ConsecutiveFailures)
}
if got.CooldownSec != agent.CooldownSec {
t.Errorf("CooldownSec changed: got %d, want %d", got.CooldownSec, agent.CooldownSec)
}
}
// TestUpdateAgentReportsTo_UpdatesOnlyReportsTo proves that the second config
// import pass cannot overwrite state that belongs to the agent runtime.
func TestUpdateAgentReportsTo_UpdatesOnlyReportsTo(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
agent := fullAgentInstance("agent-1", "ws-1", "ada")
if err := repo.CreateAgentInstance(ctx, agent); err != nil {
t.Fatalf("CreateAgentInstance: %v", err)
}
const wantSettings = `{"routing":{"tier_source":"explicit"}}`
if err := repo.UpdateAgentSettings(ctx, agent.ID, wantSettings); err != nil {
t.Fatalf("UpdateAgentSettings: %v", err)
}
if err := repo.UpdateAgentStatusFields(ctx, agent.ID, "paused", "manual pause"); err != nil {
t.Fatalf("UpdateAgentStatusFields: %v", err)
}
if err := repo.UpdateAgentReportsTo(ctx, agent.ID, "manager-1"); err != nil {
t.Fatalf("UpdateAgentReportsTo: %v", err)
}
got, err := repo.GetAgentInstance(ctx, agent.ID)
if err != nil {
t.Fatalf("GetAgentInstance: %v", err)
}
if got.ReportsTo != "manager-1" {
t.Errorf("ReportsTo = %q, want %q", got.ReportsTo, "manager-1")
}
if string(got.Status) != "paused" {
t.Errorf("Status changed: got %q, want %q", got.Status, "paused")
}
if got.PauseReason != "manual pause" {
t.Errorf("PauseReason changed: got %q, want %q", got.PauseReason, "manual pause")
}
if got.Settings != wantSettings {
t.Errorf("Settings changed: got %q, want %q", got.Settings, wantSettings)
}
if got.Role != agent.Role {
t.Errorf("Role changed: got %q, want %q", got.Role, agent.Role)
}
}
@@ -95,6 +95,34 @@ func (r *Repository) UpdateProject(ctx context.Context, project *models.Project)
return err
}
// ProjectConfigFields is the subset of office_projects columns a config
// import owns (see UpdateProjectConfigFields).
type ProjectConfigFields struct {
Description string
Color string
BudgetCents int
Repositories string
ExecutorConfig string
}
// UpdateProjectConfigFields updates only the columns a config import owns
// (description, color, budget, repositories, executor config), leaving
// status and lead_agent_profile_id untouched instead of reverting them to
// a stale read-then-write snapshot.
func (r *Repository) UpdateProjectConfigFields(
ctx context.Context, id string, fields ProjectConfigFields,
) error {
now := time.Now().UTC()
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE office_projects SET
description = ?, color = ?, budget_cents = ?,
repositories = ?, executor_config = ?, updated_at = ?
WHERE id = ?
`), fields.Description, fields.Color, fields.BudgetCents,
fields.Repositories, fields.ExecutorConfig, now, id)
return err
}
// DeleteProject deletes a project by ID.
func (r *Repository) DeleteProject(ctx context.Context, id string) error {
_, err := r.db.ExecContext(ctx, r.db.Rebind(
@@ -0,0 +1,76 @@
package sqlite_test
import (
"context"
"testing"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// TestUpdateProjectConfigFields_UpdatesOnlyOwnedColumns proves the
// column-scoped writer touches only description, color, budget_cents,
// repositories, and executor_config - leaving status and
// lead_agent_profile_id, which a config import does not own, exactly as
// they were.
func TestUpdateProjectConfigFields_UpdatesOnlyOwnedColumns(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
project := &models.Project{
ID: "project-1",
WorkspaceID: "ws-1",
Name: "apollo",
Description: "moon shot",
Status: models.ProjectStatusActive,
LeadAgentProfileID: "agent-1",
Color: "#ff00ff",
BudgetCents: 99000,
Repositories: `["kdlbs/kandev"]`,
ExecutorConfig: `{"type":"local_docker"}`,
}
if err := repo.CreateProject(ctx, project); err != nil {
t.Fatalf("CreateProject: %v", err)
}
fields := sqlite.ProjectConfigFields{
Description: "mars shot",
Color: "#00ff00",
BudgetCents: 1234,
Repositories: `["kdlbs/other"]`,
ExecutorConfig: `{"type":"local_pc"}`,
}
if err := repo.UpdateProjectConfigFields(ctx, project.ID, fields); err != nil {
t.Fatalf("UpdateProjectConfigFields: %v", err)
}
got, err := repo.GetProject(ctx, project.ID)
if err != nil {
t.Fatalf("GetProject: %v", err)
}
// Owned columns landed.
if got.Description != fields.Description {
t.Errorf("Description = %q, want %q", got.Description, fields.Description)
}
if got.Color != fields.Color {
t.Errorf("Color = %q, want %q", got.Color, fields.Color)
}
if got.BudgetCents != fields.BudgetCents {
t.Errorf("BudgetCents = %d, want %d", got.BudgetCents, fields.BudgetCents)
}
if got.Repositories != fields.Repositories {
t.Errorf("Repositories = %q, want %q", got.Repositories, fields.Repositories)
}
if got.ExecutorConfig != fields.ExecutorConfig {
t.Errorf("ExecutorConfig = %q, want %q", got.ExecutorConfig, fields.ExecutorConfig)
}
// Everything else stayed exactly as seeded.
if got.Status != project.Status {
t.Errorf("Status changed: got %q, want %q", got.Status, project.Status)
}
if got.LeadAgentProfileID != project.LeadAgentProfileID {
t.Errorf("LeadAgentProfileID changed: got %q, want %q", got.LeadAgentProfileID, project.LeadAgentProfileID)
}
}
@@ -189,6 +189,31 @@ func (r *Repository) UpdateRoutine(ctx context.Context, routine *models.Routine)
return err
}
// RoutineConfigFields is the subset of office_routines columns a config
// import owns (see UpdateRoutineConfigFields).
type RoutineConfigFields struct {
Description string
TaskTemplate string
ConcurrencyPolicy models.RoutineConcurrencyPolicy
}
// UpdateRoutineConfigFields updates only the columns a config import owns
// (description, task template, concurrency policy), leaving
// assignee_agent_profile_id, status, catch_up_policy, catch_up_max,
// variables, and last_run_at untouched instead of reverting them to a
// stale read-then-write snapshot.
func (r *Repository) UpdateRoutineConfigFields(
ctx context.Context, id string, fields RoutineConfigFields,
) error {
now := time.Now().UTC()
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE office_routines SET
description = ?, task_template = ?, concurrency_policy = ?, updated_at = ?
WHERE id = ?
`), fields.Description, fields.TaskTemplate, fields.ConcurrencyPolicy, now, id)
return err
}
// DeleteRoutine deletes a routine by ID.
func (r *Repository) DeleteRoutine(ctx context.Context, id string) error {
_, err := r.db.ExecContext(ctx, r.db.Rebind(
@@ -0,0 +1,78 @@
package sqlite_test
import (
"context"
"testing"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// TestUpdateRoutineConfigFields_UpdatesOnlyOwnedColumns proves the
// column-scoped writer touches only description, task_template, and
// concurrency_policy - leaving status and every other column, which a
// config import does not own, exactly as they were.
func TestUpdateRoutineConfigFields_UpdatesOnlyOwnedColumns(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
routine := &models.Routine{
ID: "routine-1",
WorkspaceID: "ws-1",
Name: "standup",
Description: "daily standup",
TaskTemplate: "summarise yesterday",
AssigneeAgentProfileID: "agent-1",
Status: "active",
ConcurrencyPolicy: models.ConcurrencyPolicySkipIfActive,
CatchUpPolicy: "enqueue_missed_with_cap",
CatchUpMax: 25,
Variables: `{"team":"platform"}`,
}
if err := repo.CreateRoutine(ctx, routine); err != nil {
t.Fatalf("CreateRoutine: %v", err)
}
fields := sqlite.RoutineConfigFields{
Description: "weekly",
TaskTemplate: "summarise the week",
ConcurrencyPolicy: models.ConcurrencyPolicyAlwaysCreate,
}
if err := repo.UpdateRoutineConfigFields(ctx, routine.ID, fields); err != nil {
t.Fatalf("UpdateRoutineConfigFields: %v", err)
}
got, err := repo.GetRoutine(ctx, routine.ID)
if err != nil {
t.Fatalf("GetRoutine: %v", err)
}
// Owned columns landed.
if got.Description != fields.Description {
t.Errorf("Description = %q, want %q", got.Description, fields.Description)
}
if got.TaskTemplate != fields.TaskTemplate {
t.Errorf("TaskTemplate = %q, want %q", got.TaskTemplate, fields.TaskTemplate)
}
if got.ConcurrencyPolicy != fields.ConcurrencyPolicy {
t.Errorf("ConcurrencyPolicy = %q, want %q", got.ConcurrencyPolicy, fields.ConcurrencyPolicy)
}
// Everything else stayed exactly as seeded.
if got.AssigneeAgentProfileID != routine.AssigneeAgentProfileID {
t.Errorf("AssigneeAgentProfileID changed: got %q, want %q",
got.AssigneeAgentProfileID, routine.AssigneeAgentProfileID)
}
if got.Status != routine.Status {
t.Errorf("Status changed: got %q, want %q", got.Status, routine.Status)
}
if got.CatchUpPolicy != routine.CatchUpPolicy {
t.Errorf("CatchUpPolicy changed: got %q, want %q", got.CatchUpPolicy, routine.CatchUpPolicy)
}
if got.CatchUpMax != routine.CatchUpMax {
t.Errorf("CatchUpMax changed: got %d, want %d", got.CatchUpMax, routine.CatchUpMax)
}
if got.Variables != routine.Variables {
t.Errorf("Variables changed: got %q, want %q", got.Variables, routine.Variables)
}
}
@@ -141,6 +141,70 @@ func (r *Repository) UpdateSkill(ctx context.Context, skill *models.Skill) error
return err
}
// SkillConfigFields is the subset of office_skills columns a config import
// owns (see UpdateSkillConfigFields).
type SkillConfigFields struct {
Name string
Description string
SourceType models.SkillSourceType
Content string
}
// UpdateSkillConfigFields updates only the columns a config import owns
// (name, description, source type, content) and the content_hash derived from
// content plus the current package metadata. It leaves source_locator,
// file_inventory, version, approval_state, and the other columns untouched
// instead of reverting them to a stale read-then-write snapshot.
func (r *Repository) UpdateSkillConfigFields(
ctx context.Context, id string, fields SkillConfigFields,
) error {
const maxMetadataReadAttempts = 3
type packageMetadata struct {
SourceLocator string `db:"source_locator"`
FileInventory string `db:"file_inventory"`
}
for range maxMetadataReadAttempts {
var metadata packageMetadata
if err := r.db.QueryRowxContext(ctx, r.db.Rebind(`
SELECT COALESCE(source_locator, '') AS source_locator,
COALESCE(file_inventory, '') AS file_inventory
FROM office_skills
WHERE id = ?
`), id).StructScan(&metadata); err != nil {
if err == sql.ErrNoRows {
return fmt.Errorf("skill not found: %s", id)
}
return err
}
contentHash := models.SkillPackageContentHash(
fields.Content, metadata.FileInventory, metadata.SourceLocator,
)
result, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE office_skills SET
name = ?, description = ?, source_type = ?, content = ?,
content_hash = ?, updated_at = ?
WHERE id = ?
AND COALESCE(source_locator, '') = ?
AND COALESCE(file_inventory, '') = ?
`), fields.Name, fields.Description, fields.SourceType, fields.Content,
contentHash, time.Now().UTC(), id, metadata.SourceLocator, metadata.FileInventory)
if err != nil {
return err
}
updated, err := result.RowsAffected()
if err != nil {
return err
}
if updated == 1 {
return nil
}
}
return fmt.Errorf("update skill %s: package metadata changed concurrently", id)
}
// DeleteSkill deletes a skill by ID.
func (r *Repository) DeleteSkill(ctx context.Context, id string) error {
_, err := r.db.ExecContext(ctx, r.db.Rebind(
@@ -0,0 +1,96 @@
package sqlite_test
import (
"context"
"crypto/sha256"
"encoding/hex"
"testing"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// TestUpdateSkillConfigFields_UpdatesOnlyOwnedColumns proves the
// column-scoped writer touches only name, description, source_type, content,
// and the hash derived from content. It leaves approval_state and package
// metadata, which a config import does not own, exactly as they were.
func TestUpdateSkillConfigFields_UpdatesOnlyOwnedColumns(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
skill := &models.Skill{
ID: "skill-1",
WorkspaceID: "ws-1",
Name: "Code Review",
Slug: "code-review",
Description: "reviews diffs",
SourceType: "inline",
SourceLocator: "local://code-review",
Content: "# Code Review\n",
FileInventory: `["SKILL.md"]`,
Version: "v1",
ContentHash: "hash-v1",
ApprovalState: "approved",
CreatedByAgentProfileID: "agent-1",
IsSystem: false,
SystemVersion: "",
DefaultForRoles: `["cto"]`,
}
if err := repo.CreateSkill(ctx, skill); err != nil {
t.Fatalf("CreateSkill: %v", err)
}
fields := sqlite.SkillConfigFields{
Name: "Deep Review",
Description: "reads everything",
SourceType: models.SkillSourceType("git"),
Content: "# Deep Review\n",
}
if err := repo.UpdateSkillConfigFields(ctx, skill.ID, fields); err != nil {
t.Fatalf("UpdateSkillConfigFields: %v", err)
}
got, err := repo.GetSkill(ctx, skill.ID)
if err != nil {
t.Fatalf("GetSkill: %v", err)
}
// Owned columns landed.
if got.Name != fields.Name {
t.Errorf("Name = %q, want %q", got.Name, fields.Name)
}
if got.Description != fields.Description {
t.Errorf("Description = %q, want %q", got.Description, fields.Description)
}
if got.SourceType != fields.SourceType {
t.Errorf("SourceType = %q, want %q", got.SourceType, fields.SourceType)
}
if got.Content != fields.Content {
t.Errorf("Content = %q, want %q", got.Content, fields.Content)
}
// Everything else stayed exactly as seeded.
if got.Slug != skill.Slug {
t.Errorf("Slug changed: got %q, want %q", got.Slug, skill.Slug)
}
if got.SourceLocator != skill.SourceLocator {
t.Errorf("SourceLocator changed: got %q, want %q", got.SourceLocator, skill.SourceLocator)
}
if got.FileInventory != skill.FileInventory {
t.Errorf("FileInventory changed: got %q, want %q", got.FileInventory, skill.FileInventory)
}
if got.Version != skill.Version {
t.Errorf("Version changed: got %q, want %q", got.Version, skill.Version)
}
wantHashBytes := sha256.Sum256([]byte(fields.Content + "\x00" + skill.FileInventory + "\x00" + skill.SourceLocator))
wantHash := hex.EncodeToString(wantHashBytes[:])
if got.ContentHash != wantHash {
t.Errorf("ContentHash = %q, want %q for updated content", got.ContentHash, wantHash)
}
if got.ApprovalState != skill.ApprovalState {
t.Errorf("ApprovalState changed: got %q, want %q", got.ApprovalState, skill.ApprovalState)
}
if got.DefaultForRoles != skill.DefaultForRoles {
t.Errorf("DefaultForRoles changed: got %q, want %q", got.DefaultForRoles, skill.DefaultForRoles)
}
}
@@ -3,8 +3,6 @@ package service
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"os"
@@ -500,8 +498,9 @@ func prepareServiceSkillPackageMetadata(skill *models.Skill) {
if skill.ApprovalState == "" {
skill.ApprovalState = "approved"
}
sum := sha256.Sum256([]byte(skill.Content + "\x00" + skill.FileInventory + "\x00" + skill.SourceLocator))
skill.ContentHash = hex.EncodeToString(sum[:])
skill.ContentHash = models.SkillPackageContentHash(
skill.Content, skill.FileInventory, skill.SourceLocator,
)
}
// DeleteSkill deletes a skill from the DB.
@@ -2,8 +2,6 @@ package skills
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"regexp"
"strings"
@@ -153,8 +151,9 @@ func prepareSkillPackageMetadata(skill *models.Skill) {
if skill.ApprovalState == "" {
skill.ApprovalState = "approved"
}
sum := sha256.Sum256([]byte(skill.Content + "\x00" + skill.FileInventory + "\x00" + skill.SourceLocator))
skill.ContentHash = hex.EncodeToString(sum[:])
skill.ContentHash = models.SkillPackageContentHash(
skill.Content, skill.FileInventory, skill.SourceLocator,
)
}
// ValidateSkillUpdate validates a skill update for slug uniqueness.