wehub-resource-sync
f99010fae1
Desktop Artifacts / Desktop Build (Linux) (push) Waiting to run
Desktop Artifacts / Desktop Build (Windows) (push) Waiting to run
Desktop Artifacts / Desktop Build (Linux (arm64)) (push) Waiting to run
Desktop Artifacts (macOS) / Desktop Build (macOS (aarch64)) (push) Waiting to run
Desktop Artifacts (macOS) / Desktop Build (macOS (x86_64)) (push) Waiting to run
CI / lint (push) Failing after 1s
CI / frontend (push) Failing after 1s
CI / scripts (push) Failing after 1s
CI / Go Test (ubuntu-latest) (push) Failing after 0s
CI / frontend-node-25 (push) Failing after 1s
CI / docs (push) Failing after 0s
CI / coverage (push) Failing after 0s
CI / e2e (push) Failing after 0s
Docker / build-and-push (push) Failing after 1s
CI / integration (push) Failing after 4m43s
CI / Go Test (windows-latest) (push) Has been cancelled
CI / Desktop Unit Tests (Windows) (push) Has been cancelled
1589 lines
44 KiB
Go
1589 lines
44 KiB
Go
//go:build pgtest
|
|
|
|
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"regexp"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"go.kenn.io/agentsview/internal/db"
|
|
"go.kenn.io/agentsview/internal/export"
|
|
)
|
|
|
|
func cleanPGSchema(t *testing.T, pgURL string) {
|
|
t.Helper()
|
|
pg, err := sql.Open("pgx", pgURL)
|
|
require.NoError(t, err, "connecting to pg")
|
|
defer pg.Close()
|
|
_, _ = pg.Exec(
|
|
"DROP SCHEMA IF EXISTS agentsview CASCADE",
|
|
)
|
|
}
|
|
|
|
func cleanNamedPGSchema(t *testing.T, pgURL, schema string) {
|
|
t.Helper()
|
|
pg, err := sql.Open("pgx", pgURL)
|
|
require.NoError(t, err, "connecting to pg")
|
|
defer pg.Close()
|
|
quoted, err := quoteIdentifier(schema)
|
|
require.NoError(t, err, "quote schema")
|
|
_, _ = pg.Exec("DROP SCHEMA IF EXISTS " + quoted + " CASCADE")
|
|
}
|
|
|
|
func TestEnsureSchemaIdempotent(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
|
|
require.NoError(t, ps.EnsureSchema(ctx), "first EnsureSchema")
|
|
require.NoError(t, ps.EnsureSchema(ctx), "second EnsureSchema")
|
|
|
|
var eventIndex int
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT event_index FROM tool_result_events LIMIT 0",
|
|
).Scan(&eventIndex)
|
|
if err != nil && err != sql.ErrNoRows {
|
|
t.Fatalf("tool_result_events schema probe: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestSyncEffectiveSyncStateFallsBackToLocalDB(t *testing.T) {
|
|
local := testDB(t)
|
|
require.NoError(t, local.SetSyncState(
|
|
"last_push_at",
|
|
"2026-03-11T12:34:56.123456789Z",
|
|
))
|
|
|
|
sync := &Sync{local: local}
|
|
require.NoError(t, NormalizeLocalSyncStateTimestamps(
|
|
sync.effectiveSyncState(),
|
|
))
|
|
|
|
got, err := sync.effectiveSyncState().GetSyncState("last_push_at")
|
|
require.NoError(t, err)
|
|
assert.Equal(t, "2026-03-11T12:34:56.123Z", got)
|
|
}
|
|
|
|
func TestSyncScopedStateUsesTargetKeys(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{
|
|
SyncStateTarget: "work",
|
|
MigrateLegacySyncState: true,
|
|
},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := "2026-03-11T12:00:00Z"
|
|
require.NoError(t, local.UpsertSession(db.Session{
|
|
ID: "sess-scoped-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: 1,
|
|
}), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{{
|
|
SessionID: "sess-scoped-001",
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "hello",
|
|
}}), "insert message")
|
|
|
|
_, err = ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
|
|
scopedLastPush, err := local.GetSyncState("last_push_at:work")
|
|
require.NoError(t, err)
|
|
assert.NotEmpty(t, scopedLastPush)
|
|
|
|
legacyLastPush, err := local.GetSyncState("last_push_at")
|
|
require.NoError(t, err)
|
|
assert.Empty(t, legacyLastPush)
|
|
|
|
scopedBoundary, err := local.GetSyncState(
|
|
lastPushBoundaryStateKey + ":work",
|
|
)
|
|
require.NoError(t, err)
|
|
assert.NotEmpty(t, scopedBoundary)
|
|
|
|
scopedFingerprint, err := local.GetSyncState(
|
|
lastPushTargetFingerprintKey + ":work",
|
|
)
|
|
require.NoError(t, err)
|
|
assert.NotEmpty(t, scopedFingerprint)
|
|
|
|
status, err := ps.Status(ctx)
|
|
require.NoError(t, err, "status")
|
|
assert.Equal(t, scopedLastPush, status.LastPushAt)
|
|
}
|
|
|
|
func TestEnsureSchemaMigratesLegacySchema(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
pg, err := Open(pgURL, "agentsview", true)
|
|
require.NoError(t, err, "connecting to pg")
|
|
defer pg.Close()
|
|
|
|
ctx := context.Background()
|
|
|
|
// Simulate a 0.16.x schema: create the schema and core
|
|
// tables but omit tool_result_events.
|
|
_, err = pg.ExecContext(ctx,
|
|
"CREATE SCHEMA IF NOT EXISTS agentsview",
|
|
)
|
|
require.NoError(t, err, "creating schema")
|
|
legacyDDL := `
|
|
CREATE TABLE IF NOT EXISTS sync_metadata (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
id TEXT PRIMARY KEY,
|
|
machine TEXT NOT NULL,
|
|
project TEXT NOT NULL,
|
|
agent TEXT NOT NULL,
|
|
first_message TEXT,
|
|
display_name TEXT,
|
|
created_at TIMESTAMPTZ,
|
|
started_at TIMESTAMPTZ,
|
|
ended_at TIMESTAMPTZ,
|
|
deleted_at TIMESTAMPTZ,
|
|
message_count INT NOT NULL DEFAULT 0,
|
|
user_message_count INT NOT NULL DEFAULT 0,
|
|
parent_session_id TEXT,
|
|
relationship_type TEXT NOT NULL DEFAULT '',
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
session_id TEXT NOT NULL,
|
|
ordinal INT NOT NULL,
|
|
role TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
timestamp TIMESTAMPTZ,
|
|
has_thinking BOOLEAN NOT NULL DEFAULT FALSE,
|
|
has_tool_use BOOLEAN NOT NULL DEFAULT FALSE,
|
|
content_length INT NOT NULL DEFAULT 0,
|
|
is_system BOOLEAN NOT NULL DEFAULT FALSE,
|
|
PRIMARY KEY (session_id, ordinal),
|
|
FOREIGN KEY (session_id)
|
|
REFERENCES sessions(id) ON DELETE CASCADE
|
|
);
|
|
CREATE TABLE IF NOT EXISTS tool_calls (
|
|
id BIGSERIAL PRIMARY KEY,
|
|
session_id TEXT NOT NULL,
|
|
tool_name TEXT NOT NULL,
|
|
category TEXT NOT NULL,
|
|
call_index INT NOT NULL DEFAULT 0,
|
|
tool_use_id TEXT NOT NULL DEFAULT '',
|
|
input_json TEXT,
|
|
skill_name TEXT,
|
|
result_content_length INT,
|
|
result_content TEXT,
|
|
subagent_session_id TEXT,
|
|
message_ordinal INT NOT NULL,
|
|
FOREIGN KEY (session_id)
|
|
REFERENCES sessions(id) ON DELETE CASCADE
|
|
);`
|
|
_, err = pg.ExecContext(ctx, legacyDDL)
|
|
require.NoError(t, err, "creating legacy tables")
|
|
|
|
// Verify tool_result_events does not exist yet.
|
|
require.Error(t, CheckSchemaCompat(ctx, pg),
|
|
"expected CheckSchemaCompat to fail on legacy schema")
|
|
|
|
// Run EnsureSchema — should create the missing table.
|
|
require.NoError(t, EnsureSchema(ctx, pg, "agentsview"),
|
|
"EnsureSchema on legacy schema")
|
|
|
|
// Now the compat check should pass.
|
|
require.NoError(t, CheckSchemaCompat(ctx, pg),
|
|
"CheckSchemaCompat after migration")
|
|
}
|
|
|
|
// TestCheckSchemaCompatMissingSecretsRulesVersion pins the schema-compat
|
|
// probe against a regression where pgSessionCols selects a column the
|
|
// compat check doesn't probe for: a PG schema missing only
|
|
// sessions.secrets_rules_version must fail CheckSchemaCompat rather
|
|
// than passing the probe and 500-ing at runtime on the first session
|
|
// query. EnsureSchema brings the column in via migration; this test
|
|
// then drops it to simulate a legacy/read-only schema and verifies the
|
|
// probe catches the absence.
|
|
func TestCheckSchemaCompatMissingSecretsRulesVersion(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
pg, err := Open(pgURL, "agentsview", true)
|
|
require.NoError(t, err, "connecting to pg")
|
|
defer pg.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, EnsureSchema(ctx, pg, "agentsview"), "EnsureSchema")
|
|
require.NoError(t, CheckSchemaCompat(ctx, pg),
|
|
"precondition: CheckSchemaCompat should pass after EnsureSchema")
|
|
_, err = pg.ExecContext(ctx,
|
|
`ALTER TABLE sessions DROP COLUMN secrets_rules_version`,
|
|
)
|
|
require.NoError(t, err, "dropping secrets_rules_version")
|
|
require.Error(t, CheckSchemaCompat(ctx, pg),
|
|
"CheckSchemaCompat should fail when secrets_rules_version is missing")
|
|
}
|
|
|
|
func TestPushSingleSession(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := "2026-03-11T12:00:00Z"
|
|
firstMsg := "hello world"
|
|
sess := db.Session{
|
|
ID: "sess-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
FirstMessage: &firstMsg,
|
|
StartedAt: &started,
|
|
MessageCount: 1,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: "sess-001",
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: firstMsg,
|
|
},
|
|
}), "insert messages")
|
|
|
|
result, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
assert.Equal(t, 1, result.SessionsPushed)
|
|
assert.Equal(t, 1, result.MessagesPushed)
|
|
|
|
var pgProject, pgMachine string
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT project, machine FROM sessions WHERE id = $1",
|
|
"sess-001",
|
|
).Scan(&pgProject, &pgMachine)
|
|
require.NoError(t, err, "querying pg session")
|
|
assert.Equal(t, "test-project", pgProject)
|
|
assert.Equal(t, "test-machine", pgMachine)
|
|
|
|
var pgMsgContent string
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT content FROM messages WHERE session_id = $1 AND ordinal = 0",
|
|
"sess-001",
|
|
).Scan(&pgMsgContent)
|
|
require.NoError(t, err, "querying pg message")
|
|
assert.Equal(t, firstMsg, pgMsgContent)
|
|
}
|
|
|
|
func TestPushIdempotent(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: "sess-002",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: 0,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
|
|
result1, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first push")
|
|
assert.Equal(t, 1, result1.SessionsPushed)
|
|
|
|
result2, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "second push")
|
|
assert.Equal(t, 0, result2.SessionsPushed)
|
|
}
|
|
|
|
func TestPushWithToolCalls(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: "sess-tc-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: 1,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: "sess-tc-001",
|
|
Ordinal: 0,
|
|
Role: "assistant",
|
|
Content: "tool use response",
|
|
HasToolUse: true,
|
|
ToolCalls: []db.ToolCall{
|
|
{
|
|
ToolName: "Read",
|
|
Category: "Read",
|
|
ToolUseID: "toolu_001",
|
|
ResultContentLength: 42,
|
|
ResultContent: "file content here",
|
|
},
|
|
},
|
|
},
|
|
}), "insert messages")
|
|
|
|
result, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
assert.Equal(t, 1, result.MessagesPushed)
|
|
|
|
var toolName string
|
|
var resultLen int
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT tool_name, result_content_length FROM tool_calls WHERE session_id = $1",
|
|
"sess-tc-001",
|
|
).Scan(&toolName, &resultLen)
|
|
require.NoError(t, err, "querying pg tool_call")
|
|
assert.Equal(t, "Read", toolName)
|
|
assert.Equal(t, 42, resultLen)
|
|
}
|
|
|
|
func TestPushWithToolResultEvents(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
sess := db.Session{
|
|
ID: "sess-events-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "codex",
|
|
MessageCount: 1,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: "sess-events-001",
|
|
Ordinal: 0,
|
|
Role: "assistant",
|
|
Content: "tool use response",
|
|
HasToolUse: true,
|
|
ToolCalls: []db.ToolCall{
|
|
{
|
|
ToolName: "wait",
|
|
Category: "Task",
|
|
ToolUseID: "call_wait",
|
|
ResultEvents: []db.ToolResultEvent{
|
|
{
|
|
ToolUseID: "call_wait",
|
|
AgentID: "agent-1",
|
|
SubagentSessionID: "codex:agent-1",
|
|
Source: "wait_output",
|
|
Status: "completed",
|
|
Content: "first result",
|
|
ContentLength: len("first result"),
|
|
Timestamp: "2026-03-27T10:00:00Z",
|
|
EventIndex: 0,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}), "insert messages")
|
|
|
|
_, err = ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
|
|
var count int
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM tool_result_events WHERE session_id = $1",
|
|
"sess-events-001",
|
|
).Scan(&count)
|
|
require.NoError(t, err, "querying pg tool_result_events")
|
|
assert.Equal(t, 1, count)
|
|
}
|
|
|
|
func TestStatus(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
status, err := ps.Status(ctx)
|
|
require.NoError(t, err, "status")
|
|
assert.Equal(t, "test-machine", status.Machine)
|
|
assert.Equal(t, 0, status.PGSessions)
|
|
}
|
|
|
|
func TestStatusMissingSchema(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
status, err := ps.Status(ctx)
|
|
require.NoError(t, err, "status on missing schema")
|
|
assert.Equal(t, 0, status.PGSessions)
|
|
assert.Equal(t, 0, status.PGMessages)
|
|
assert.Equal(t, "test-machine", status.Machine)
|
|
}
|
|
|
|
func TestNewRejectsMachineLocal(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
local := testDB(t)
|
|
_, err := New(
|
|
pgURL, "agentsview", local, "local", true,
|
|
SyncOptions{},
|
|
)
|
|
require.Error(t, err, "expected error for machine=local")
|
|
}
|
|
|
|
func TestNewRejectsEmptyMachine(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
local := testDB(t)
|
|
_, err := New(
|
|
pgURL, "agentsview", local, "", true,
|
|
SyncOptions{},
|
|
)
|
|
require.Error(t, err, "expected error for empty machine")
|
|
}
|
|
|
|
func TestNewRejectsEmptyURL(t *testing.T) {
|
|
local := testDB(t)
|
|
_, err := New(
|
|
"", "agentsview", local, "test", true,
|
|
SyncOptions{},
|
|
)
|
|
require.Error(t, err, "expected error for empty URL")
|
|
}
|
|
|
|
func TestPushUpdatedAtFormat(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: "sess-ts-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
|
|
_, err = ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
|
|
var updatedAt time.Time
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT updated_at FROM sessions WHERE id = $1",
|
|
"sess-ts-001",
|
|
).Scan(&updatedAt)
|
|
require.NoError(t, err, "querying updated_at")
|
|
|
|
formatted := updatedAt.UTC().Format(
|
|
"2006-01-02T15:04:05.000000Z",
|
|
)
|
|
pattern := regexp.MustCompile(
|
|
`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$`,
|
|
)
|
|
assert.True(t, pattern.MatchString(formatted),
|
|
"updated_at = %q, want ISO-8601 microsecond format", formatted)
|
|
}
|
|
|
|
func TestPushBumpsUpdatedAtOnMessageRewrite(
|
|
t *testing.T,
|
|
) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"machine-a", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := time.Now().UTC().Format(time.RFC3339)
|
|
sess := db.Session{
|
|
ID: "sess-bump-001",
|
|
Project: "test",
|
|
Machine: "local",
|
|
Agent: "test-agent",
|
|
StartedAt: &started,
|
|
MessageCount: 1,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
msg := db.Message{
|
|
SessionID: "sess-bump-001",
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "hello",
|
|
ContentLength: 5,
|
|
}
|
|
require.NoError(t, local.ReplaceSessionMessages(
|
|
"sess-bump-001", []db.Message{msg},
|
|
), "replace messages")
|
|
|
|
_, err = ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial push")
|
|
|
|
var updatedAt1 time.Time
|
|
require.NoError(t, ps.pg.QueryRowContext(ctx,
|
|
"SELECT updated_at FROM sessions WHERE id = $1",
|
|
"sess-bump-001",
|
|
).Scan(&updatedAt1), "querying updated_at")
|
|
|
|
time.Sleep(50 * time.Millisecond)
|
|
|
|
result, err := ps.Push(ctx, true, nil)
|
|
require.NoError(t, err, "full push")
|
|
require.NotZero(t, result.MessagesPushed,
|
|
"expected messages to be pushed on full push")
|
|
|
|
var updatedAt2 time.Time
|
|
require.NoError(t, ps.pg.QueryRowContext(ctx,
|
|
"SELECT updated_at FROM sessions WHERE id = $1",
|
|
"sess-bump-001",
|
|
).Scan(&updatedAt2), "querying updated_at after full push")
|
|
|
|
assert.True(t, updatedAt2.After(updatedAt1),
|
|
"updated_at not bumped: before=%v, after=%v", updatedAt1, updatedAt2)
|
|
}
|
|
|
|
func TestPushFullBypassesHeuristic(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: "sess-full-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: 1,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: "sess-full-001",
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "test",
|
|
},
|
|
}), "insert messages")
|
|
|
|
_, err = ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first push")
|
|
|
|
require.NoError(t, local.SetSyncState(
|
|
"last_push_at", "",
|
|
), "resetting watermark")
|
|
|
|
result, err := ps.Push(ctx, true, nil)
|
|
require.NoError(t, err, "full push")
|
|
assert.Equal(t, 1, result.SessionsPushed)
|
|
assert.Equal(t, 1, result.MessagesPushed)
|
|
}
|
|
|
|
func TestPushDetectsSchemaReset(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
// Push a session so the watermark advances.
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: "sess-reset-001",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: 1,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{{
|
|
SessionID: "sess-reset-001",
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "hello",
|
|
ContentLength: 5,
|
|
}}), "insert message")
|
|
|
|
r1, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial push")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
// Simulate a PG schema reset — don't manually recreate;
|
|
// let Push detect and handle it via the coherence check.
|
|
cleanPGSchema(t, pgURL)
|
|
|
|
// An incremental push should detect the mismatch
|
|
// (local watermark set, PG has 0 sessions), recreate
|
|
// the schema, and automatically force a full push.
|
|
r2, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "post-reset push")
|
|
assert.Equal(t, 1, r2.SessionsPushed,
|
|
"should auto-detect schema reset")
|
|
assert.Equal(t, 1, r2.MessagesPushed)
|
|
}
|
|
|
|
func TestPushDetectsPGTargetChange(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_a")
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_b")
|
|
t.Cleanup(func() {
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_a")
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_b")
|
|
})
|
|
|
|
local := testDB(t)
|
|
ctx := context.Background()
|
|
|
|
insertSession := func(id, createdAt string) {
|
|
require.NoError(t, local.UpsertSession(db.Session{
|
|
ID: id,
|
|
Project: "target-change",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
CreatedAt: createdAt,
|
|
MessageCount: 1,
|
|
}), "upsert session %s", id)
|
|
require.NoError(t, local.InsertMessages([]db.Message{{
|
|
SessionID: id,
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "hello " + id,
|
|
ContentLength: len("hello " + id),
|
|
Timestamp: createdAt,
|
|
}}), "insert message %s", id)
|
|
}
|
|
|
|
insertSession("sess-target-001", "2026-03-11T12:00:00Z")
|
|
|
|
syncA, err := New(
|
|
pgURL, "agentsview_a", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync A")
|
|
defer syncA.Close()
|
|
|
|
syncB, err := New(
|
|
pgURL, "agentsview_b", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync B")
|
|
defer syncB.Close()
|
|
|
|
r1, err := syncA.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial push to schema A")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
r2, err := syncB.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial push to schema B")
|
|
require.Equal(t, 1, r2.SessionsPushed)
|
|
|
|
insertSession("sess-target-002", "2026-03-11T12:10:00Z")
|
|
|
|
r3, err := syncA.Push(ctx, false, nil)
|
|
require.NoError(t, err, "incremental push back to schema A")
|
|
require.GreaterOrEqual(t, r3.SessionsPushed, 1)
|
|
|
|
r4, err := syncB.Push(ctx, false, nil)
|
|
require.NoError(t, err, "switch back to populated schema B")
|
|
require.Equal(t, 2, r4.SessionsPushed)
|
|
|
|
var count int
|
|
err = syncB.DB().QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM sessions WHERE id = $1",
|
|
"sess-target-002",
|
|
).Scan(&count)
|
|
require.NoError(t, err, "counting repushed session")
|
|
assert.Equal(t, 1, count)
|
|
}
|
|
|
|
func TestPushDetectsPGTargetChangeAfterFilteredPush(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_filtered_a")
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_filtered_b")
|
|
t.Cleanup(func() {
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_filtered_a")
|
|
cleanNamedPGSchema(t, pgURL, "agentsview_filtered_b")
|
|
})
|
|
|
|
local := testDB(t)
|
|
ctx := context.Background()
|
|
|
|
const project = "alpha"
|
|
const createdAt = "2026-03-11T12:00:00Z"
|
|
require.NoError(t, local.UpsertSession(db.Session{
|
|
ID: "sess-filtered-target-001",
|
|
Project: project,
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
CreatedAt: createdAt,
|
|
MessageCount: 1,
|
|
}), "upsert session")
|
|
require.NoError(t, local.InsertMessages([]db.Message{{
|
|
SessionID: "sess-filtered-target-001",
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "hello filtered target",
|
|
ContentLength: len("hello filtered target"),
|
|
Timestamp: createdAt,
|
|
}}), "insert message")
|
|
|
|
filteredA, err := New(
|
|
pgURL, "agentsview_filtered_a", local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{project}},
|
|
)
|
|
require.NoError(t, err, "creating filtered sync A")
|
|
defer filteredA.Close()
|
|
|
|
unfilteredB, err := New(
|
|
pgURL, "agentsview_filtered_b", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating unfiltered sync B")
|
|
defer unfilteredB.Close()
|
|
|
|
r1, err := filteredA.Push(ctx, false, nil)
|
|
require.NoError(t, err, "filtered push to schema A")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
lastPush, err := local.GetSyncState("last_push_at")
|
|
require.NoError(t, err, "reading filtered watermark")
|
|
assert.Empty(t, lastPush,
|
|
"filtered push should keep global last_push_at empty")
|
|
|
|
scopedLastPush, err := filteredA.effectiveSyncState().GetSyncState("last_push_at")
|
|
require.NoError(t, err, "reading filtered scoped watermark")
|
|
assert.NotEmpty(t, scopedLastPush,
|
|
"filtered push should advance scoped last_push_at")
|
|
|
|
boundaryState, err := filteredA.effectiveSyncState().GetSyncState(lastPushBoundaryStateKey)
|
|
require.NoError(t, err, "reading filtered boundary state")
|
|
require.NotEmpty(t, boundaryState,
|
|
"filtered push should persist boundary fingerprints")
|
|
|
|
r2, err := unfilteredB.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push to schema B after filtered target change")
|
|
require.Equal(t, 1, r2.SessionsPushed)
|
|
|
|
var count int
|
|
err = unfilteredB.DB().QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM sessions WHERE id = $1",
|
|
"sess-filtered-target-001",
|
|
).Scan(&count)
|
|
require.NoError(t, err, "counting session in schema B")
|
|
assert.Equal(t, 1, count)
|
|
}
|
|
|
|
func TestPushFullAfterSchemaDropRecreatesSchema(
|
|
t *testing.T,
|
|
) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
ctx := context.Background()
|
|
|
|
sess := db.Session{
|
|
ID: "sess-full-drop",
|
|
Project: "proj",
|
|
Machine: "test-machine",
|
|
Agent: "claude",
|
|
CreatedAt: "2026-03-11T12:00:00.000Z",
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
|
|
r1, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial push")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
// Drop the schema without clearing local state.
|
|
cleanPGSchema(t, pgURL)
|
|
|
|
// A full push should recreate the schema even though
|
|
// schemaDone is memoized from the first push.
|
|
r2, err := ps.Push(ctx, true, nil)
|
|
require.NoError(t, err, "full push after drop")
|
|
assert.Equal(t, 1, r2.SessionsPushed)
|
|
}
|
|
|
|
func TestScopedPushFullAfterSchemaDropRecreatesSchema(
|
|
t *testing.T,
|
|
) {
|
|
pgURL := testPGURL(t)
|
|
const schema = "agentsview_scoped_full_drop"
|
|
cleanNamedPGSchema(t, pgURL, schema)
|
|
t.Cleanup(func() { cleanNamedPGSchema(t, pgURL, schema) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, schema, local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{"proj"}},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
ctx := context.Background()
|
|
|
|
sess := db.Session{
|
|
ID: "sess-scoped-full-drop",
|
|
Project: "proj",
|
|
Machine: "test-machine",
|
|
Agent: "claude",
|
|
CreatedAt: "2026-03-11T12:00:00.000Z",
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
|
|
r1, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial push")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
cleanNamedPGSchema(t, pgURL, schema)
|
|
|
|
r2, err := ps.Push(ctx, true, nil)
|
|
require.NoError(t, err, "scoped full push after drop")
|
|
assert.Equal(t, 1, r2.SessionsPushed)
|
|
}
|
|
|
|
func TestPushBatchesMultipleSessions(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
// Create 75 sessions to exercise two batches (50 + 25).
|
|
const totalSessions = 75
|
|
for i := range totalSessions {
|
|
id := fmt.Sprintf("batch-sess-%03d", i)
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: id,
|
|
Project: "batch-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: 2,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess),
|
|
"upsert session %d", i)
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: id,
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: fmt.Sprintf("msg %d", i),
|
|
ContentLength: 5,
|
|
},
|
|
{
|
|
SessionID: id,
|
|
Ordinal: 1,
|
|
Role: "assistant",
|
|
Content: fmt.Sprintf("reply %d", i),
|
|
ContentLength: 7,
|
|
},
|
|
}), "insert messages %d", i)
|
|
}
|
|
|
|
result, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
assert.Equal(t, totalSessions, result.SessionsPushed)
|
|
assert.Equal(t, totalSessions*2, result.MessagesPushed)
|
|
assert.Equal(t, 0, result.Errors)
|
|
|
|
// Verify PG state.
|
|
var pgSessions, pgMessages int
|
|
require.NoError(t, ps.pg.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM sessions",
|
|
).Scan(&pgSessions), "counting pg sessions")
|
|
require.NoError(t, ps.pg.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM messages",
|
|
).Scan(&pgMessages), "counting pg messages")
|
|
assert.Equal(t, totalSessions, pgSessions)
|
|
assert.Equal(t, totalSessions*2, pgMessages)
|
|
}
|
|
|
|
func TestPushBulkInsertManyMessages(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
// Create a session with 250 messages to exercise
|
|
// multi-row VALUES batching (100 per batch).
|
|
const msgCount = 250
|
|
started := "2026-03-11T12:00:00Z"
|
|
sess := db.Session{
|
|
ID: "bulk-msg-sess",
|
|
Project: "test-project",
|
|
Machine: "local",
|
|
Agent: "claude",
|
|
StartedAt: &started,
|
|
MessageCount: msgCount,
|
|
}
|
|
require.NoError(t, local.UpsertSession(sess), "upsert session")
|
|
msgs := make([]db.Message, msgCount)
|
|
for i := range msgs {
|
|
role := "user"
|
|
if i%2 == 1 {
|
|
role = "assistant"
|
|
}
|
|
msgs[i] = db.Message{
|
|
SessionID: "bulk-msg-sess",
|
|
Ordinal: i,
|
|
Role: role,
|
|
Content: fmt.Sprintf("message %d", i),
|
|
ContentLength: len(fmt.Sprintf("message %d", i)),
|
|
}
|
|
// Add a tool call on every 10th assistant message.
|
|
if role == "assistant" && i%10 == 1 {
|
|
msgs[i].HasToolUse = true
|
|
msgs[i].ToolCalls = []db.ToolCall{{
|
|
ToolName: "Read",
|
|
Category: "Read",
|
|
ToolUseID: fmt.Sprintf("toolu_%d", i),
|
|
ResultContentLength: 10,
|
|
ResultContent: "some result",
|
|
}}
|
|
}
|
|
}
|
|
require.NoError(t, local.InsertMessages(msgs), "insert messages")
|
|
|
|
result, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
assert.Equal(t, 1, result.SessionsPushed)
|
|
assert.Equal(t, msgCount, result.MessagesPushed)
|
|
|
|
// Verify all messages landed in PG.
|
|
var pgMsgCount int
|
|
require.NoError(t, ps.pg.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM messages WHERE session_id = $1",
|
|
"bulk-msg-sess",
|
|
).Scan(&pgMsgCount), "counting pg messages")
|
|
assert.Equal(t, msgCount, pgMsgCount)
|
|
|
|
// Verify tool calls landed.
|
|
var pgTCCount int
|
|
require.NoError(t, ps.pg.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM tool_calls WHERE session_id = $1",
|
|
"bulk-msg-sess",
|
|
).Scan(&pgTCCount), "counting pg tool_calls")
|
|
// Every 10th assistant message (ordinals 1, 11, 21, ...).
|
|
expectedTC := 0
|
|
for i := range msgCount {
|
|
if i%2 == 1 && i%10 == 1 {
|
|
expectedTC++
|
|
}
|
|
}
|
|
assert.Equal(t, expectedTC, pgTCCount)
|
|
}
|
|
|
|
func TestPushSimplePK(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
ctx := context.Background()
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
var constraintDef string
|
|
err = ps.pg.QueryRowContext(ctx, `
|
|
SELECT pg_get_constraintdef(c.oid)
|
|
FROM pg_constraint c
|
|
JOIN pg_namespace n ON n.oid = c.connamespace
|
|
WHERE n.nspname = 'agentsview'
|
|
AND c.conrelid = 'agentsview.sessions'::regclass
|
|
AND c.contype = 'p'
|
|
`).Scan(&constraintDef)
|
|
require.NoError(t, err, "querying sessions PK")
|
|
assert.Equal(t, "PRIMARY KEY (id)", constraintDef)
|
|
|
|
err = ps.pg.QueryRowContext(ctx, `
|
|
SELECT pg_get_constraintdef(c.oid)
|
|
FROM pg_constraint c
|
|
JOIN pg_namespace n ON n.oid = c.connamespace
|
|
WHERE n.nspname = 'agentsview'
|
|
AND c.conrelid = 'agentsview.messages'::regclass
|
|
AND c.contype = 'p'
|
|
`).Scan(&constraintDef)
|
|
require.NoError(t, err, "querying messages PK")
|
|
assert.Equal(t, "PRIMARY KEY (session_id, ordinal)", constraintDef)
|
|
}
|
|
|
|
func TestPushFilteredByProject(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
|
|
// Seed three sessions across three projects.
|
|
for _, s := range []db.Session{
|
|
{
|
|
ID: "s-alpha", Project: "alpha",
|
|
Machine: "local", Agent: "claude",
|
|
MessageCount: 1,
|
|
},
|
|
{
|
|
ID: "s-beta", Project: "beta",
|
|
Machine: "local", Agent: "claude",
|
|
MessageCount: 1,
|
|
},
|
|
{
|
|
ID: "s-gamma", Project: "gamma",
|
|
Machine: "local", Agent: "claude",
|
|
MessageCount: 1,
|
|
},
|
|
} {
|
|
require.NoError(t, local.UpsertSession(s), "upsert %s", s.ID)
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: s.ID, Ordinal: 0,
|
|
Role: "user", Content: "msg " + s.ID,
|
|
},
|
|
}), "insert msg %s", s.ID)
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// Step 1: push with project filter = ["alpha"].
|
|
filtered, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{"alpha"}},
|
|
)
|
|
require.NoError(t, err, "creating filtered sync")
|
|
defer filtered.Close()
|
|
|
|
require.NoError(t, filtered.EnsureSchema(ctx), "ensure schema")
|
|
r1, err := filtered.Push(ctx, false, nil)
|
|
require.NoError(t, err, "filtered push")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
// Verify only alpha is in PG.
|
|
pgSessionCount := func(project string) int {
|
|
t.Helper()
|
|
var n int
|
|
err := filtered.pg.QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM sessions "+
|
|
"WHERE project = $1",
|
|
project,
|
|
).Scan(&n)
|
|
require.NoError(t, err, "count %s", project)
|
|
return n
|
|
}
|
|
assert.Equal(t, 1, pgSessionCount("alpha"))
|
|
assert.Equal(t, 0, pgSessionCount("beta"))
|
|
assert.Equal(t, 0, pgSessionCount("gamma"))
|
|
|
|
// Step 2: push unfiltered — beta and gamma should arrive.
|
|
unfiltered, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating unfiltered sync")
|
|
defer unfiltered.Close()
|
|
|
|
r2, err := unfiltered.Push(ctx, false, nil)
|
|
require.NoError(t, err, "unfiltered push")
|
|
require.GreaterOrEqual(t, r2.SessionsPushed, 2)
|
|
|
|
// Verify all three projects are in PG.
|
|
for _, p := range []string{"alpha", "beta", "gamma"} {
|
|
assert.Equal(t, 1, pgSessionCount(p), "project %s", p)
|
|
}
|
|
|
|
// Step 3: second filtered push is a no-op (fingerprints
|
|
// match).
|
|
r3, err := filtered.Push(ctx, false, nil)
|
|
require.NoError(t, err, "second filtered push")
|
|
assert.Equal(t, 0, r3.SessionsPushed)
|
|
}
|
|
|
|
func TestFilteredPushAfterResetDoesNotMaskUnfilteredResetRecovery(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
const schema = "agentsview_filtered_reset_marker"
|
|
cleanNamedPGSchema(t, pgURL, schema)
|
|
t.Cleanup(func() { cleanNamedPGSchema(t, pgURL, schema) })
|
|
|
|
local := testDB(t)
|
|
ctx := context.Background()
|
|
|
|
for _, s := range []db.Session{
|
|
{
|
|
ID: "reset-alpha", Project: "alpha",
|
|
Machine: "local", Agent: "claude",
|
|
CreatedAt: "2026-03-11T12:00:00Z",
|
|
MessageCount: 1,
|
|
},
|
|
{
|
|
ID: "reset-beta", Project: "beta",
|
|
Machine: "local", Agent: "claude",
|
|
CreatedAt: "2026-03-11T12:05:00Z",
|
|
MessageCount: 1,
|
|
},
|
|
} {
|
|
require.NoError(t, local.UpsertSession(s), "upsert %s", s.ID)
|
|
require.NoError(t, local.InsertMessages([]db.Message{{
|
|
SessionID: s.ID,
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "msg " + s.ID,
|
|
Timestamp: s.CreatedAt,
|
|
}}), "insert message %s", s.ID)
|
|
require.NoError(t, local.UpsertProjectIdentityObservation(
|
|
ctx, export.ProjectIdentityObservation{
|
|
SessionID: s.ID, Project: s.Project, Machine: s.Machine,
|
|
RootPath: "/repo/alpha", ObservedAt: time.Now().UTC(),
|
|
},
|
|
), "upsert identity %s", s.ID)
|
|
}
|
|
|
|
unfiltered, err := New(
|
|
pgURL, schema, local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating unfiltered sync")
|
|
defer unfiltered.Close()
|
|
|
|
r1, err := unfiltered.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial unfiltered push")
|
|
require.Equal(t, 2, r1.SessionsPushed)
|
|
|
|
cleanNamedPGSchema(t, pgURL, schema)
|
|
|
|
filtered, err := New(
|
|
pgURL, schema, local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{"alpha"}},
|
|
)
|
|
require.NoError(t, err, "creating filtered sync")
|
|
defer filtered.Close()
|
|
|
|
r2, err := filtered.Push(ctx, false, nil)
|
|
require.NoError(t, err, "filtered push after reset")
|
|
require.Equal(t, 1, r2.SessionsPushed)
|
|
|
|
unfilteredAfterReset, err := New(
|
|
pgURL, schema, local,
|
|
"test-machine", true,
|
|
SyncOptions{},
|
|
)
|
|
require.NoError(t, err, "creating unfiltered sync after reset")
|
|
defer unfilteredAfterReset.Close()
|
|
|
|
_, err = unfilteredAfterReset.Push(ctx, false, nil)
|
|
require.NoError(t, err, "unfiltered push after filtered reset")
|
|
|
|
var count int
|
|
err = unfilteredAfterReset.DB().QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM sessions WHERE project IN ('alpha', 'beta')",
|
|
).Scan(&count)
|
|
require.NoError(t, err, "counting restored sessions")
|
|
assert.Equal(t, 2, count)
|
|
}
|
|
|
|
func TestFilteredPartialPushDetectsResetWithEmptyWatermark(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
const schema = "agentsview_filtered_partial_reset"
|
|
cleanNamedPGSchema(t, pgURL, schema)
|
|
t.Cleanup(func() { cleanNamedPGSchema(t, pgURL, schema) })
|
|
|
|
local := testDB(t)
|
|
ctx := context.Background()
|
|
|
|
for _, s := range []db.Session{
|
|
{
|
|
ID: "partial-good", Project: "alpha",
|
|
Machine: "local", Agent: "claude",
|
|
CreatedAt: "2026-03-11T12:00:00Z",
|
|
MessageCount: 1,
|
|
},
|
|
{
|
|
ID: "partial-bad", Project: "alpha",
|
|
Machine: "local", Agent: "claude",
|
|
CreatedAt: "2026-03-11T12:05:00Z",
|
|
MessageCount: 1,
|
|
},
|
|
} {
|
|
require.NoError(t, local.UpsertSession(s), "upsert %s", s.ID)
|
|
require.NoError(t, local.InsertMessages([]db.Message{{
|
|
SessionID: s.ID,
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: "msg " + s.ID,
|
|
Timestamp: s.CreatedAt,
|
|
}}), "insert message %s", s.ID)
|
|
}
|
|
|
|
filtered, err := New(
|
|
pgURL, schema, local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{"alpha"}},
|
|
)
|
|
require.NoError(t, err, "creating filtered sync")
|
|
defer filtered.Close()
|
|
require.NoError(t, filtered.EnsureSchema(ctx), "ensure schema")
|
|
|
|
quotedSchema, err := quoteIdentifier(schema)
|
|
require.NoError(t, err, "quote schema")
|
|
_, err = filtered.DB().ExecContext(ctx, fmt.Sprintf(`
|
|
CREATE OR REPLACE FUNCTION %[1]s.fail_partial_bad()
|
|
RETURNS trigger LANGUAGE plpgsql AS $$
|
|
BEGIN
|
|
IF NEW.id = 'partial-bad' THEN
|
|
RAISE EXCEPTION 'forced partial push failure';
|
|
END IF;
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
CREATE TRIGGER fail_partial_bad
|
|
BEFORE INSERT OR UPDATE ON %[1]s.sessions
|
|
FOR EACH ROW EXECUTE FUNCTION %[1]s.fail_partial_bad();
|
|
`, quotedSchema))
|
|
require.NoError(t, err, "install partial failure trigger")
|
|
|
|
r1, err := filtered.Push(ctx, false, nil)
|
|
require.NoError(t, err, "initial partial filtered push")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
require.Equal(t, 1, r1.Errors)
|
|
var identityCount int
|
|
require.NoError(t, filtered.DB().QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM source_project_identity_observations",
|
|
).Scan(&identityCount))
|
|
assert.Zero(t, identityCount)
|
|
require.NoError(t, filtered.DB().QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM source_session_project_identity_snapshots",
|
|
).Scan(&identityCount))
|
|
assert.Zero(t, identityCount)
|
|
|
|
scopedLastPush, err := filtered.effectiveSyncState().GetSyncState("last_push_at")
|
|
require.NoError(t, err, "reading scoped watermark")
|
|
require.Empty(t, scopedLastPush)
|
|
boundaryState, err := filtered.effectiveSyncState().GetSyncState(lastPushBoundaryStateKey)
|
|
require.NoError(t, err, "reading scoped boundary state")
|
|
require.NotEmpty(t, boundaryState)
|
|
|
|
cleanNamedPGSchema(t, pgURL, schema)
|
|
|
|
filteredAfterReset, err := New(
|
|
pgURL, schema, local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{"alpha"}},
|
|
)
|
|
require.NoError(t, err, "creating filtered sync after reset")
|
|
defer filteredAfterReset.Close()
|
|
|
|
r2, err := filteredAfterReset.Push(ctx, false, nil)
|
|
require.NoError(t, err, "filtered push after reset")
|
|
require.Equal(t, 2, r2.SessionsPushed)
|
|
|
|
var count int
|
|
err = filteredAfterReset.DB().QueryRowContext(ctx,
|
|
"SELECT COUNT(*) FROM sessions WHERE project = 'alpha'",
|
|
).Scan(&count)
|
|
require.NoError(t, err, "counting restored filtered sessions")
|
|
assert.Equal(t, 2, count)
|
|
}
|
|
|
|
func TestPushExcludeProject(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
|
|
for _, s := range []db.Session{
|
|
{
|
|
ID: "s-a", Project: "alpha",
|
|
Machine: "local", Agent: "claude",
|
|
MessageCount: 1,
|
|
},
|
|
{
|
|
ID: "s-b", Project: "beta",
|
|
Machine: "local", Agent: "claude",
|
|
MessageCount: 1,
|
|
},
|
|
} {
|
|
require.NoError(t, local.UpsertSession(s), "upsert %s", s.ID)
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: s.ID, Ordinal: 0,
|
|
Role: "user", Content: "msg",
|
|
},
|
|
}), "insert msg %s", s.ID)
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{ExcludeProjects: []string{"beta"}},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
r, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push")
|
|
require.Equal(t, 1, r.SessionsPushed)
|
|
|
|
var pgProject string
|
|
err = ps.pg.QueryRowContext(ctx,
|
|
"SELECT project FROM sessions LIMIT 1",
|
|
).Scan(&pgProject)
|
|
require.NoError(t, err, "query pg")
|
|
assert.Equal(t, "alpha", pgProject)
|
|
}
|
|
|
|
func TestPushFilteredFullIsIncremental(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
cleanPGSchema(t, pgURL)
|
|
t.Cleanup(func() { cleanPGSchema(t, pgURL) })
|
|
|
|
local := testDB(t)
|
|
|
|
require.NoError(t, local.UpsertSession(db.Session{
|
|
ID: "s1", Project: "alpha",
|
|
Machine: "local", Agent: "claude",
|
|
MessageCount: 1,
|
|
}), "upsert")
|
|
require.NoError(t, local.InsertMessages([]db.Message{
|
|
{
|
|
SessionID: "s1", Ordinal: 0,
|
|
Role: "user", Content: "hello",
|
|
},
|
|
}), "insert msg")
|
|
|
|
ctx := context.Background()
|
|
ps, err := New(
|
|
pgURL, "agentsview", local,
|
|
"test-machine", true,
|
|
SyncOptions{Projects: []string{"alpha"}},
|
|
)
|
|
require.NoError(t, err, "creating sync")
|
|
defer ps.Close()
|
|
|
|
require.NoError(t, ps.EnsureSchema(ctx), "ensure schema")
|
|
|
|
// First push with --full.
|
|
r1, err := ps.Push(ctx, true, nil)
|
|
require.NoError(t, err, "first push")
|
|
require.Equal(t, 1, r1.SessionsPushed)
|
|
|
|
// Filtered --full must not advance the global watermark.
|
|
wm, err := local.GetSyncState("last_push_at")
|
|
require.NoError(t, err, "reading watermark")
|
|
assert.Empty(t, wm, "watermark after filtered --full")
|
|
|
|
scopedWM, err := ps.effectiveSyncState().GetSyncState("last_push_at")
|
|
require.NoError(t, err, "reading scoped watermark")
|
|
assert.NotEmpty(t, scopedWM, "scoped watermark after filtered --full")
|
|
|
|
// Boundary fingerprints must have been written.
|
|
bs, err := ps.effectiveSyncState().GetSyncState(lastPushBoundaryStateKey)
|
|
require.NoError(t, err, "reading boundary state")
|
|
require.NotEmpty(t, bs, "boundary state empty after filtered --full")
|
|
|
|
// Second push (not --full) should be a no-op because
|
|
// fingerprints were persisted after the filtered --full.
|
|
r2, err := ps.Push(ctx, false, nil)
|
|
require.NoError(t, err, "second push")
|
|
assert.Equal(t, 0, r2.SessionsPushed)
|
|
}
|