postgres

package
v1.10.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 4, 2026 License: MIT Imports: 43 Imported by: 1

Documentation

Overview

Package postgres provides a PostgreSQL implementation of the Cortex composite store using grove ORM with programmatic migrations.

Index

Constants

This section is empty.

Variables

View Source 
var Migrations = func() *migrate.Group {
	g := migrate.NewGroup("cortex")
	g.MustRegister(
		&migrate.Migration{
			Name:    "create_agents",
			Version: "20240101000001",
			Comment: "Create cortex_agents table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_agents (
    id              TEXT PRIMARY KEY,
    name            TEXT NOT NULL,
    description     TEXT DEFAULT '',
    app_id          TEXT NOT NULL,
    system_prompt   TEXT DEFAULT '',
    model           TEXT DEFAULT '',
    tools           JSONB DEFAULT '[]',
    max_steps       INTEGER DEFAULT 0,
    max_tokens      INTEGER DEFAULT 0,
    temperature     DOUBLE PRECISION DEFAULT 0,
    reasoning_loop  TEXT DEFAULT '',
    guardrails      JSONB DEFAULT '{}',
    metadata        JSONB DEFAULT '{}',
    enabled         BOOLEAN DEFAULT TRUE,
    persona_ref     TEXT DEFAULT '',
    inline_skills   JSONB DEFAULT '[]',
    inline_traits   JSONB DEFAULT '[]',
    inline_behaviors JSONB DEFAULT '[]',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_agents_app_name ON cortex_agents (app_id, name);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_agents CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_runs",
			Version: "20240101000002",
			Comment: "Create cortex_runs table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_runs (
    id            TEXT PRIMARY KEY,
    agent_id      TEXT NOT NULL REFERENCES cortex_agents(id) ON DELETE CASCADE,
    tenant_id     TEXT DEFAULT '',
    state         TEXT NOT NULL DEFAULT 'created',
    input         TEXT DEFAULT '',
    output        TEXT DEFAULT '',
    error         TEXT DEFAULT '',
    step_count    INTEGER DEFAULT 0,
    tokens_used   INTEGER DEFAULT 0,
    started_at    TIMESTAMPTZ,
    completed_at  TIMESTAMPTZ,
    persona_ref   TEXT DEFAULT '',
    metadata      JSONB DEFAULT '{}',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_runs_agent ON cortex_runs (agent_id);
CREATE INDEX IF NOT EXISTS idx_cortex_runs_state ON cortex_runs (state);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_runs CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_steps",
			Version: "20240101000003",
			Comment: "Create cortex_steps table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_steps (
    id            TEXT PRIMARY KEY,
    run_id        TEXT NOT NULL REFERENCES cortex_runs(id) ON DELETE CASCADE,
    "index"       INTEGER NOT NULL,
    type          TEXT DEFAULT '',
    input         TEXT DEFAULT '',
    output        TEXT DEFAULT '',
    tokens_used   INTEGER DEFAULT 0,
    started_at    TIMESTAMPTZ,
    completed_at  TIMESTAMPTZ,
    metadata      JSONB DEFAULT '{}',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_steps_run ON cortex_steps (run_id);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_steps CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_tool_calls",
			Version: "20240101000004",
			Comment: "Create cortex_tool_calls table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_tool_calls (
    id            TEXT PRIMARY KEY,
    step_id       TEXT NOT NULL REFERENCES cortex_steps(id) ON DELETE CASCADE,
    run_id        TEXT NOT NULL REFERENCES cortex_runs(id) ON DELETE CASCADE,
    tool_name     TEXT NOT NULL,
    arguments     TEXT DEFAULT '',
    result        TEXT DEFAULT '',
    error         TEXT DEFAULT '',
    started_at    TIMESTAMPTZ,
    completed_at  TIMESTAMPTZ,
    metadata      JSONB DEFAULT '{}',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_tool_calls_step ON cortex_tool_calls (step_id);
CREATE INDEX IF NOT EXISTS idx_cortex_tool_calls_run ON cortex_tool_calls (run_id);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_tool_calls CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_memories",
			Version: "20240101000005",
			Comment: "Create cortex_memories table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_memories (
    id          BIGSERIAL PRIMARY KEY,
    agent_id    TEXT NOT NULL,
    tenant_id   TEXT DEFAULT '',
    kind        TEXT NOT NULL,
    key         TEXT DEFAULT '',
    content     TEXT NOT NULL,
    metadata    JSONB DEFAULT '{}',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_memories_agent_kind ON cortex_memories (agent_id, kind);
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_memories_working ON cortex_memories (agent_id, kind, key) WHERE kind = 'working';
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_memories CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_checkpoints",
			Version: "20240101000006",
			Comment: "Create cortex_checkpoints table",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_checkpoints (
    id          TEXT PRIMARY KEY,
    run_id      TEXT NOT NULL REFERENCES cortex_runs(id) ON DELETE CASCADE,
    agent_id    TEXT NOT NULL REFERENCES cortex_agents(id) ON DELETE CASCADE,
    tenant_id   TEXT DEFAULT '',
    reason      TEXT DEFAULT '',
    step_index  INTEGER DEFAULT 0,
    state       TEXT NOT NULL DEFAULT 'pending',
    decision    JSONB,
    metadata    JSONB DEFAULT '{}',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_checkpoints_run ON cortex_checkpoints (run_id);
CREATE INDEX IF NOT EXISTS idx_cortex_checkpoints_state ON cortex_checkpoints (state);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_checkpoints CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_skills_traits",
			Version: "20240101000007",
			Comment: "Create cortex_skills and cortex_traits tables",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_skills (
    id                      TEXT PRIMARY KEY,
    name                    TEXT NOT NULL,
    description             TEXT DEFAULT '',
    app_id                  TEXT NOT NULL,
    tools                   JSONB DEFAULT '[]',
    knowledge               JSONB DEFAULT '[]',
    system_prompt_fragment  TEXT DEFAULT '',
    dependencies            JSONB DEFAULT '[]',
    default_proficiency     TEXT DEFAULT '',
    metadata                JSONB DEFAULT '{}',
    created_at              TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at              TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_skills_app_name ON cortex_skills (app_id, name);

CREATE TABLE IF NOT EXISTS cortex_traits (
    id          TEXT PRIMARY KEY,
    name        TEXT NOT NULL,
    description TEXT DEFAULT '',
    app_id      TEXT NOT NULL,
    dimensions  JSONB DEFAULT '[]',
    influences  JSONB DEFAULT '[]',
    category    TEXT DEFAULT '',
    metadata    JSONB DEFAULT '{}',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_traits_app_name ON cortex_traits (app_id, name);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP TABLE IF EXISTS cortex_traits CASCADE;
DROP TABLE IF EXISTS cortex_skills CASCADE;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_behaviors_personas",
			Version: "20240101000008",
			Comment: "Create cortex_behaviors and cortex_personas tables",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_behaviors (
    id              TEXT PRIMARY KEY,
    name            TEXT NOT NULL,
    description     TEXT DEFAULT '',
    app_id          TEXT NOT NULL,
    triggers        JSONB DEFAULT '[]',
    actions         JSONB DEFAULT '[]',
    priority        INTEGER DEFAULT 0,
    requires_skill  TEXT DEFAULT '',
    requires_trait  TEXT DEFAULT '',
    metadata        JSONB DEFAULT '{}',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_behaviors_app_name ON cortex_behaviors (app_id, name);

CREATE TABLE IF NOT EXISTS cortex_personas (
    id                   TEXT PRIMARY KEY,
    name                 TEXT NOT NULL,
    description          TEXT DEFAULT '',
    app_id               TEXT NOT NULL,
    identity             TEXT DEFAULT '',
    skills               JSONB DEFAULT '[]',
    traits               JSONB DEFAULT '[]',
    behaviors            JSONB DEFAULT '[]',
    cognitive_style      JSONB DEFAULT '{}',
    communication_style  JSONB DEFAULT '{}',
    perception           JSONB DEFAULT '{}',
    metadata             JSONB DEFAULT '{}',
    created_at           TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at           TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_personas_app_name ON cortex_personas (app_id, name);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP TABLE IF EXISTS cortex_personas CASCADE;
DROP TABLE IF EXISTS cortex_behaviors CASCADE;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_orchestrations",
			Version: "20240101000009",
			Comment: "Create cortex_orchestration_configs and cortex_orchestration_runs tables",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_orchestration_configs (
    id            TEXT PRIMARY KEY,
    name          TEXT NOT NULL,
    description   TEXT DEFAULT '',
    app_id        TEXT NOT NULL,
    strategy      TEXT DEFAULT '',
    participants  JSONB DEFAULT '[]',
    settings      JSONB DEFAULT '{}',
    metadata      JSONB DEFAULT '{}',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_orchestration_configs_app_name ON cortex_orchestration_configs (app_id, name);

CREATE TABLE IF NOT EXISTS cortex_orchestration_runs (
    id             TEXT PRIMARY KEY,
    config_id      TEXT DEFAULT '',
    app_id         TEXT NOT NULL,
    tenant_id      TEXT DEFAULT '',
    strategy       TEXT DEFAULT '',
    status         TEXT NOT NULL DEFAULT 'running',
    input          TEXT DEFAULT '',
    output         TEXT DEFAULT '',
    error          TEXT DEFAULT '',
    agent_run_ids  JSONB DEFAULT '[]',
    started_at     TIMESTAMPTZ,
    completed_at   TIMESTAMPTZ,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_orchestration_runs_app_status ON cortex_orchestration_runs (app_id, status);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP TABLE IF EXISTS cortex_orchestration_runs;
DROP TABLE IF EXISTS cortex_orchestration_configs;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "add_scope_columns",
			Version: "20260821000001",
			Comment: "Add host-defined scope columns; pre-v2 rows are left unscoped",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, table := range []string{
					"cortex_agents",
					"cortex_runs",
					"cortex_memories",
					"cortex_checkpoints",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    ADD COLUMN IF NOT EXISTS scope_l0    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l1    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l2    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_extra JSONB NOT NULL DEFAULT '{}',
    ADD COLUMN IF NOT EXISTS scope_canon TEXT  NOT NULL DEFAULT ''
`); err != nil {
						return fmt.Errorf("add scope columns to %s: %w", table, err)
					}
					if _, err := exec.Exec(ctx, `
CREATE INDEX IF NOT EXISTS idx_`+table+`_scope ON `+table+` (scope_l0, scope_l1, scope_l2)
`); err != nil {
						return fmt.Errorf("index scope on %s: %w", table, err)
					}
				}

				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				for _, table := range []string{
					"cortex_agents",
					"cortex_runs",
					"cortex_memories",
					"cortex_checkpoints",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    DROP COLUMN IF EXISTS scope_l0,
    DROP COLUMN IF EXISTS scope_l1,
    DROP COLUMN IF EXISTS scope_l2,
    DROP COLUMN IF EXISTS scope_extra,
    DROP COLUMN IF EXISTS scope_canon
`); err != nil {
						return fmt.Errorf("drop scope columns from %s: %w", table, err)
					}
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "add_scope_columns_steps_tool_calls",
			Version: "20260822000001",
			Comment: "Add host-defined scope columns to cortex_steps and cortex_tool_calls; pre-existing rows are left unscoped",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, table := range []string{
					"cortex_steps",
					"cortex_tool_calls",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    ADD COLUMN IF NOT EXISTS scope_l0    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l1    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l2    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_extra JSONB NOT NULL DEFAULT '{}',
    ADD COLUMN IF NOT EXISTS scope_canon TEXT  NOT NULL DEFAULT ''
`); err != nil {
						return fmt.Errorf("add scope columns to %s: %w", table, err)
					}
					if _, err := exec.Exec(ctx, `
CREATE INDEX IF NOT EXISTS idx_`+table+`_scope ON `+table+` (scope_l0, scope_l1, scope_l2)
`); err != nil {
						return fmt.Errorf("index scope on %s: %w", table, err)
					}
				}

				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				for _, table := range []string{
					"cortex_steps",
					"cortex_tool_calls",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    DROP COLUMN IF EXISTS scope_l0,
    DROP COLUMN IF EXISTS scope_l1,
    DROP COLUMN IF EXISTS scope_l2,
    DROP COLUMN IF EXISTS scope_extra,
    DROP COLUMN IF EXISTS scope_canon
`); err != nil {
						return fmt.Errorf("drop scope columns from %s: %w", table, err)
					}
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "scope_working_memory_unique_index",
			Version: "20260822000002",
			Comment: "Make the working-memory partial unique index scope-aware",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `DROP INDEX CONCURRENTLY IF EXISTS idx_cortex_memories_working`); err != nil {
					return fmt.Errorf("drop old working-memory index: %w", err)
				}
				if _, err := exec.Exec(ctx, `CREATE UNIQUE INDEX CONCURRENTLY idx_cortex_memories_working ON cortex_memories (agent_id, kind, key, scope_canon) WHERE kind = 'working'`); err != nil {
					return fmt.Errorf("create scope-aware working-memory index: %w", err)
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `DROP INDEX CONCURRENTLY IF EXISTS idx_cortex_memories_working`); err != nil {
					return fmt.Errorf("drop scope-aware working-memory index: %w", err)
				}
				if _, err := exec.Exec(ctx, `CREATE UNIQUE INDEX CONCURRENTLY idx_cortex_memories_working ON cortex_memories (agent_id, kind, key) WHERE kind = 'working'`); err != nil {
					return fmt.Errorf("recreate pre-scope working-memory index: %w", err)
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "scope_agents_unique_index",
			Version: "20260823000001",
			Comment: "Replace the app_id-keyed agent name uniqueness with a scope-keyed one",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_agents_app_name;
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_agents_scope_name ON cortex_agents (scope_canon, name) WHERE scope_canon != '';
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_agents_scope_name;
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_agents_app_name ON cortex_agents (app_id, name);
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "scope_skills_traits_behaviors",
			Version: "20260823000002",
			Comment: "Add host-defined scope columns to cortex_skills/cortex_traits/cortex_behaviors and replace their app_id-keyed unique indexes with scope-keyed ones",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, table := range []string{
					"cortex_skills",
					"cortex_traits",
					"cortex_behaviors",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    ADD COLUMN IF NOT EXISTS scope_l0    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l1    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l2    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_extra JSONB NOT NULL DEFAULT '{}',
    ADD COLUMN IF NOT EXISTS scope_canon TEXT  NOT NULL DEFAULT ''
`); err != nil {
						return fmt.Errorf("add scope columns to %s: %w", table, err)
					}
					if _, err := exec.Exec(ctx, `
CREATE INDEX IF NOT EXISTS idx_`+table+`_scope ON `+table+` (scope_l0, scope_l1, scope_l2)
`); err != nil {
						return fmt.Errorf("index scope on %s: %w", table, err)
					}
				}

				for _, spec := range []struct{ table, oldIdx, newIdx string }{
					{"cortex_skills", "idx_cortex_skills_app_name", "idx_cortex_skills_scope_name"},
					{"cortex_traits", "idx_cortex_traits_app_name", "idx_cortex_traits_scope_name"},
					{"cortex_behaviors", "idx_cortex_behaviors_app_name", "idx_cortex_behaviors_scope_name"},
				} {
					if _, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS `+spec.oldIdx+`;
CREATE UNIQUE INDEX IF NOT EXISTS `+spec.newIdx+` ON `+spec.table+` (scope_canon, name) WHERE scope_canon != '';
`); err != nil {
						return fmt.Errorf("scope-key unique index on %s: %w", spec.table, err)
					}
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {

				for _, spec := range []struct{ table, oldIdx, newIdx string }{
					{"cortex_skills", "idx_cortex_skills_app_name", "idx_cortex_skills_scope_name"},
					{"cortex_traits", "idx_cortex_traits_app_name", "idx_cortex_traits_scope_name"},
					{"cortex_behaviors", "idx_cortex_behaviors_app_name", "idx_cortex_behaviors_scope_name"},
				} {
					if _, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS `+spec.newIdx+`;
CREATE UNIQUE INDEX IF NOT EXISTS `+spec.oldIdx+` ON `+spec.table+` (app_id, name);
`); err != nil {
						return fmt.Errorf("restore app_id-key unique index on %s: %w", spec.table, err)
					}
				}
				for _, table := range []string{
					"cortex_skills",
					"cortex_traits",
					"cortex_behaviors",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    DROP COLUMN IF EXISTS scope_l0,
    DROP COLUMN IF EXISTS scope_l1,
    DROP COLUMN IF EXISTS scope_l2,
    DROP COLUMN IF EXISTS scope_extra,
    DROP COLUMN IF EXISTS scope_canon
`); err != nil {
						return fmt.Errorf("drop scope columns from %s: %w", table, err)
					}
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "scope_personas",
			Version: "20260823000003",
			Comment: "Add host-defined scope columns to cortex_personas and replace its app_id-keyed unique index with a scope-keyed one",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `
ALTER TABLE cortex_personas
    ADD COLUMN IF NOT EXISTS scope_l0    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l1    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l2    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_extra JSONB NOT NULL DEFAULT '{}',
    ADD COLUMN IF NOT EXISTS scope_canon TEXT  NOT NULL DEFAULT ''
`); err != nil {
					return fmt.Errorf("add scope columns to cortex_personas: %w", err)
				}
				if _, err := exec.Exec(ctx, `
CREATE INDEX IF NOT EXISTS idx_cortex_personas_scope ON cortex_personas (scope_l0, scope_l1, scope_l2)
`); err != nil {
					return fmt.Errorf("index scope on cortex_personas: %w", err)
				}

				if _, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_personas_app_name;
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_personas_scope_name ON cortex_personas (scope_canon, name) WHERE scope_canon != '';
`); err != nil {
					return fmt.Errorf("scope-key unique index on cortex_personas: %w", err)
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_personas_scope_name;
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_personas_app_name ON cortex_personas (app_id, name);
`); err != nil {
					return fmt.Errorf("restore app_id-key unique index on cortex_personas: %w", err)
				}
				if _, err := exec.Exec(ctx, `
ALTER TABLE cortex_personas
    DROP COLUMN IF EXISTS scope_l0,
    DROP COLUMN IF EXISTS scope_l1,
    DROP COLUMN IF EXISTS scope_l2,
    DROP COLUMN IF EXISTS scope_extra,
    DROP COLUMN IF EXISTS scope_canon
`); err != nil {
					return fmt.Errorf("drop scope columns from cortex_personas: %w", err)
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "scope_orchestrations",
			Version: "20260823000004",
			Comment: "Add host-defined scope columns to cortex_orchestration_configs/cortex_orchestration_runs and replace the configs' app_id-keyed unique index with a scope-keyed one",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				for _, table := range []string{
					"cortex_orchestration_configs",
					"cortex_orchestration_runs",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    ADD COLUMN IF NOT EXISTS scope_l0    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l1    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_l2    TEXT  NOT NULL DEFAULT '',
    ADD COLUMN IF NOT EXISTS scope_extra JSONB NOT NULL DEFAULT '{}',
    ADD COLUMN IF NOT EXISTS scope_canon TEXT  NOT NULL DEFAULT ''
`); err != nil {
						return fmt.Errorf("add scope columns to %s: %w", table, err)
					}
					if _, err := exec.Exec(ctx, `
CREATE INDEX IF NOT EXISTS idx_`+table+`_scope ON `+table+` (scope_l0, scope_l1, scope_l2)
`); err != nil {
						return fmt.Errorf("index scope on %s: %w", table, err)
					}
				}

				if _, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_orchestration_configs_app_name;
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_orchestration_configs_scope_name ON cortex_orchestration_configs (scope_canon, name) WHERE scope_canon != '';
`); err != nil {
					return fmt.Errorf("scope-key unique index on cortex_orchestration_configs: %w", err)
				}

				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_orchestration_configs_scope_name;
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_orchestration_configs_app_name ON cortex_orchestration_configs (app_id, name);
`); err != nil {
					return fmt.Errorf("restore app_id-key unique index on cortex_orchestration_configs: %w", err)
				}
				for _, table := range []string{
					"cortex_orchestration_configs",
					"cortex_orchestration_runs",
				} {
					if _, err := exec.Exec(ctx, `
ALTER TABLE `+table+`
    DROP COLUMN IF EXISTS scope_l0,
    DROP COLUMN IF EXISTS scope_l1,
    DROP COLUMN IF EXISTS scope_l2,
    DROP COLUMN IF EXISTS scope_extra,
    DROP COLUMN IF EXISTS scope_canon
`); err != nil {
						return fmt.Errorf("drop scope columns from %s: %w", table, err)
					}
				}
				return nil
			},
		},
		&migrate.Migration{
			Name:    "create_sessions",
			Version: "20260824000001",
			Comment: "Create cortex_sessions table, scoped from birth",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_sessions (
    id            TEXT PRIMARY KEY,
    agent_id      TEXT NOT NULL,
    title         TEXT NOT NULL DEFAULT '',
    message_count INTEGER NOT NULL DEFAULT 0,
    last_message  TEXT NOT NULL DEFAULT '',
    is_default    BOOLEAN NOT NULL DEFAULT FALSE,
    metadata      JSONB NOT NULL DEFAULT '{}',
    backfilled_by TEXT NOT NULL DEFAULT '',
    scope_l0      TEXT NOT NULL DEFAULT '',
    scope_l1      TEXT NOT NULL DEFAULT '',
    scope_l2      TEXT NOT NULL DEFAULT '',
    scope_extra   JSONB NOT NULL DEFAULT '{}',
    scope_canon   TEXT NOT NULL DEFAULT '',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_sessions_scope
    ON cortex_sessions (scope_l0, scope_l1, scope_l2);

CREATE INDEX IF NOT EXISTS idx_cortex_sessions_agent
    ON cortex_sessions (agent_id, scope_canon);
`); err != nil {
					return fmt.Errorf("create cortex_sessions: %w", err)
				}

				if _, err := exec.Exec(ctx, `
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_sessions_default
    ON cortex_sessions (agent_id, scope_canon)
    WHERE is_default AND scope_canon != '';
`); err != nil {
					return fmt.Errorf("default-session unique index on cortex_sessions: %w", err)
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_sessions CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "add_memory_session_id",
			Version: "20260824000002",
			Comment: "Key conversation memory on a session",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
ALTER TABLE cortex_memories ADD COLUMN IF NOT EXISTS session_id TEXT NOT NULL DEFAULT '';

CREATE INDEX IF NOT EXISTS idx_cortex_memories_session ON cortex_memories (session_id, scope_canon);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP INDEX IF EXISTS idx_cortex_memories_session;
ALTER TABLE cortex_memories DROP COLUMN IF EXISTS session_id;
`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "add_run_session_id",
			Version: "20260824000003",
			Comment: "Carry the resolved session on a run",
			Up: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `ALTER TABLE cortex_runs ADD COLUMN IF NOT EXISTS session_id TEXT NOT NULL DEFAULT ''`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `ALTER TABLE cortex_runs DROP COLUMN IF EXISTS session_id`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "backfill_default_sessions",
			Version: "20260824000004",
			Comment: "Reserved: the default-session backfill runs unconditionally in Store.Migrate, after rescopeLegacyRows, not as this migration's one-shot Up -- see the Up func's comment",

			Up: func(context.Context, migrate.Executor) error { return nil },

			Down: unbackfillDefaultSessions,
		},
		&migrate.Migration{
			Name:    "create_suspensions",
			Version: "20260825000001",
			Comment: "Create cortex_suspensions table, scoped from birth",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_suspensions (
    id            TEXT PRIMARY KEY,
    run_id        TEXT NOT NULL,
    reason        TEXT NOT NULL,
    pending       JSONB NOT NULL DEFAULT '[]',
    continuation  JSONB NOT NULL DEFAULT '{}',
    expires_at    TIMESTAMPTZ,
    scope_l0      TEXT NOT NULL DEFAULT '',
    scope_l1      TEXT NOT NULL DEFAULT '',
    scope_l2      TEXT NOT NULL DEFAULT '',
    scope_extra   JSONB NOT NULL DEFAULT '{}',
    scope_canon   TEXT NOT NULL DEFAULT '',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_suspensions_scope
    ON cortex_suspensions (scope_l0, scope_l1, scope_l2);
`); err != nil {
					return fmt.Errorf("create cortex_suspensions: %w", err)
				}

				if _, err := exec.Exec(ctx, `
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_suspensions_run
    ON cortex_suspensions (run_id) WHERE scope_canon != '';
`); err != nil {
					return fmt.Errorf("run unique index on cortex_suspensions: %w", err)
				}

				if _, err := exec.Exec(ctx, `
CREATE INDEX IF NOT EXISTS idx_cortex_suspensions_expiry
    ON cortex_suspensions (expires_at) WHERE expires_at IS NOT NULL;
`); err != nil {
					return fmt.Errorf("expiry index on cortex_suspensions: %w", err)
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_suspensions CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_overlays",
			Version: "20260826000001",
			Comment: "Create cortex_overlays table, scoped from birth",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				if _, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_overlays (
    id            TEXT PRIMARY KEY,
    agent_id      TEXT NOT NULL,
    patches       JSONB NOT NULL DEFAULT '[]',
    tools_added   JSONB NOT NULL DEFAULT '[]',
    tools_removed JSONB NOT NULL DEFAULT '[]',
    model         TEXT NOT NULL DEFAULT '',
    temperature   DOUBLE PRECISION,
    max_tokens    INTEGER,
    scope_l0      TEXT NOT NULL DEFAULT '',
    scope_l1      TEXT NOT NULL DEFAULT '',
    scope_l2      TEXT NOT NULL DEFAULT '',
    scope_extra   JSONB NOT NULL DEFAULT '{}',
    scope_canon   TEXT NOT NULL DEFAULT '',
    created_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at    TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_overlays_scope
    ON cortex_overlays (scope_l0, scope_l1, scope_l2);
`); err != nil {
					return fmt.Errorf("create cortex_overlays: %w", err)
				}

				if _, err := exec.Exec(ctx, `
CREATE UNIQUE INDEX IF NOT EXISTS idx_cortex_overlays_agent_scope
    ON cortex_overlays (agent_id, scope_canon)
    WHERE scope_canon != '';
`); err != nil {
					return fmt.Errorf("agent+scope unique index on cortex_overlays: %w", err)
				}
				return nil
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `DROP TABLE IF EXISTS cortex_overlays CASCADE`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "add_agent_sections",
			Version: "20260826000002",
			Comment: "Carry the system prompt as addressable sections on an agent",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `ALTER TABLE cortex_agents ADD COLUMN IF NOT EXISTS sections JSONB NOT NULL DEFAULT '[]'`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `ALTER TABLE cortex_agents DROP COLUMN IF EXISTS sections`)
				return err
			},
		},
		&migrate.Migration{
			Name:    "create_a2a",
			Version: "20260826000003",
			Comment: "Create the four cortex_a2a_* tables: messages, conversations, deliveries and pending asks",
			Up: func(ctx context.Context, exec migrate.Executor) error {

				_, err := exec.Exec(ctx, `
CREATE TABLE IF NOT EXISTS cortex_a2a_messages (
    id              TEXT PRIMARY KEY,
    performative    TEXT NOT NULL,
    sender_agent    TEXT NOT NULL,
    sender_node     TEXT NOT NULL DEFAULT '',
    receivers       JSONB NOT NULL DEFAULT '[]',
    reply_to        JSONB NOT NULL DEFAULT '[]',
    content         TEXT NOT NULL DEFAULT '',
    language        TEXT NOT NULL DEFAULT '',
    encoding        TEXT NOT NULL DEFAULT '',
    ontology        TEXT NOT NULL DEFAULT '',
    protocol        TEXT NOT NULL DEFAULT '',
    conversation_id TEXT NOT NULL DEFAULT '',
    reply_with      TEXT NOT NULL DEFAULT '',
    in_reply_to     TEXT NOT NULL DEFAULT '',
    reply_by        TIMESTAMPTZ,
    hops            INTEGER NOT NULL DEFAULT 0,
    origin_run_id   TEXT NOT NULL DEFAULT '',
    metadata        JSONB NOT NULL DEFAULT '{}',
    scope_l0        TEXT NOT NULL DEFAULT '',
    scope_l1        TEXT NOT NULL DEFAULT '',
    scope_l2        TEXT NOT NULL DEFAULT '',
    scope_extra     JSONB NOT NULL DEFAULT '{}',
    scope_canon     TEXT NOT NULL DEFAULT '',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_a2a_messages_scope_conv ON cortex_a2a_messages (scope_canon, conversation_id);

CREATE TABLE IF NOT EXISTS cortex_a2a_conversations (
    id               TEXT PRIMARY KEY,
    protocol         TEXT NOT NULL DEFAULT '',
    initiator_agent  TEXT NOT NULL DEFAULT '',
    initiator_node   TEXT NOT NULL DEFAULT '',
    participants     JSONB NOT NULL DEFAULT '[]',
    status           TEXT NOT NULL DEFAULT 'open',
    hop_ceiling      INTEGER NOT NULL DEFAULT 0,
    hops_used        INTEGER NOT NULL DEFAULT 0,
    deadline         TIMESTAMPTZ,
    peer_node        TEXT NOT NULL DEFAULT '',
    peer_context     TEXT NOT NULL DEFAULT '',
    scope_l0         TEXT NOT NULL DEFAULT '',
    scope_l1         TEXT NOT NULL DEFAULT '',
    scope_l2         TEXT NOT NULL DEFAULT '',
    scope_extra      JSONB NOT NULL DEFAULT '{}',
    scope_canon      TEXT NOT NULL DEFAULT '',
    created_at       TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at       TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_a2a_conversations_scope_status ON cortex_a2a_conversations (scope_canon, status);
CREATE INDEX IF NOT EXISTS idx_cortex_a2a_conversations_peer ON cortex_a2a_conversations (scope_canon, peer_node, peer_context);

CREATE TABLE IF NOT EXISTS cortex_a2a_deliveries (
    id             TEXT PRIMARY KEY,
    message_id     TEXT NOT NULL,
    receiver_agent TEXT NOT NULL,
    receiver_node  TEXT NOT NULL DEFAULT '',
    state          TEXT NOT NULL DEFAULT 'queued',
    error          TEXT NOT NULL DEFAULT '',
    claimed_at     TIMESTAMPTZ,
    delivered_at   TIMESTAMPTZ,
    read_at        TIMESTAMPTZ,
    run_id         TEXT NOT NULL DEFAULT '',
    scope_l0       TEXT NOT NULL DEFAULT '',
    scope_l1       TEXT NOT NULL DEFAULT '',
    scope_l2       TEXT NOT NULL DEFAULT '',
    scope_extra    JSONB NOT NULL DEFAULT '{}',
    scope_canon    TEXT NOT NULL DEFAULT '',
    created_at     TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_a2a_deliveries_inbox ON cortex_a2a_deliveries (scope_canon, receiver_agent, state);
CREATE INDEX IF NOT EXISTS idx_cortex_a2a_deliveries_state ON cortex_a2a_deliveries (state);

CREATE TABLE IF NOT EXISTS cortex_a2a_pending_asks (
    reply_with      TEXT PRIMARY KEY,
    conversation_id TEXT NOT NULL DEFAULT '',
    message_id      TEXT NOT NULL DEFAULT '',
    asker_run_id    TEXT NOT NULL DEFAULT '',
    asker_agent     TEXT NOT NULL DEFAULT '',
    tool_call_id    TEXT NOT NULL DEFAULT '',
    expected_agent  TEXT NOT NULL DEFAULT '',
    expected_node   TEXT NOT NULL DEFAULT '',
    deadline        TIMESTAMPTZ,
    claimed_at      TIMESTAMPTZ,
    scope_l0        TEXT NOT NULL DEFAULT '',
    scope_l1        TEXT NOT NULL DEFAULT '',
    scope_l2        TEXT NOT NULL DEFAULT '',
    scope_extra     JSONB NOT NULL DEFAULT '{}',
    scope_canon     TEXT NOT NULL DEFAULT '',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_cortex_a2a_pending_asks_deadline ON cortex_a2a_pending_asks (claimed_at, deadline);
`)
				return err
			},
			Down: func(ctx context.Context, exec migrate.Executor) error {
				_, err := exec.Exec(ctx, `
DROP TABLE IF EXISTS cortex_a2a_pending_asks;
DROP TABLE IF EXISTS cortex_a2a_deliveries;
DROP TABLE IF EXISTS cortex_a2a_conversations;
DROP TABLE IF EXISTS cortex_a2a_messages;
`)
				return err
			},
		},
	)
	return g
}()

Migrations is the grove migration group for the Cortex postgres store. It contains all schema migrations in version order.

Functions

This section is empty.

Types

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store is a PostgreSQL implementation of the composite Cortex store.

func New 

func New(db *grove.DB) *Store

New creates a new PostgreSQL store.

func (*Store) BindDeliveryRun 

func (s *Store) BindDeliveryRun(ctx context.Context, d *a2a.Delivery, runID id.AgentRunID) error

BindDeliveryRun records a reserved child only while this claim still owns it.

func (*Store) CapturePreparation 

func (s *Store) CapturePreparation(ctx context.Context, collect func(store.PreparationSources) error) (*run.SourceSnapshot, error)

CapturePreparation collects sources under one read-only repeatable-read transaction.

func (*Store) ClaimDelivery 

func (s *Store) ClaimDelivery(ctx context.Context, deliveryID id.DeliveryID) (*a2a.Delivery, error)

ClaimDelivery takes a queued delivery and marks it delivering. The state = 'queued' predicate is the claim: two workers racing both issue this UPDATE and exactly one of them changes a row, which is what stops one directive starting two runs.

func (*Store) ClaimExpiredSuspension 

func (s *Store) ClaimExpiredSuspension(ctx context.Context, runID id.AgentRunID, now time.Time) (*suspension.Suspension, error)

ClaimExpiredSuspension is ClaimSuspension with the deadline predicate inverted, for the sweeper. See the interface doc for why the two partition the same row rather than overlapping on it.

The rowcount check carries more weight here than it does above, because the sweeper's next move is to fail the run. Without it a sweep would mark failed a run some resume had already taken and was part way through continuing, and the run would keep executing under a terminal state nothing put it in.

Zero rows is reported as ErrNotSuspended without a second read to classify it. The three ways to get here (the row is gone, the run is no longer paused, the deadline moved) all mean the same thing to the only caller: somebody else owns this run, leave it alone.

func (*Store) ClaimOrchestrationContinuation 

func (s *Store) ClaimOrchestrationContinuation(ctx context.Context, team id.OrchestrationID) (*orchestration.Run, error)

ClaimOrchestrationContinuation elects one waiting team pass at the exact scope.

func (*Store) ClaimPendingAsk 

func (s *Store) ClaimPendingAsk(ctx context.Context, replyWith string) (*a2a.PendingAsk, error)

ClaimPendingAsk takes the ask carrying replyWith, but only if nobody has yet. The claimed_at IS NULL predicate is the whole guarantee: a late reply, the deadline sweep and a cancel can all reach for one row, and exactly one of them changes it and goes on to resume the waiting run.

func (*Store) ClaimSuspension 

func (s *Store) ClaimSuspension(ctx context.Context, runID id.AgentRunID) (*suspension.Suspension, error)

ClaimSuspension is the concurrency primitive for resume: it performs the run's paused-to-running transition and the suspension read as ONE operation, so two concurrent resumes cannot both proceed.

The conditional UPDATE and the suspension read share a transaction. The UPDATE carries its own state = 'paused' predicate, so under READ COMMITTED a second caller blocks on the row lock the first holds, then re-evaluates that predicate against the committed row and matches nothing. Zero rows affected means someone else won the race, and the loser gets cortex.ErrNotSuspended rather than the suspension it lost.

A read followed by a separate write is the exact race this method exists to close, so the rowcount check below is load-bearing: without it both callers would return the same suspension and both would go on to resume the same run.

The expiry predicate lives in that same UPDATE rather than in a check the caller runs afterwards, and the difference matters. A caller that claimed first and rejected an expired suspension second would already have flipped the run to running, leaving it stuck there with a suspension nobody can claim any more: strictly worse than the expired-but-paused state the sweeper exists to clean up. Conditioning the write on an unexpired suspension means an expired resume never moves the run at all.

func (*Store) ClearConversation 

func (s *Store) ClearConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID) error

ClearConversation deletes the session's message rows and resets its counters in the same transaction, for the same drift reason as SaveConversation.

func (*Store) ClearWorking 

func (s *Store) ClearWorking(ctx context.Context, runID id.AgentRunID) error

func (*Store) Close 

func (s *Store) Close() error

Close closes the database connection.

func (*Store) CountAgents 

func (s *Store) CountAgents(ctx context.Context, filter *agent.ListFilter) (int64, error)

func (*Store) CountBehaviors 

func (s *Store) CountBehaviors(ctx context.Context, filter *behavior.ListFilter) (int64, error)

func (*Store) CountOrchestrationRuns 

func (s *Store) CountOrchestrationRuns(ctx context.Context, filter *orchestration.RunListFilter) (int64, error)

func (*Store) CountOrchestrations 

func (s *Store) CountOrchestrations(ctx context.Context, filter *orchestration.ConfigListFilter) (int64, error)

func (*Store) CountPending 

func (s *Store) CountPending(ctx context.Context, filter *checkpoint.ListFilter) (int64, error)

func (*Store) CountPersonas 

func (s *Store) CountPersonas(ctx context.Context, filter *persona.ListFilter) (int64, error)

func (*Store) CountRuns 

func (s *Store) CountRuns(ctx context.Context, filter *run.ListFilter) (int64, error)

func (*Store) CountSessions 

func (s *Store) CountSessions(ctx context.Context, filter *session.ListFilter) (int64, error)

func (*Store) CountSkills 

func (s *Store) CountSkills(ctx context.Context, filter *skill.ListFilter) (int64, error)

func (*Store) CountTraits 

func (s *Store) CountTraits(ctx context.Context, filter *trait.ListFilter) (int64, error)

func (*Store) Create 

func (s *Store) Create(ctx context.Context, config *agent.Config) error

Create persists a new agent config, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.

func (*Store) CreateBehavior 

func (s *Store) CreateBehavior(ctx context.Context, b *behavior.Behavior) error

CreateBehavior persists a new behavior, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.

func (*Store) CreateCheckpoint 

func (s *Store) CreateCheckpoint(ctx context.Context, cp *checkpoint.Checkpoint) error

func (*Store) CreateConversation 

func (s *Store) CreateConversation(ctx context.Context, c *a2a.Conversation) error

func (*Store) CreateDelivery 

func (s *Store) CreateDelivery(ctx context.Context, d *a2a.Delivery) error

func (*Store) CreateMessage 

func (s *Store) CreateMessage(ctx context.Context, e *a2a.Envelope) error

CreateMessage persists an envelope, stamping the scope from the context. Envelopes are immutable once written, so there is no update counterpart.

func (*Store) CreateOrchestration 

func (s *Store) CreateOrchestration(ctx context.Context, c *orchestration.Config) error

CreateOrchestration persists a new orchestration config, stamping the scope from the context.

func (*Store) CreateOrchestrationRun 

func (s *Store) CreateOrchestrationRun(ctx context.Context, r *orchestration.Run) error

CreateOrchestrationRun persists a new orchestration run, stamping the scope from the context.

func (*Store) CreateOverlay 

func (s *Store) CreateOverlay(ctx context.Context, o *prompt.Overlay) error

CreateOverlay persists a new overlay, stamping the scope from the context. The partial UNIQUE (agent_id, scope_canon) is what enforces one overlay per agent per scope, so a second create for the same pair comes back as cortex.ErrAlreadyExists rather than quietly shadowing the first.

func (*Store) CreatePendingAsk 

func (s *Store) CreatePendingAsk(ctx context.Context, a *a2a.PendingAsk) error

func (*Store) CreatePersona 

func (s *Store) CreatePersona(ctx context.Context, p *persona.Persona) error

CreatePersona persists a new persona, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.

func (*Store) CreateRun 

func (s *Store) CreateRun(ctx context.Context, r *run.Run) error

func (*Store) CreateSession 

func (s *Store) CreateSession(ctx context.Context, sess *session.Session) error

CreateSession persists a new session, stamping the scope from the context. The partial UNIQUE (agent_id, scope_canon) WHERE is_default is what actually enforces one default session per agent per scope; a non-default session never touches that constraint.

func (*Store) CreateSkill 

func (s *Store) CreateSkill(ctx context.Context, sk *skill.Skill) error

CreateSkill persists a new skill, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.

func (*Store) CreateStep 

func (s *Store) CreateStep(ctx context.Context, step *run.Step) error

func (*Store) CreateSuspension 

func (s *Store) CreateSuspension(ctx context.Context, susp *suspension.Suspension) error

CreateSuspension writes a new suspension row for a run that just paused, stamping the scope from the context. The partial UNIQUE (run_id) WHERE the scope is non-empty is what enforces one suspension per run, which is what makes ClaimSuspension a meaningful primitive.

func (*Store) CreateToolCall 

func (s *Store) CreateToolCall(ctx context.Context, tc *run.ToolCall) error

func (*Store) CreateTrait 

func (s *Store) CreateTrait(ctx context.Context, t *trait.Trait) error

CreateTrait persists a new trait, stamping the scope from the context. UNIQUE (scope_canon, name) is what actually enforces per-scope name uniqueness; app_id survives on the row as a vestigial column no longer read or filtered on.

func (*Store) Delete 

func (s *Store) Delete(ctx context.Context, agentID id.AgentID) error

func (*Store) DeleteBehavior 

func (s *Store) DeleteBehavior(ctx context.Context, behaviorID id.BehaviorID) error

func (*Store) DeleteOrchestration 

func (s *Store) DeleteOrchestration(ctx context.Context, orchID id.OrchestrationConfigID) error

func (*Store) DeleteOverlay 

func (s *Store) DeleteOverlay(ctx context.Context, overlayID id.OverlayID) error

DeleteOverlay removes an overlay within the caller's scope.

func (*Store) DeletePersona 

func (s *Store) DeletePersona(ctx context.Context, personaID id.PersonaID) error

func (*Store) DeleteProgress added in v1.10.0 

func (s *Store) DeleteProgress(ctx context.Context, runID id.AgentRunID) error

DeleteProgress removes a run's snapshot at the exact context scope. A run with none is not an error.

func (*Store) DeleteSession 

func (s *Store) DeleteSession(ctx context.Context, sessionID id.SessionID) error

DeleteSession removes a session within the caller's scope, along with every conversation message it owns.

This is explicit application-level cascade, not a database FOREIGN KEY -- cortex_memories.session_id can't carry one. The column is shared by every memory kind, not just conversation: working and summary rows never set it at all, so they sit at its DEFAULT "" permanently. A real FOREIGN KEY (session_id) REFERENCES cortex_sessions(id) would require every row's session_id to resolve to an existing session, including those "" rows -- which would need a session literally named "" to satisfy, and would reject every SaveWorking/SaveSummary write the moment the constraint existed. NOT VALID only skips validating pre-existing rows at creation time; it does not exempt future inserts from being checked, so it doesn't get around this either. The delete below is what keeps postgres, sqlite, and mongo observably identical: deleting a session leaves no orphaned messages on any of the three, whichever mechanism gets it there.

func (*Store) DeleteSkill 

func (s *Store) DeleteSkill(ctx context.Context, skillID id.SkillID) error

func (*Store) DeleteSuspension 

func (s *Store) DeleteSuspension(ctx context.Context, runID id.AgentRunID) error

DeleteSuspension removes a run's suspension within the caller's scope.

func (*Store) DeleteTrait 

func (s *Store) DeleteTrait(ctx context.Context, traitID id.TraitID) error

func (*Store) Get 

func (s *Store) Get(ctx context.Context, agentID id.AgentID) (*agent.Config, error)

func (*Store) GetBehavior 

func (s *Store) GetBehavior(ctx context.Context, behaviorID id.BehaviorID) (*behavior.Behavior, error)

func (*Store) GetBehaviorByName 

func (s *Store) GetBehaviorByName(ctx context.Context, name string) (*behavior.Behavior, error)

func (*Store) GetByName 

func (s *Store) GetByName(ctx context.Context, name string) (*agent.Config, error)

func (*Store) GetCheckpoint 

func (s *Store) GetCheckpoint(ctx context.Context, cpID id.CheckpointID) (*checkpoint.Checkpoint, error)

func (*Store) GetConversation 

func (s *Store) GetConversation(ctx context.Context, convID id.ConversationID) (*a2a.Conversation, error)

func (*Store) GetConversationByPeerContext 

func (s *Store) GetConversationByPeerContext(ctx context.Context, node, peerContext string) (*a2a.Conversation, error)

GetConversationByPeerContext finds the conversation opened for a remote thread. The node is half the key: two peers can use the same context id, and joining one peer's thread to another's would leak a conversation across a trust boundary.

func (*Store) GetDelivery 

func (s *Store) GetDelivery(ctx context.Context, deliveryID id.DeliveryID) (*a2a.Delivery, error)

GetDelivery reads one delivery row within the caller's scope.

func (*Store) GetMessage 

func (s *Store) GetMessage(ctx context.Context, msgID id.MessageID) (*a2a.Envelope, error)

func (*Store) GetOrchestration 

func (s *Store) GetOrchestration(ctx context.Context, orchID id.OrchestrationConfigID) (*orchestration.Config, error)

func (*Store) GetOrchestrationByName 

func (s *Store) GetOrchestrationByName(ctx context.Context, name string) (*orchestration.Config, error)

GetOrchestrationByName returns an orchestration config by name within the caller's scope. Fix round 1 dropped the appID parameter this method used to also filter on: with UNIQUE (scope_canon, name), at most one row can ever exist per (scope, name), so an app_id predicate on top could only ever turn a hit into a miss, never disambiguate two rows — the same reasoning that dropped AppID from agent and persona.

func (*Store) GetOrchestrationRun 

func (s *Store) GetOrchestrationRun(ctx context.Context, runID id.OrchestrationID) (*orchestration.Run, error)

func (*Store) GetOverlay 

func (s *Store) GetOverlay(ctx context.Context, overlayID id.OverlayID) (*prompt.Overlay, error)

GetOverlay reads one overlay by id within the caller's scope.

func (*Store) GetOverlayForAgent 

func (s *Store) GetOverlayForAgent(ctx context.Context, agentID id.AgentID) (*prompt.Overlay, error)

GetOverlayForAgent reads the overlay an agent has at the caller's exact scope. See prompt.Store for why this one read matches exactly rather than by prefix.

func (*Store) GetOverlayForAgentAt 

func (s *Store) GetOverlayForAgentAt(ctx context.Context, agentID id.AgentID, scope cortex.Scope) (*prompt.Overlay, error)

GetOverlayForAgentAt reads the overlay an agent has at exactly the given scope. See prompt.Store for why inheritance is an ancestor walk rather than a prefix match, and why the scope argument is bounded to the caller's own ancestry.

func (*Store) GetPersona 

func (s *Store) GetPersona(ctx context.Context, personaID id.PersonaID) (*persona.Persona, error)

func (*Store) GetPersonaByName 

func (s *Store) GetPersonaByName(ctx context.Context, name string) (*persona.Persona, error)

func (*Store) GetProgress added in v1.10.0 

func (s *Store) GetProgress(ctx context.Context, runID id.AgentRunID) (*suspension.Progress, error)

GetProgress reads a run's snapshot at the exact context scope.

func (*Store) GetRun 

func (s *Store) GetRun(ctx context.Context, runID id.AgentRunID) (*run.Run, error)

func (*Store) GetSession 

func (s *Store) GetSession(ctx context.Context, sessionID id.SessionID) (*session.Session, error)

func (*Store) GetSkill 

func (s *Store) GetSkill(ctx context.Context, skillID id.SkillID) (*skill.Skill, error)

func (*Store) GetSkillByName 

func (s *Store) GetSkillByName(ctx context.Context, name string) (*skill.Skill, error)

func (*Store) GetSuspension 

func (s *Store) GetSuspension(ctx context.Context, runID id.AgentRunID) (*suspension.Suspension, error)

GetSuspension reads a run's suspension within the caller's scope without changing anything. Callers that intend to resume must use ClaimSuspension instead.

func (*Store) GetTrait 

func (s *Store) GetTrait(ctx context.Context, traitID id.TraitID) (*trait.Trait, error)

func (*Store) GetTraitByName 

func (s *Store) GetTraitByName(ctx context.Context, name string) (*trait.Trait, error)

func (*Store) HoldLease added in v1.10.0 

func (s *Store) HoldLease(ctx context.Context, runID id.AgentRunID, lease run.Lease) error

HoldLease sets or extends a running run's lease in one conditional UPDATE. The owner predicate is what keeps two workers from both holding the run: only an unowned lease or the caller's own matches.

func (*Store) InspectScopes 

func (s *Store) InspectScopes(ctx context.Context, visit func(store.ScopeIssue) error) error

InspectScopes streams invalid stored identities to an operator callback. This administrative scan crosses all scopes and never changes data.

func (*Store) LinkOrchestrationChild 

func (s *Store) LinkOrchestrationChild(ctx context.Context, team id.OrchestrationID, child id.AgentRunID) error

LinkOrchestrationChild atomically admits one child while its parent is running.

func (*Store) List 

func (s *Store) List(ctx context.Context, filter *agent.ListFilter) ([]*agent.Config, error)

func (*Store) ListAgentReplyRuns 

func (s *Store) ListAgentReplyRuns(ctx context.Context, after id.AgentRunID, limit int) ([]id.AgentRunID, error)

ListAgentReplyRuns includes saved pauses and running admissions whose pause was removed.

func (*Store) ListBehaviors 

func (s *Store) ListBehaviors(ctx context.Context, filter *behavior.ListFilter) ([]*behavior.Behavior, error)

func (*Store) ListContinuableAcrossScopes added in v1.10.0 

func (s *Store) ListContinuableAcrossScopes(ctx context.Context, notifiedBefore time.Time, limit int) ([]*run.Run, error)

ListContinuableAcrossScopes reads ready and reclaimable runs nobody was told about since notifiedBefore, from every scope, with no scope guard and no scope predicates. Like ListLapsedAcrossScopes it deliberately bypasses scope, for the sweeper only.

func (*Store) ListConversations 

func (s *Store) ListConversations(ctx context.Context, filter *a2a.ConversationListFilter) ([]*a2a.Conversation, error)

func (*Store) ListDeferredRecipients 

func (s *Store) ListDeferredRecipients(ctx context.Context, after id.DeliveryID, limit int) ([]id.DeliveryID, error)

ListDeferredRecipients projects bound pending delivery IDs in the exact scope.

func (*Store) ListExpired 

func (s *Store) ListExpired(ctx context.Context, now time.Time, limit int) ([]*suspension.Suspension, error)

ListExpired returns suspensions whose deadline has passed, oldest first, within the caller's scope.

func (*Store) ListExpiredAcrossScopes 

func (s *Store) ListExpiredAcrossScopes(ctx context.Context, now time.Time, limit int) ([]*suspension.Suspension, error)

ListExpiredAcrossScopes reads expired suspensions from every scope, with no scope guard and no scope predicates. That is deliberate and it is why the method is named for it: the expiry sweeper runs without a request scope, and a sweep that filtered would silently sweep nothing. See the suspension.Store doc for the full argument.

func (*Store) ListExpiredAsks 

func (s *Store) ListExpiredAsks(ctx context.Context, now time.Time, limit int) ([]*a2a.PendingAsk, error)

ListExpiredAsks returns unclaimed asks past their deadline. Claimed rows are excluded, which is what makes sweeping the same backlog twice safe.

func (*Store) ListInbox 

func (s *Store) ListInbox(ctx context.Context, agentName string, filter a2a.InboxFilter) ([]*a2a.Delivery, error)

ListInbox returns messages that have ARRIVED for an agent. A queued delivery is deliberately excluded: it has not reached anyone yet, and an inbox that showed it would be showing mail that is still in transit.

func (*Store) ListLapsedAcrossScopes added in v1.10.0 

func (s *Store) ListLapsedAcrossScopes(ctx context.Context, now time.Time, limit int) ([]*run.Run, error)

ListLapsedAcrossScopes reads running runs whose lease has lapsed from every scope, with no scope guard and no scope predicates. That is deliberate and it is why the method is named for it: the sweeper runs without a request scope, and a sweep that filtered would silently sweep nothing. It is for the sweeper only, which acts on each run under that run's own scope.

func (*Store) ListMessages 

func (s *Store) ListMessages(ctx context.Context, filter *a2a.MessageListFilter) ([]*a2a.Envelope, error)

ListMessages returns a conversation's messages oldest first, because a conversation is read as a transcript.

func (*Store) ListOrchestrationRuns 

func (s *Store) ListOrchestrationRuns(ctx context.Context, filter *orchestration.RunListFilter) ([]*orchestration.Run, error)

func (*Store) ListOrchestrations 

func (s *Store) ListOrchestrations(ctx context.Context, filter *orchestration.ConfigListFilter) ([]*orchestration.Config, error)

func (*Store) ListOverlays 

func (s *Store) ListOverlays(ctx context.Context, filter *prompt.ListFilter) ([]*prompt.Overlay, error)

ListOverlays returns overlays within the caller's scope, oldest first.

func (*Store) ListPending 

func (s *Store) ListPending(ctx context.Context, filter *checkpoint.ListFilter) ([]*checkpoint.Checkpoint, error)

func (*Store) ListPendingAcrossScopes added in v1.10.0 

func (s *Store) ListPendingAcrossScopes(ctx context.Context, after id.CheckpointID, limit int) ([]*checkpoint.Checkpoint, error)

ListPendingAcrossScopes reads pending checkpoints from every scope, in id order, with no scope guard and no scope predicates. That is deliberate: the sweeper has no request scope, and it acts on each checkpoint under that checkpoint's own scope.

func (*Store) ListPendingAsksByConversation 

func (s *Store) ListPendingAsksByConversation(ctx context.Context, convID id.ConversationID) ([]*a2a.PendingAsk, error)

func (*Store) ListPersonas 

func (s *Store) ListPersonas(ctx context.Context, filter *persona.ListFilter) ([]*persona.Persona, error)

func (*Store) ListQueuedDeliveries 

func (s *Store) ListQueuedDeliveries(ctx context.Context, limit int) ([]*a2a.Delivery, error)

ListQueuedDeliveries discovers work within the selected routing partition. Roster-bound buses also select their exact captured scope. Other dispatchers discover across scopes and restore each delivery's scope before claiming it.

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, filter *run.ListFilter) ([]*run.Run, error)

func (*Store) ListSessions 

func (s *Store) ListSessions(ctx context.Context, filter *session.ListFilter) ([]*session.Session, error)

func (*Store) ListSkills 

func (s *Store) ListSkills(ctx context.Context, filter *skill.ListFilter) ([]*skill.Skill, error)

func (*Store) ListSteps 

func (s *Store) ListSteps(ctx context.Context, runID id.AgentRunID) ([]*run.Step, error)

func (*Store) ListToolCalls 

func (s *Store) ListToolCalls(ctx context.Context, stepID id.StepID) ([]*run.ToolCall, error)

func (*Store) ListTraits 

func (s *Store) ListTraits(ctx context.Context, filter *trait.ListFilter) ([]*trait.Trait, error)

func (*Store) LoadConversation 

func (s *Store) LoadConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID, limit int) ([]memory.Message, error)

LoadConversation returns the NEWEST `limit` messages in chronological order, not the oldest. The query orders created_at/id DESC and takes LIMIT off the front -- the recent end of the conversation -- then the loop below walks the result backwards to hand callers chronological order again. Ordering ASC and limiting (the old behavior) always returns the SAME oldest prefix once a conversation grows past the limit: every later turn is written but never read back into context, and an install that had already accumulated duplicate rows from the pre-fix re-save bug stayed pinned on that stale prefix even after the duplication itself was fixed upstream. id DESC breaks ties within a batch that shares one created_at timestamp, so the reversal is a stable chronological order rather than an arbitrary one.

func (*Store) LoadSummaries 

func (s *Store) LoadSummaries(ctx context.Context, agentID id.AgentID) ([]string, error)

func (*Store) LoadWorking 

func (s *Store) LoadWorking(ctx context.Context, runID id.AgentRunID, key string) (any, error)

func (*Store) MarkDeliveryRead 

func (s *Store) MarkDeliveryRead(ctx context.Context, deliveryID id.DeliveryID) error

func (*Store) MarkNotified added in v1.10.0 

func (s *Store) MarkNotified(ctx context.Context, runID id.AgentRunID, at time.Time) error

MarkNotified records when the host was last told about a waiting run.

func (*Store) Migrate 

func (s *Store) Migrate(ctx context.Context, opts ...cortex.MigrateOption) error

Migrate runs programmatic migrations via the grove orchestrator.

It holds the cortex migration lock (migrateLockKey) for the whole call, so replicas starting together on one database run it one at a time: a second caller waits for the first to finish rather than failing. The lock also covers grove's own tracking-table creation and the backfills after the migration group. Cancelling ctx stops the wait. A lost race on grove's tracking tables against another grove user is retried (see migrateGroup).

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

Ping verifies the database connection.

func (*Store) ReclaimStaleDeliveries 

func (s *Store) ReclaimStaleDeliveries(ctx context.Context, olderThan time.Time, limit int) (int, error)

ReclaimStaleDeliveries requeues abandoned deliveries without a bound run. It uses the same partition and optional exact scope as queue discovery. The cutoff and claim token are rechecked on each update.

func (*Store) Resolve 

func (s *Store) Resolve(ctx context.Context, cpID id.CheckpointID, decision checkpoint.Decision) error

func (*Store) SaveConversation 

func (s *Store) SaveConversation(ctx context.Context, agentID id.AgentID, sessionID id.SessionID, messages []memory.Message) error

SaveConversation inserts messages and updates the owning session's message_count/last_message inside one transaction. The counter is derived from the rows this call actually writes (len(messages), not a flat +1 per call), so it always matches the physical row count even when a caller's batch size varies from call to call. If the session row can't be found — a session id that doesn't correspond to a real session — the whole write rolls back rather than leaving orphaned message rows a reader can never reach through GetSession/ListSessions.

func (*Store) SaveSummary 

func (s *Store) SaveSummary(ctx context.Context, agentID id.AgentID, summary string) error

func (*Store) SaveWorking 

func (s *Store) SaveWorking(ctx context.Context, runID id.AgentRunID, key string, value any) error

SaveWorking upserts a working-memory value for a run within the caller's scope. A run ID is a bearer capability, not an isolation boundary — anyone who learns it could otherwise read or clobber another tenant's scratch state — so this is guarded the same as SaveConversation.

func (*Store) SupportsOrchestrationRouting 

func (s *Store) SupportsOrchestrationRouting() bool

SupportsOrchestrationRouting reports durable A2A partition support.

func (*Store) SupportsScopedOrchestrationDiscovery 

func (s *Store) SupportsScopedOrchestrationDiscovery() bool

SupportsScopedOrchestrationDiscovery reports exact-scope background routing.

func (*Store) Update 

func (s *Store) Update(ctx context.Context, config *agent.Config) error

Update modifies an existing agent config's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the agent's stored scope to touch it), and is never written back — mutableAgentColumns excludes the five scope columns from what gets written.

func (*Store) UpdateBehavior 

func (s *Store) UpdateBehavior(ctx context.Context, b *behavior.Behavior) error

UpdateBehavior modifies an existing behavior's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the behavior's stored scope to touch it), and is never written back — mutableBehaviorColumns excludes the five scope columns from what gets written.

func (*Store) UpdateConversation 

func (s *Store) UpdateConversation(ctx context.Context, c *a2a.Conversation) error

func (*Store) UpdateDelivery 

func (s *Store) UpdateDelivery(ctx context.Context, d *a2a.Delivery) error

func (*Store) UpdateOrchestration 

func (s *Store) UpdateOrchestration(ctx context.Context, c *orchestration.Config) error

UpdateOrchestration modifies an existing orchestration config's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the config's stored scope to touch it), and is never written back — mutableOrchestrationConfigColumns excludes the five scope columns from what gets written.

func (*Store) UpdateOrchestrationRun 

func (s *Store) UpdateOrchestrationRun(ctx context.Context, r *orchestration.Run) error

UpdateOrchestrationRun modifies an existing orchestration run's mutable fields within the caller's scope. Scope is immutable after creation, same as UpdateOrchestration above.

func (*Store) UpdateOverlay 

func (s *Store) UpdateOverlay(ctx context.Context, o *prompt.Overlay) error

UpdateOverlay rewrites an overlay's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the overlay's stored scope to touch it), and is never written back.

func (*Store) UpdatePersona 

func (s *Store) UpdatePersona(ctx context.Context, p *persona.Persona) error

UpdatePersona modifies an existing persona's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the persona's stored scope to touch it), and is never written back — mutablePersonaColumns excludes the five scope columns from what gets written.

func (*Store) UpdateRun 

func (s *Store) UpdateRun(ctx context.Context, r *run.Run) error

UpdateRun updates a run's mutable fields. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the run's stored scope to touch it), and is never written back. r.Scope is left untouched — whatever runToModel derives from it is excluded from the SET clause by mutableRunColumns.

func (*Store) UpdateSession 

func (s *Store) UpdateSession(ctx context.Context, sess *session.Session) error

UpdateSession modifies an existing session's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the session's stored scope to touch it), and is never written back — mutableSessionColumns excludes the five scope columns from what gets written.

func (*Store) UpdateSkill 

func (s *Store) UpdateSkill(ctx context.Context, sk *skill.Skill) error

UpdateSkill modifies an existing skill's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the skill's stored scope to touch it), and is never written back — mutableSkillColumns excludes the five scope columns from what gets written.

func (*Store) UpdateTrait 

func (s *Store) UpdateTrait(ctx context.Context, t *trait.Trait) error

UpdateTrait modifies an existing trait's mutable fields within the caller's scope. Scope is immutable after creation: the context scope is used only as an authorization predicate (the caller must be at the exact scope of the trait's stored scope to touch it), and is never written back — mutableTraitColumns excludes the five scope columns from what gets written.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.