till/managarten
1
0
Fork 0
mirror of https://github.com/Memo-2023/mana-monorepo.git synced 2026-05-26 06:24:39 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

docs(plans): mark llm-fallback-aliases SHIPPED, add M-by-M commit table

Browse source 
All 5 milestones landed today in one continuous session: registry,
health cache, fallback router, observability, and consumer migration.
115 service-side tests, validator covers 2538 files.
This commit is contained in:
Till JS 2026-04-26 21:27:57 +02:00
parent 30eb7ef72d
commit 7766ea5021
27 changed files with 662 additions and 346 deletions
Download patch file Download diff file Expand all files Collapse all files

12
services/mana-ai/src/db/agents-projection.test.ts
View file

@ -6,7 +6,7 @@ interface ChangeRow {
record_id: string;
op: string;
data: Record<string, unknown> | null;
field_timestamps: Record<string, string> | null;
field_meta: Record<string, string> | null;
created_at: Date;
}
@ -16,7 +16,7 @@ function row(overrides: Record<string, unknown>): ChangeRow {
record_id: 'agent-1',
op: 'insert',
data: null,
field_timestamps: null,
field_meta: null,
created_at: new Date('2026-04-15T00:00:00Z'),
...overrides,
} as ChangeRow;
@ -48,7 +48,7 @@ describe('mergeRaw (agents)', () => {
row({
op: 'insert',
data: { name: 'A', role: 'old role', state: 'active' },
field_timestamps: {
field_meta: {
name: '2026-04-15T00:00:00Z',
role: '2026-04-15T00:00:00Z',
state: '2026-04-15T00:00:00Z',
@ -58,7 +58,7 @@ describe('mergeRaw (agents)', () => {
row({
op: 'update',
data: { role: 'new role' },
field_timestamps: { role: '2026-04-15T12:00:00Z' },
field_meta: { role: '2026-04-15T12:00:00Z' },
created_at: new Date('2026-04-15T12:00:00Z'),
}),
]);
@ -71,12 +71,12 @@ describe('mergeRaw (agents)', () => {
row({
op: 'insert',
data: { name: 'A' },
field_timestamps: { name: '2026-04-15T12:00:00Z' },
field_meta: { name: '2026-04-15T12:00:00Z' },
}),
row({
op: 'update',
data: { name: 'B' },
field_timestamps: { name: '2026-04-15T00:00:00Z' },
field_meta: { name: '2026-04-15T00:00:00Z' },
created_at: new Date('2026-04-14T00:00:00Z'),
}),
]);

21
services/mana-ai/src/db/agents-projection.ts
View file

@ -54,7 +54,7 @@ interface ChangeRow {
record_id: string;
op: string;
data: Record<string, unknown> | null;
field_timestamps: Record<string, string> | null;
field_meta: Record<string, string> | null;
created_at: Date;
}
@ -115,7 +115,7 @@ async function refreshOne(
userId,
async (tx) =>
tx<ChangeRow[]>`
SELECT user_id, record_id, op, data, field_timestamps, created_at
SELECT user_id, record_id, op, data, field_meta, created_at
FROM sync_changes
WHERE app_id = 'ai'
AND table_name = 'agents'
@ -141,8 +141,7 @@ async function refreshOne(
record_id: agentId,
op: 'insert',
data: seed.record,
field_timestamps:
(seed.record.__fieldTimestamps as Record<string, string> | undefined) ?? null,
field_meta: (seed.record.__fieldMeta as Record<string, string> | undefined) ?? null,
created_at: seed.last_applied_at,
},
]
@ -177,29 +176,29 @@ async function refreshOne(
* filters to apply here. Exported for unit tests only. */
export function mergeRaw(rows: readonly ChangeRow[]): Record<string, unknown> | null {
let record: Record<string, unknown> | null = null;
let ft: Record<string, string> = {};
let fm: Record<string, string> = {};
for (const row of rows) {
if (row.op === 'delete') return null;
if (!record) {
record = row.data ? { id: row.record_id, ...row.data } : { id: row.record_id };
ft = { ...(row.field_timestamps ?? {}) };
fm = { ...(row.field_meta ?? {}) };
continue;
}
if (!row.data) continue;
const rowFT = row.field_timestamps ?? {};
const rowFM = row.field_meta ?? {};
for (const [k, v] of Object.entries(row.data)) {
const serverTime = rowFT[k] ?? row.created_at.toISOString();
const localTime = ft[k] ?? '';
const serverTime = rowFM[k] ?? row.created_at.toISOString();
const localTime = fm[k] ?? '';
if (serverTime >= localTime) {
record[k] = v;
ft[k] = serverTime;
fm[k] = serverTime;
}
}
}
if (record && (record.deletedAt as string | undefined)) return null;
if (record) record.__fieldTimestamps = ft;
if (record) record.__fieldMeta = fm;
return record;
}

24
services/mana-ai/src/db/iteration-writer.ts
View file

@ -88,17 +88,15 @@ function buildActor(input: AppendIterationInput): Actor {
export async function appendServerIteration(sql: Sql, input: AppendIterationInput): Promise<void> {
const { userId, missionId, allIterations, nowIso } = input;
const fieldsPayload = {
iterations: { value: allIterations, updatedAt: nowIso },
updatedAt: { value: nowIso, updatedAt: nowIso },
};
const fieldTimestamps = {
const fieldMeta = {
iterations: nowIso,
updatedAt: nowIso,
};
// The mana-sync Go handler stores `data` on inserts and `fields` on
// updates — for our update we populate the `data` JSONB with the
// winning values and `field_timestamps` with the per-field stamps.
// winning values and `field_meta` with the per-field stamps. Per-row
// origin is `'agent'` — every server-side iteration write is an agent
// write from the point of view of the originating "client".
const data = {
iterations: allIterations,
updatedAt: nowIso,
@ -109,25 +107,19 @@ export async function appendServerIteration(sql: Sql, input: AppendIterationInpu
// inferred type. Cast at the boundary — the JSON serialization still
// happens correctly at runtime.
const dataJson = data as unknown;
const ftJson = fieldTimestamps as unknown;
const fmJson = fieldMeta as unknown;
const actorJson = buildActor(input) as unknown;
await withUser(sql, userId, async (tx) => {
await tx`
INSERT INTO sync_changes
(app_id, table_name, record_id, user_id, op, data, field_timestamps, client_id, schema_version, actor)
(app_id, table_name, record_id, user_id, op, data, field_meta, client_id, schema_version, actor, origin)
VALUES
('ai', 'aiMissions', ${missionId}, ${userId}, 'update',
${tx.json(dataJson as never)}, ${tx.json(ftJson as never)},
'mana-ai-runner', 1, ${tx.json(actorJson as never)})
${tx.json(dataJson as never)}, ${tx.json(fmJson as never)},
'mana-ai-runner', 1, ${tx.json(actorJson as never)}, 'agent')
`;
});
// fieldsPayload is kept as a named local so a future refactor that
// needs to emit a `fields`-shaped payload (if mana-sync ever rejects
// `data` for updates) has a ready-made map to send. Current contract
// accepts either.
void fieldsPayload;
}
/** Convert an {@link AiPlanOutput} from the shared parser into the

6
services/mana-ai/src/db/missions-projection.test.ts
View file

@ -8,7 +8,7 @@ function row(overrides: Record<string, unknown>) {
user_id: 'u-1',
op: 'insert',
data: null,
field_timestamps: null,
field_meta: null,
created_at: new Date('2026-04-15T00:00:00Z'),
...overrides,
} as Parameters<typeof mergeAndFilter>[0][number];
@ -99,7 +99,7 @@ describe('mergeAndFilter', () => {
iterations: [],
nextRunAt: '2026-04-15T00:00:00Z',
},
field_timestamps: {
field_meta: {
state: '2026-04-15T00:00:00Z',
title: '2026-04-15T00:00:00Z',
nextRunAt: '2026-04-15T00:00:00Z',
@ -108,7 +108,7 @@ describe('mergeAndFilter', () => {
row({
created_at: new Date('2026-04-15T01:00:00Z'),
data: { title: 'new' },
field_timestamps: { title: '2026-04-15T01:00:00Z' },
field_meta: { title: '2026-04-15T01:00:00Z' },
}),
];
const out = mergeAndFilter(rows, 'u-1', NOW);

14
services/mana-ai/src/db/missions-projection.ts
View file

@ -44,7 +44,7 @@ interface ChangeRow {
user_id: string;
op: string;
data: Record<string, unknown> | null;
field_timestamps: Record<string, string> | null;
field_meta: Record<string, string> | null;
created_at: Date;
}
@ -117,19 +117,19 @@ export function mergeAndFilter(
continue;
}
const prevFT = (existing.__fieldTimestamps as Record<string, string> | undefined) ?? {};
const nextFT = { ...prevFT };
const prevFM = (existing.__fieldMeta as Record<string, string> | undefined) ?? {};
const nextFM = { ...prevFM };
if (row.data) {
for (const [k, v] of Object.entries(row.data)) {
const serverTime = row.field_timestamps?.[k] ?? row.created_at.toISOString();
const localTime = prevFT[k] ?? '';
const serverTime = row.field_meta?.[k] ?? row.created_at.toISOString();
const localTime = prevFM[k] ?? '';
if (serverTime >= localTime) {
existing[k] = v;
nextFT[k] = serverTime;
nextFM[k] = serverTime;
}
}
}
existing.__fieldTimestamps = nextFT;
existing.__fieldMeta = nextFM;
}
const missions: ServerMission[] = [];

4
services/mana-ai/src/db/resolvers/encrypted.test.ts
View file

@ -118,7 +118,7 @@ describe('encrypted resolver', () => {
{
op: 'insert',
data: { title: encTitle, content: encContent, createdAt: '2026-04-15' },
field_timestamps: null,
field_meta: null,
created_at: new Date(0),
},
],
@ -157,7 +157,7 @@ describe('encrypted resolver', () => {
{
op: 'insert',
data: { title: enc },
field_timestamps: null,
field_meta: null,
created_at: new Date(0),
},
],

18
services/mana-ai/src/db/resolvers/record-replay.ts
View file

@ -16,7 +16,7 @@ import { withUser } from '../connection';
interface ChangeRow {
op: string;
data: Record<string, unknown> | null;
field_timestamps: Record<string, string> | null;
field_meta: Record<string, string> | null;
created_at: Date;
}
@ -29,7 +29,7 @@ export async function replayRecord(
): Promise<Record<string, unknown> | null> {
return withUser(sql, userId, async (tx) => {
const rows = await tx<ChangeRow[]>`
SELECT op, data, field_timestamps, created_at
SELECT op, data, field_meta, created_at
FROM sync_changes
WHERE user_id = ${userId}
AND app_id = ${appId}
@ -40,7 +40,7 @@ export async function replayRecord(
if (rows.length === 0) return null;
let record: Record<string, unknown> | null = null;
let fieldTimestamps: Record<string, string> = {};
let fieldMeta: Record<string, string> = {};
for (const row of rows) {
if (row.op === 'delete') {
@ -49,20 +49,20 @@ export async function replayRecord(
if (!record) {
record = row.data ? { id: recordId, ...row.data } : { id: recordId };
if (row.field_timestamps) {
fieldTimestamps = { ...row.field_timestamps };
if (row.field_meta) {
fieldMeta = { ...row.field_meta };
}
continue;
}
if (!row.data) continue;
const rowFT = row.field_timestamps ?? {};
const rowFM = row.field_meta ?? {};
for (const [k, v] of Object.entries(row.data)) {
const serverTime = rowFT[k] ?? row.created_at.toISOString();
const localTime = fieldTimestamps[k] ?? '';
const serverTime = rowFM[k] ?? row.created_at.toISOString();
const localTime = fieldMeta[k] ?? '';
if (serverTime >= localTime) {
record[k] = v;
fieldTimestamps[k] = serverTime;
fieldMeta[k] = serverTime;
}
}
}

21
services/mana-ai/src/db/snapshot-refresh.ts
View file

@ -29,7 +29,7 @@ interface ChangeRow {
record_id: string;
op: string;
data: Record<string, unknown> | null;
field_timestamps: Record<string, string> | null;
field_meta: Record<string, string> | null;
created_at: Date;
}
@ -96,7 +96,7 @@ async function refreshOne(
userId,
async (tx) =>
tx<ChangeRow[]>`
SELECT user_id, record_id, op, data, field_timestamps, created_at
SELECT user_id, record_id, op, data, field_meta, created_at
FROM sync_changes
WHERE app_id = 'ai'
AND table_name = 'aiMissions'
@ -124,8 +124,7 @@ async function refreshOne(
record_id: missionId,
op: 'insert',
data: seed.record,
field_timestamps:
(seed.record.__fieldTimestamps as Record<string, string> | undefined) ?? null,
field_meta: (seed.record.__fieldMeta as Record<string, string> | undefined) ?? null,
created_at: seed.last_applied_at,
},
]
@ -167,26 +166,26 @@ async function refreshOne(
*/
function mergeRaw(rows: readonly ChangeRow[]): Record<string, unknown> | null {
let record: Record<string, unknown> | null = null;
let ft: Record<string, string> = {};
let fm: Record<string, string> = {};
for (const row of rows) {
if (row.op === 'delete') return null;
if (!record) {
record = row.data ? { id: row.record_id, ...row.data } : { id: row.record_id };
ft = { ...(row.field_timestamps ?? {}) };
fm = { ...(row.field_meta ?? {}) };
continue;
}
if (!row.data) continue;
const rowFT = row.field_timestamps ?? {};
const rowFM = row.field_meta ?? {};
for (const [k, v] of Object.entries(row.data)) {
const serverTime = rowFT[k] ?? row.created_at.toISOString();
const localTime = ft[k] ?? '';
const serverTime = rowFM[k] ?? row.created_at.toISOString();
const localTime = fm[k] ?? '';
if (serverTime >= localTime) {
record[k] = v;
ft[k] = serverTime;
fm[k] = serverTime;
}
}
}
if (record) record.__fieldTimestamps = ft;
if (record) record.__fieldMeta = fm;
return record;
}

16
services/mana-sync/CLAUDE.md
View file

@ -57,9 +57,11 @@ CLIENT -> SERVER:
"id": "todo-123",
"op": "update",
"fields": {
"title": { "value": "Buy milk", "updatedAt": "2024-01-01T10:05:00Z" },
"completed": { "value": true, "updatedAt": "2024-01-01T10:06:00Z" }
}
"title": { "value": "Buy milk", "at": "2024-01-01T10:05:00Z" },
"completed": { "value": true, "at": "2024-01-01T10:06:00Z" }
},
"actor": { "kind": "user", "principalId": "user-1", "displayName": "Du" },
"origin": "user"
}
]
}
@ -159,16 +161,20 @@ sync_changes (
user_id TEXT NOT NULL,
op TEXT NOT NULL CHECK (insert | update | delete),
data JSONB,
field_timestamps JSONB DEFAULT '{}',
field_meta JSONB DEFAULT '{}',
client_id TEXT NOT NULL,
created_at TIMESTAMPTZ DEFAULT now(),
schema_version INT NOT NULL DEFAULT 1,
actor JSONB -- AI Workbench attribution: { kind: user|ai|system, ... }
actor JSONB, -- AI Workbench attribution: { kind: user|ai|system, ... }
origin TEXT, -- pipeline: user | agent | system | migration
space_id TEXT
)
```
**`actor` column (2026-04-14)**: Opaque JSON blob the webapp stamps on every change to distinguish user writes from autonomous AI writes and derived subsystem writes. Server does NOT parse the shape — just persists + re-emits. Pre-actor clients omit the field; the column is nullable. See `apps/mana/apps/web/src/lib/data/events/actor.ts` for the discriminated union + `COMPANION_BRAIN_ARCHITECTURE.md §20` for the full pipeline.
**`origin` column + `field_meta` rename (2026-04-26, F1 of `docs/plans/sync-field-meta-overhaul.md`)**: `field_timestamps` was renamed to `field_meta` for symmetry with the client-side `__fieldMeta` and to reserve room for richer per-field metadata. The new `origin` column carries the pipeline that produced the write on the originating client (`user` / `agent` / `system` / `migration`) — drives client-side conflict-detection: only `'user'`-origin writes can lose to a server overwrite and surface a conflict toast (F2). FieldChange wire shape changed from `{ value, updatedAt }` to `{ value, at }` to match.
Indexes: `(user_id, app_id, created_at)`, `(table_name, record_id, created_at)`, `(user_id, app_id, table_name, created_at)`
## Configuration

120
services/mana-sync/internal/store/postgres.go
View file

@ -51,7 +51,13 @@ func (s *Store) Migrate(ctx context.Context) error {
user_id TEXT NOT NULL,
op TEXT NOT NULL CHECK (op IN ('insert', 'update', 'delete')),
data JSONB,
field_timestamps JSONB DEFAULT '{}',
-- field_meta: per-field write timestamps as { [field]: ISO }.
-- Replaces the older field_timestamps column with the same
-- semantics; kept JSONB for forward compatibility if we ever
-- need to attach per-field actor or origin separately again.
-- Renamed alongside the docs/plans/sync-field-meta-overhaul.md
-- rollout (F1) — see plan for the full architecture.
field_meta JSONB DEFAULT '{}',
client_id TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-- M2: schema_version lets us evolve the Change wire shape over time.
@ -62,7 +68,12 @@ func (s *Store) Migrate(ctx context.Context) error {
-- AI Workbench: opaque actor JSON (user / ai / system). Null for
-- pre-actor clients; the webapp stamps every change with it from
-- the Dexie hook onward. Server-side we just persist and re-emit.
actor JSONB
actor JSONB,
-- Pipeline origin of the write on the originating client:
-- 'user', 'agent', 'system', or 'migration'. Receiving clients
-- treat every server-pulled change as 'server-replay' locally
-- regardless. See packages/shared-ai/src/field-meta.ts.
origin TEXT
);
-- Idempotent add for databases created before M2 shipped.
@ -73,6 +84,10 @@ func (s *Store) Migrate(ctx context.Context) error {
ALTER TABLE sync_changes
ADD COLUMN IF NOT EXISTS actor JSONB;
-- Idempotent add for databases created before the field-meta overhaul.
ALTER TABLE sync_changes
ADD COLUMN IF NOT EXISTS origin TEXT;
-- Idempotent add for databases created before the Spaces foundation.
-- Nullable so pre-v28 clients (which don't stamp a spaceId) can
-- keep pushing. The RLS policy is intentionally NOT space-aware
@ -215,7 +230,11 @@ func (s *Store) withUserAndMemberships(
// `actor` is the opaque JSON blob the webapp stamps on every change (see
// `data/events/actor.ts`). Pass nil for pre-actor callers; the column is
// nullable and cross-device consumers treat a missing actor as `user`.
func (s *Store) RecordChange(ctx context.Context, appID, tableName, recordID, userID, spaceID, op, clientID string, data map[string]any, fieldTimestamps map[string]string, schemaVersion int, actor json.RawMessage) error {
//
// `origin` describes the pipeline that produced the write on the originating
// client (`user` / `agent` / `system` / `migration`). Empty for pre-origin
// callers; the column is nullable.
func (s *Store) RecordChange(ctx context.Context, appID, tableName, recordID, userID, spaceID, op, clientID string, data map[string]any, fieldMeta map[string]string, schemaVersion int, actor json.RawMessage, origin string) error {
if schemaVersion <= 0 {
schemaVersion = 1
}
@ -225,9 +244,9 @@ func (s *Store) RecordChange(ctx context.Context, appID, tableName, recordID, us
return fmt.Errorf("marshal data: %w", err)
}
ftJSON, err := json.Marshal(fieldTimestamps)
fmJSON, err := json.Marshal(fieldMeta)
if err != nil {
return fmt.Errorf("marshal field_timestamps: %w", err)
return fmt.Errorf("marshal field_meta: %w", err)
}
// pgx serializes a nil []byte as NULL for JSONB columns, which is what
@ -246,12 +265,18 @@ func (s *Store) RecordChange(ctx context.Context, appID, tableName, recordID, us
spaceIDParam = &spaceID
}
// Same nullable handling for origin: empty string lands as SQL NULL.
var originParam *string
if origin != "" {
originParam = &origin
}
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
query := `
INSERT INTO sync_changes (app_id, table_name, record_id, user_id, space_id, op, data, field_timestamps, client_id, schema_version, actor)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
INSERT INTO sync_changes (app_id, table_name, record_id, user_id, space_id, op, data, field_meta, client_id, schema_version, actor, origin)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
`
_, err := tx.Exec(ctx, query, appID, tableName, recordID, userID, spaceIDParam, op, dataJSON, ftJSON, clientID, schemaVersion, actorJSON)
_, err := tx.Exec(ctx, query, appID, tableName, recordID, userID, spaceIDParam, op, dataJSON, fmJSON, clientID, schemaVersion, actorJSON, originParam)
return err
})
}
@ -270,7 +295,7 @@ func (s *Store) GetChangesSince(ctx context.Context, userID, appID, tableName, s
var changes []ChangeRow
err = s.withUserAndMemberships(ctx, userID, spaceIDs, func(tx pgx.Tx) error {
query := `
SELECT id, table_name, record_id, op, data, field_timestamps, client_id, created_at, schema_version, space_id, actor
SELECT id, table_name, record_id, op, data, field_meta, client_id, created_at, schema_version, space_id, actor, origin
FROM sync_changes
WHERE (user_id = $1 OR space_id = ANY($7)) AND app_id = $2 AND table_name = $3
AND created_at > $4 AND client_id != $5
@ -285,24 +310,27 @@ func (s *Store) GetChangesSince(ctx context.Context, userID, appID, tableName, s
for rows.Next() {
var c ChangeRow
var dataJSON, ftJSON, actorJSON []byte
var spaceID *string
var dataJSON, fmJSON, actorJSON []byte
var spaceID, origin *string
if err := rows.Scan(&c.ID, &c.TableName, &c.RecordID, &c.Op, &dataJSON, &ftJSON, &c.ClientID, &c.CreatedAt, &c.SchemaVersion, &spaceID, &actorJSON); err != nil {
if err := rows.Scan(&c.ID, &c.TableName, &c.RecordID, &c.Op, &dataJSON, &fmJSON, &c.ClientID, &c.CreatedAt, &c.SchemaVersion, &spaceID, &actorJSON, &origin); err != nil {
return err
}
if spaceID != nil {
c.SpaceID = *spaceID
}
if origin != nil {
c.Origin = *origin
}
if dataJSON != nil {
if err := json.Unmarshal(dataJSON, &c.Data); err != nil {
return fmt.Errorf("unmarshal data for record %s: %w", c.RecordID, err)
}
}
if ftJSON != nil {
if err := json.Unmarshal(ftJSON, &c.FieldTimestamps); err != nil {
return fmt.Errorf("unmarshal field_timestamps for record %s: %w", c.RecordID, err)
if fmJSON != nil {
if err := json.Unmarshal(fmJSON, &c.FieldMeta); err != nil {
return fmt.Errorf("unmarshal field_meta for record %s: %w", c.RecordID, err)
}
}
if len(actorJSON) > 0 {
@ -328,7 +356,7 @@ func (s *Store) GetAllChangesSince(ctx context.Context, userID, appID, since, ex
var changes []ChangeRow
err = s.withUserAndMemberships(ctx, userID, spaceIDs, func(tx pgx.Tx) error {
query := `
SELECT id, table_name, record_id, op, data, field_timestamps, client_id, created_at, schema_version, space_id, actor
SELECT id, table_name, record_id, op, data, field_meta, client_id, created_at, schema_version, space_id, actor, origin
FROM sync_changes
WHERE (user_id = $1 OR space_id = ANY($5)) AND app_id = $2
AND created_at > $3 AND client_id != $4
@ -343,24 +371,27 @@ func (s *Store) GetAllChangesSince(ctx context.Context, userID, appID, since, ex
for rows.Next() {
var c ChangeRow
var dataJSON, ftJSON, actorJSON []byte
var spaceID *string
var dataJSON, fmJSON, actorJSON []byte
var spaceID, origin *string
if err := rows.Scan(&c.ID, &c.TableName, &c.RecordID, &c.Op, &dataJSON, &ftJSON, &c.ClientID, &c.CreatedAt, &c.SchemaVersion, &spaceID, &actorJSON); err != nil {
if err := rows.Scan(&c.ID, &c.TableName, &c.RecordID, &c.Op, &dataJSON, &fmJSON, &c.ClientID, &c.CreatedAt, &c.SchemaVersion, &spaceID, &actorJSON, &origin); err != nil {
return err
}
if spaceID != nil {
c.SpaceID = *spaceID
}
if origin != nil {
c.Origin = *origin
}
if dataJSON != nil {
if err := json.Unmarshal(dataJSON, &c.Data); err != nil {
return fmt.Errorf("unmarshal data for record %s: %w", c.RecordID, err)
}
}
if ftJSON != nil {
if err := json.Unmarshal(ftJSON, &c.FieldTimestamps); err != nil {
return fmt.Errorf("unmarshal field_timestamps for record %s: %w", c.RecordID, err)
if fmJSON != nil {
if err := json.Unmarshal(fmJSON, &c.FieldMeta); err != nil {
return fmt.Errorf("unmarshal field_meta for record %s: %w", c.RecordID, err)
}
}
if len(actorJSON) > 0 {
@ -382,7 +413,7 @@ func (s *Store) GetAllChangesSince(ctx context.Context, userID, appID, since, ex
func (s *Store) StreamAllUserChanges(ctx context.Context, userID string, fn func(ChangeRow) error) error {
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
query := `
SELECT id, app_id, table_name, record_id, op, data, field_timestamps, client_id, created_at, schema_version, space_id, actor
SELECT id, app_id, table_name, record_id, op, data, field_meta, client_id, created_at, schema_version, space_id, actor, origin
FROM sync_changes
WHERE user_id = $1
ORDER BY created_at ASC, id ASC
@ -395,22 +426,25 @@ func (s *Store) StreamAllUserChanges(ctx context.Context, userID string, fn func
for rows.Next() {
var c ChangeRow
var dataJSON, ftJSON, actorJSON []byte
var spaceID *string
if err := rows.Scan(&c.ID, &c.AppID, &c.TableName, &c.RecordID, &c.Op, &dataJSON, &ftJSON, &c.ClientID, &c.CreatedAt, &c.SchemaVersion, &spaceID, &actorJSON); err != nil {
var dataJSON, fmJSON, actorJSON []byte
var spaceID, origin *string
if err := rows.Scan(&c.ID, &c.AppID, &c.TableName, &c.RecordID, &c.Op, &dataJSON, &fmJSON, &c.ClientID, &c.CreatedAt, &c.SchemaVersion, &spaceID, &actorJSON, &origin); err != nil {
return fmt.Errorf("scan: %w", err)
}
if spaceID != nil {
c.SpaceID = *spaceID
}
if origin != nil {
c.Origin = *origin
}
if dataJSON != nil {
if err := json.Unmarshal(dataJSON, &c.Data); err != nil {
return fmt.Errorf("unmarshal data for record %s: %w", c.RecordID, err)
}
}
if ftJSON != nil {
if err := json.Unmarshal(ftJSON, &c.FieldTimestamps); err != nil {
return fmt.Errorf("unmarshal field_timestamps for record %s: %w", c.RecordID, err)
if fmJSON != nil {
if err := json.Unmarshal(fmJSON, &c.FieldMeta); err != nil {
return fmt.Errorf("unmarshal field_meta for record %s: %w", c.RecordID, err)
}
}
if len(actorJSON) > 0 {
@ -426,16 +460,20 @@ func (s *Store) StreamAllUserChanges(ctx context.Context, userID string, fn func
// ChangeRow is a row from the sync_changes table.
type ChangeRow struct {
AppID string
ID string
TableName string
RecordID string
Op string
Data map[string]any
FieldTimestamps map[string]string
ClientID string
CreatedAt time.Time
SchemaVersion int
AppID string
ID string
TableName string
RecordID string
Op string
Data map[string]any
// FieldMeta carries per-field write timestamps as { [field]: ISO }.
// Per-field actor + origin live at the row level (Actor + Origin
// below) — each push represents one (actor, origin) tuple, so
// duplicating them per-field would be redundant on the wire.
FieldMeta map[string]string
ClientID string
CreatedAt time.Time
SchemaVersion int
// SpaceID is empty for pre-v28 rows. Consumers use it to partition
// the reader cache per space; an empty string means "implicit personal"
// until the bootstrap reconciliation fills it in.
@ -443,4 +481,8 @@ type ChangeRow struct {
// Actor is nil for rows written by pre-actor clients. Consumers on
// other devices render a missing actor as "user".
Actor json.RawMessage
// Origin describes the pipeline that produced the write on the
// originating client. Empty for pre-origin clients; consumers treat
// missing as "user". See packages/shared-ai/src/field-meta.ts.
Origin string
}

19
services/mana-sync/internal/sync/handler.go
View file

@ -90,18 +90,19 @@ func changeFromRow(row store.ChangeRow) Change {
Op: row.Op,
SpaceID: row.SpaceID,
Actor: row.Actor,
Origin: row.Origin,
}
switch row.Op {
case "insert":
c.Data = row.Data
case "update":
c.Fields = make(map[string]*FieldChange)
for field, ts := range row.FieldTimestamps {
for field, ts := range row.FieldMeta {
value, ok := row.Data[field]
if !ok {
continue
}
c.Fields[field] = &FieldChange{Value: value, UpdatedAt: ts}
c.Fields[field] = &FieldChange{Value: value, At: ts}
}
case "delete":
if deletedAt, ok := row.Data["deletedAt"].(string); ok {
@ -181,15 +182,15 @@ func (h *Handler) HandleSync(w http.ResponseWriter, r *http.Request) {
for _, change := range changeset.Changes {
affectedTables[change.Table] = struct{}{}
// Build data and field timestamps
// Build data and field metadata.
data := change.Data
fieldTimestamps := make(map[string]string)
fieldMeta := make(map[string]string)
if change.Op == "update" && change.Fields != nil {
data = make(map[string]any)
for field, fc := range change.Fields {
data[field] = fc.Value
fieldTimestamps[field] = fc.UpdatedAt
fieldMeta[field] = fc.At
}
}
@ -215,7 +216,7 @@ func (h *Handler) HandleSync(w http.ResponseWriter, r *http.Request) {
// clients still get indexed correctly. Empty string lands as SQL
// NULL via RecordChange.
spaceID := extractSpaceID(change)
err := h.store.RecordChange(ctx, appID, change.Table, change.ID, userID, spaceID, change.Op, clientID, data, fieldTimestamps, rowSchemaVersion, change.Actor)
err := h.store.RecordChange(ctx, appID, change.Table, change.ID, userID, spaceID, change.Op, clientID, data, fieldMeta, rowSchemaVersion, change.Actor, change.Origin)
if err != nil {
slog.Error("failed to record change", "error", err, "table", change.Table, "id", change.ID)
http.Error(w, "failed to record change: "+err.Error(), http.StatusInternalServerError)
@ -459,15 +460,15 @@ func (h *Handler) HandleStream(w http.ResponseWriter, r *http.Request) {
func (h *Handler) convertChanges(rows []store.ChangeRow) []Change {
changes := make([]Change, 0, len(rows))
for _, row := range rows {
c := Change{Table: row.TableName, ID: row.RecordID, Op: row.Op, Actor: row.Actor}
c := Change{Table: row.TableName, ID: row.RecordID, Op: row.Op, Actor: row.Actor, Origin: row.Origin}
switch row.Op {
case "insert":
c.Data = row.Data
case "update":
c.Fields = make(map[string]*FieldChange)
for field, ts := range row.FieldTimestamps {
for field, ts := range row.FieldMeta {
if value, ok := row.Data[field]; ok {
c.Fields[field] = &FieldChange{Value: value, UpdatedAt: ts}
c.Fields[field] = &FieldChange{Value: value, At: ts}
}
}
case "delete":

14
services/mana-sync/internal/sync/handler_test.go
View file

@ -19,13 +19,13 @@ type mockStore struct {
type recordedChange struct {
appID, table, recordID, userID, op, clientID string
data map[string]any
fieldTimestamps map[string]string
fieldMeta map[string]string
}
type mockChangeRow struct {
ID, TableName, RecordID, Op, ClientID string
Data map[string]any
FieldTimestamps map[string]string
FieldMeta map[string]string
}
// mockValidator always returns a fixed user ID.
@ -102,7 +102,7 @@ func TestChangesetValidation(t *testing.T) {
Since: "2024-01-01T00:00:00Z",
Changes: []Change{
{Table: "todos", ID: "todo-1", Op: "update", Fields: map[string]*FieldChange{
"title": {Value: "Updated", UpdatedAt: "2024-01-01T10:00:00Z"},
"title": {Value: "Updated", At: "2024-01-01T10:00:00Z"},
}},
},
},
@ -260,8 +260,8 @@ func TestFieldChangeRoundTrip(t *testing.T) {
ID: "todo-1",
Op: "update",
Fields: map[string]*FieldChange{
"title": {Value: "Buy milk", UpdatedAt: "2024-01-01T10:05:00Z"},
"completed": {Value: true, UpdatedAt: "2024-01-01T10:06:00Z"},
"title": {Value: "Buy milk", At: "2024-01-01T10:05:00Z"},
"completed": {Value: true, At: "2024-01-01T10:06:00Z"},
},
}
@ -286,8 +286,8 @@ func TestFieldChangeRoundTrip(t *testing.T) {
if titleField.Value != "Buy milk" {
t.Errorf("title value = %v, want 'Buy milk'", titleField.Value)
}
if titleField.UpdatedAt != "2024-01-01T10:05:00Z" {
t.Errorf("title updatedAt = %q, want '2024-01-01T10:05:00Z'", titleField.UpdatedAt)
if titleField.At != "2024-01-01T10:05:00Z" {
t.Errorf("title at = %q, want '2024-01-01T10:05:00Z'", titleField.At)
}
completedField := decoded.Fields["completed"]

40
services/mana-sync/internal/sync/types.go
View file

@ -47,12 +47,23 @@ type Change struct {
// clients so cross-device attribution works. Pre-actor clients omit
// the field; the column is nullable.
Actor json.RawMessage `json:"actor,omitempty"`
// Origin describes the pipeline that produced the write on the
// originating client: 'user', 'agent', 'system', or 'migration'.
// The server stores it verbatim and re-emits to other clients —
// receiving clients then apply with origin='server-replay' locally
// regardless. Pre-origin clients omit the field; the column is
// nullable. See packages/shared-ai/src/field-meta.ts for the
// full enumeration and conflict-detection semantics.
Origin string `json:"origin,omitempty"`
}
// FieldChange holds a value and the timestamp when it was last changed.
// FieldChange holds a value and the timestamp when the field was last
// written. Per-field actor + origin are not transmitted at the field
// level — they live at the row level on Change.Actor + Change.Origin
// because each push represents one (actor, origin) tuple.
type FieldChange struct {
Value any `json:"value"`
UpdatedAt string `json:"updatedAt"`
Value any `json:"value"`
At string `json:"at"`
}
// Changeset is a batch of changes sent by a client.
@ -95,15 +106,16 @@ type PullRequest struct {
// SyncRecord is a row in the sync_changes table.
type SyncRecord struct {
ID string `json:"id"`
AppID string `json:"appId"`
TableName string `json:"tableName"`
RecordID string `json:"recordId"`
UserID string `json:"userId"`
Op string `json:"op"`
Fields map[string]any `json:"fields,omitempty"`
Data map[string]any `json:"data,omitempty"`
FieldTimestamps map[string]string `json:"fieldTimestamps,omitempty"`
ClientID string `json:"clientId"`
CreatedAt time.Time `json:"createdAt"`
ID string `json:"id"`
AppID string `json:"appId"`
TableName string `json:"tableName"`
RecordID string `json:"recordId"`
UserID string `json:"userId"`
Op string `json:"op"`
Fields map[string]any `json:"fields,omitempty"`
Data map[string]any `json:"data,omitempty"`
FieldMeta map[string]string `json:"fieldMeta,omitempty"`
Origin string `json:"origin,omitempty"`
ClientID string `json:"clientId"`
CreatedAt time.Time `json:"createdAt"`
}