From e893b8a6e13a1d877ec1b411ba9fcf53e528fbb2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 20:35:43 +0000 Subject: [PATCH 1/5] wip: notifications take the organization of what they are about Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- .../src/tenancy/platform-object-tenancy.ts | 22 +++ .../plugin-approvals/src/approval-service.ts | 36 +++- .../src/builtin/notify-node.ts | 173 +++++++++++++++--- .../src/messaging-service.ts | 41 ++++- 4 files changed, 233 insertions(+), 39 deletions(-) diff --git a/packages/objectql/src/tenancy/platform-object-tenancy.ts b/packages/objectql/src/tenancy/platform-object-tenancy.ts index c63d1b954f5..3965872d8f6 100644 --- a/packages/objectql/src/tenancy/platform-object-tenancy.ts +++ b/packages/objectql/src/tenancy/platform-object-tenancy.ts @@ -180,6 +180,28 @@ export const PLATFORM_OBJECT_TENANCY: Readonly; } @@ -1614,11 +1621,24 @@ export class ApprovalService implements IApprovalService { topic: string; audience: string[]; payload: ApprovalNotificationPayload; + /** + * [#23284] The organization of the request the notification is about — + * the request row's own `organization_id` (`requestOrg` while the request + * opens). A REQUIRED key, `null` included, so no call site can send an + * approval notification without having said whose it is: before it + * existed every approval notification reached `sys_notification` with + * `organization_id` NULL, a user-driven `approval.requested` included, + * because nothing here passed one. ⛔ Never the actor's organization: an + * approver acting from another organization does not move the request. + */ + organizationId: string | null | undefined; dedupKey?: string; source?: { object: string; id: string }; actorContext?: ExecutionContext; }): Promise { - const { actorContext, ...event } = input; + const { actorContext, organizationId, ...event } = input; + const organization = + organizationId != null && String(organizationId) !== '' ? String(organizationId) : undefined; const audience = event.audience.filter(a => a && !a.includes(':')); if (!this.messaging || !audience.length) return 0; const person = actorContext ? recordedActor(actorContext) : null; @@ -1637,6 +1657,7 @@ export class ApprovalService implements IApprovalService { try { await this.messaging.emit({ severity: 'info', ...event, payload, audience, ...(person ? { actorId: person } : {}), + ...(organization ? { organizationId: organization } : {}), }); return audience.length; } catch (err: any) { @@ -3495,6 +3516,7 @@ export class ApprovalService implements IApprovalService { }, { context: SYSTEM_CTX }); await this.notify({ topic: 'approval.ooo_substituted', + organizationId: requestOrg, audience: [sub.to], source: { object: 'sys_approval_request', id }, dedupKey: `approval-ooo-${id}-${sub.to}`, @@ -3506,6 +3528,7 @@ export class ApprovalService implements IApprovalService { }); await this.notify({ topic: 'approval.ooo_skipped', + organizationId: requestOrg, audience: [sub.from], source: { object: 'sys_approval_request', id }, dedupKey: `approval-ooo-skip-${id}-${sub.from}`, @@ -3565,6 +3588,7 @@ export class ApprovalService implements IApprovalService { for (const approver of openedOn) { await this.notify({ topic: 'approval.requested', + organizationId: requestOrg, audience: [approver], actorContext: context, source: { object: 'sys_approval_request', id }, @@ -4672,6 +4696,7 @@ export class ApprovalService implements IApprovalService { if (raw.submitter_id) { await this.notify({ topic: 'approval.returned', + organizationId: raw.organization_id, audience: [String(raw.submitter_id)], actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -4715,6 +4740,7 @@ export class ApprovalService implements IApprovalService { if (raw.submitter_id) { await this.notify({ topic: 'approval.returned', + organizationId: raw.organization_id, audience: [String(raw.submitter_id)], actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -4974,6 +5000,7 @@ export class ApprovalService implements IApprovalService { await this.notify({ topic: 'approval.reassigned', + organizationId: raw.organization_id, audience: [to], actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -5033,6 +5060,7 @@ export class ApprovalService implements IApprovalService { const tokens = await this.issueActionTokens(requestId, approver); notified += await this.notify({ topic: 'approval.reminder', + organizationId: raw.organization_id, audience: [approver], actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -5056,6 +5084,7 @@ export class ApprovalService implements IApprovalService { if (literals.length) { notified += await this.notify({ topic: 'approval.reminder', + organizationId: raw.organization_id, audience: literals, actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -5304,6 +5333,7 @@ export class ApprovalService implements IApprovalService { if (raw.submitter_id) { await this.notify({ topic: 'approval.request_info', + organizationId: raw.organization_id, audience: [String(raw.submitter_id)], actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -5351,6 +5381,7 @@ export class ApprovalService implements IApprovalService { const audience = isSubmitter ? pending : [String(raw.submitter_id ?? '')].filter(Boolean); await this.notify({ topic: 'approval.comment', + organizationId: raw.organization_id, audience, actorContext: context, source: { object: 'sys_approval_request', id: requestId }, @@ -6397,6 +6428,7 @@ export class ApprovalService implements IApprovalService { await this.syncApproverIndex(raw.id, escalatees, raw.organization_id ?? null, now); await this.notify({ topic: 'approval.escalated', + organizationId: raw.organization_id, audience: escalatees, source: { object: 'sys_approval_request', id: raw.id }, payload: { @@ -6415,6 +6447,7 @@ export class ApprovalService implements IApprovalService { // 'notify' (and the reassign-without-target fallback) await this.notify({ topic: 'approval.sla_breached', + organizationId: raw.organization_id, audience: [...pending, ...escalatees], source: { object: 'sys_approval_request', id: raw.id }, payload: { @@ -6428,6 +6461,7 @@ export class ApprovalService implements IApprovalService { if (esc.notifySubmitter !== false && raw.submitter_id) { await this.notify({ topic: 'approval.sla_breached', + organizationId: raw.organization_id, audience: [String(raw.submitter_id)], source: { object: 'sys_approval_request', id: raw.id }, payload: { diff --git a/packages/services/service-automation/src/builtin/notify-node.ts b/packages/services/service-automation/src/builtin/notify-node.ts index 35b267184af..2f4ea29ce00 100644 --- a/packages/services/service-automation/src/builtin/notify-node.ts +++ b/packages/services/service-automation/src/builtin/notify-node.ts @@ -1,7 +1,8 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import type { PluginContext } from '@objectstack/core'; -import type { AutomationContext } from '@objectstack/spec/contracts'; +import type { AutomationContext, IDataEngine } from '@objectstack/spec/contracts'; +import { createRecordWallOrganizationResolver, type RecordOrganizationResolver } from '@objectstack/metadata-core'; import { defineActionDescriptor, isExpressionEnvelopeShaped, NotifyConfigSchema } from '@objectstack/spec/automation'; import type { NotifyConfigParsed } from '@objectstack/spec/automation'; import type { AutomationEngine } from '../engine.js'; @@ -79,6 +80,89 @@ function toStr(value: unknown): string | undefined { return s.length > 0 ? s : undefined; } +/** A record a notification can be about — an object name and a row id. */ +interface RecordRef { + readonly object: string; + readonly id: string; +} + +/** + * [#23284] The elevated context of the one read below. The organization is read + * off the STORED row by id, never off a snapshot the run carries: a trigger + * door may hand a run a `record` its caller wrote, and a caller-written + * `organization_id` must not decide whose notification this is. + */ +const SUBJECT_ORGANIZATION_READ = { context: { isSystem: true } } as const; + +/** + * [#23284] The organization of the record a notification is ABOUT, for a run + * whose context carries none (a schedule tick, a record change written by a + * system context). + * + * The candidates are tried in order and the first one that answers wins: + * + * 1. the click-through record the node declares (`sourceObject` + `sourceId`) + * — what `sys_notification.source_object` / `source_id` store, and the + * anchor ADR-0131's ownership inventory attributes a stored notification + * by, so the producer and the inventory answer the same question the same + * way; + * 2. the run's trigger record (`AutomationContext.object` + `record.id`) — the + * record the run itself is about, the anchor the run's own history row is + * stamped from. + * + * The column is the object's WALL column (`createRecordWallOrganizationResolver`: + * an ADR-0066 opt-out has none, then a declared `tenancy.tenantField`, then the + * kernel's `organization_id`) — "which organization does this record belong + * to", never the stamp-only divergence the three sanctioned platform-row + * writers read. An object with no such column, an object the engine does not + * know, a row that no longer exists and a row whose column is empty each + * answer nothing, and the next candidate is asked. + * + * ⛔ No candidate is a guess: not the install's first organization, not the + * recipient's membership. When none answers, the emit carries no organization + * and the engine's system-write rule decides the owner (ADR-0131 D9) — the sole + * organization under `single`, a refusal where none is derivable. A read that + * fails propagates: it fails the step rather than quietly dropping to that rule. + */ +async function organizationOfRecordAbout( + data: IDataEngine, + resolver: RecordOrganizationResolver, + candidates: ReadonlyArray, +): Promise { + for (const candidate of candidates) { + if (!candidate) continue; + const column = resolver.organizationFieldFor(candidate.object); + if (!column) continue; + const row = await data.findOne( + candidate.object, + { where: { id: candidate.id }, fields: ['id', column] }, + SUBJECT_ORGANIZATION_READ, + ); + const organization = resolver.organizationOf(candidate.object, row); + if (organization) return organization; + } + return undefined; +} + +/** The run's trigger record as a {@link RecordRef}, when the run has one. */ +function triggerRecordOf(context: AutomationContext): RecordRef | undefined { + const object = toStr(context.object); + const id = toStr((context.record as { id?: unknown } | undefined)?.id); + return object && id ? { object, id } : undefined; +} + +/** + * The code a thrown failure carries (`err.code`, the ADR-0112 envelope's + * machine-readable half), so a step that fails on the engine's refusal names + * it — `ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED` and the rest — beside the + * message. Read for the step's TEXT only: `NodeExecutionResult.code` stays + * reserved for the codes an executor classifies itself. + */ +function thrownCodeOf(err: unknown): string | undefined { + const code = err !== null && typeof err === 'object' ? (err as { code?: unknown }).code : undefined; + return typeof code === 'string' && code.length > 0 ? code : undefined; +} + /** How a returned failure names what a value expression evaluated to — `null`, a map, a number (`42`). */ function evaluatedKindOf(value: unknown): string { if (value === null) return '`null`'; @@ -206,6 +290,24 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) return undefined; } }; + // [#23284] The data engine the record-anchor read goes through — resolved + // the way the CRUD nodes resolve theirs — and one wall-column resolver per + // engine (object schemas are static after registration, so its per-object + // memo stays true for the engine's life). + const getData = (): IDataEngine | undefined => { + try { + return ctx.getService('data') ?? ctx.getService('objectql'); + } catch { + return undefined; + } + }; + let wallResolver: { readonly data: IDataEngine; readonly resolver: RecordOrganizationResolver } | undefined; + const wallResolverFor = (data: IDataEngine): RecordOrganizationResolver => { + if (wallResolver?.data !== data) { + wallResolver = { data, resolver: createRecordWallOrganizationResolver(data) }; + } + return wallResolver.resolver; + }; engine.registerNodeExecutor({ type: 'notify', @@ -449,41 +551,45 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) // Maintainer ruling, 2026-08-24, verbatim: 「11303 // sys_inbox_message/sys_notification/sys_email 应该写 // organization_id。」 — a gap, not a design choice, and the - // PRODUCERS are the fix site. Everything below `emit()` was already - // threaded; this node was the origin that never supplied a value. + // PRODUCERS are the fix site. // // `AutomationContext.tenantId` is the acting run's organization — // the same source `audit-writers.ts` hands its own `collab.mention` // emit, so the two notification producers agree about whose - // organization a notification carries. + // organization a notification carries. A run that carries one keeps + // it exactly as before: nothing below is read for it. + // + // [#23284] A run that carries NONE — a schedule tick, a record change + // written by a system context — takes the organization of the record + // the notification is about (`organizationOfRecordAbout`): the + // declared click-through record, else the run's trigger record. That + // read happens inside the `try` below, so a failed read fails the + // step instead of being dropped. // - // ⛔ There is deliberately NO fallback limb here — not "the current + // ⛔ There is deliberately NO guessed limb — not "the current // organization", not the first organization on the install, not the // recipient's first membership. A wrong `organization_id` is worse // than a null: a null is visibly missing, while a wrong value is // silently authoritative to every report, export and cleanup script - // that filters by organization. When the run carries no - // organization, the notification carries none and says so (below). - const organizationId = toStr(context.tenantId); - if (!organizationId) { - // Fail-LOUD, not fail-guess — and deliberately not fail-CLOSED. - // Refusing here would break the two deployments that legitimately - // have no organization to thread: a `single`-posture install, and - // every stack before its first organization exists. So the - // org-less write stays permitted and becomes a VISIBLE event - // instead of a silent one. - ctx.logger.warn( - `[notify] no organization in scope for topic '${topic ?? 'notify'}' — the ` + - `sys_notification / sys_inbox_message / sys_notification_receipt / ` + - `sys_notification_delivery rows for this emit will carry organization_id = NULL ` + - `and will be invisible to any report or cleanup that filters by organization. ` + - `On a multi-organization install this means the triggering context lost its ` + - `tenant: give the flow's trigger an acting organization (AutomationContext.tenantId). ` + - `On a single-organization install this is expected and can be ignored.`, - ); - } + // that filters by organization. A notification about no record, from + // a run with no organization, reaches `emit()` with no organization, + // and its owner is the engine's to decide by ADR-0131 D9 — the sole + // organization under `single`, a loud refusal + // (`ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED`) wherever none is + // derivable, which fails this step below. The WARN that used to + // predict a NULL row here is gone with the NULL row. + let organizationId = toStr(context.tenantId); try { + if (!organizationId) { + const data = getData(); + if (data) { + organizationId = await organizationOfRecordAbout(data, wallResolverFor(data), [ + source, + triggerRecordOf(context), + ]); + } + } // ADR-0030 single ingress: hand the messaging service a topic + // audience + payload; it writes the L2 event and materializes // per channel. title/body/url ride in the payload (templates in @@ -513,10 +619,10 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) severity, source, actorId, - // [#11303] Absent (not null) when the run has no - // organization: `EmitInput.organizationId` is optional, and - // the chain below normalizes a missing value to NULL exactly - // once, in `writeEvent`. + // [#11303] Absent (not null) when neither the run nor the + // record the notification is about names an organization: + // `EmitInput.organizationId` is optional, and the engine + // decides the owner of an emit that carries none (above). ...(organizationId ? { organizationId } : {}), channels: channels.length ? channels : undefined, }); @@ -601,7 +707,14 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) : { selected: recipients.length, acted: delivered }, }; } catch (err) { - return { success: false, error: `notify failed: ${(err as Error).message}` }; + // [#23284] A refusal fails the step and names the engine's code — + // the ADR-0131 D9 refusal of a notification no organization owns + // reads `notify failed: ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED: …`. + const code = thrownCodeOf(err); + return { + success: false, + error: `notify failed: ${code ? `${code}: ` : ''}${(err as Error).message}`, + }; } }, }); diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index fd81bebe283..9c6ad4d2c70 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -1089,8 +1089,24 @@ export class MessagingService { // convergence. (Drivers that don't enforce the index fall back to the // best-effort fast-path — the catch is then simply never taken.) let notificationId: string; + // [#23284] The organization every row BELOW the event carries: the one + // the producer threaded, else the one the engine stamped on the event + // row itself. A producer that could attribute the notification to no + // organization leaves the owner to the engine's system-write rule + // (ADR-0131 D9, which now reaches `sys_notification`): the sole + // organization under `single`, a refusal where none is derivable. The + // delivery, inbox row and receipt then follow their notification, so the + // family cannot split into an owned event over org-less children — which + // the inline fan-out produced, because it stamped only what the producer + // passed. ⛔ Never a guess: the value is read off the stored event row, + // and a write that stamped nothing leaves it absent, as before. + let owned: EmitInput = input; try { - notificationId = await this.writeEvent(data, input, suppressed); + const event = await this.writeEvent(data, input, suppressed); + notificationId = event.id; + if (input.organizationId === undefined && event.organizationId !== undefined) { + owned = { ...input, organizationId: event.organizationId }; + } } catch (err) { if (input.dedupKey && data) { const winner = await this.findEventByDedupKey(data, input.dedupKey); @@ -1141,7 +1157,7 @@ export class MessagingService { // 4) Either enqueue durable deliveries (P1 outbox) or fan out inline (P0). if (this.outbox) { - const deliveries = await this.enqueueDeliveries(this.outbox, notificationId, targets, input, payload); + const deliveries = await this.enqueueDeliveries(this.outbox, notificationId, targets, owned, payload); // An accepted row here is ENQUEUED, not delivered — the dispatcher // decides that later and records it on `sys_notification_delivery`. // Reporting it as `delivered` is what let a run summary claim a @@ -1162,7 +1178,7 @@ export class MessagingService { const notification: Notification = { notificationId, - organizationId: input.organizationId, + organizationId: owned.organizationId, topic: input.topic, title: str(payload.title) ?? input.topic, body: str(payload.body) ?? '', @@ -1375,17 +1391,19 @@ export class MessagingService { } /** - * Persist the L2 event and return its id. With no data layer (minimal/test - * stacks) we warn and synthesize an id so fan-out can still be exercised. + * Persist the L2 event and return its id, with the organization the stored + * row carries (#23284) — the producer's, or the one the engine stamped on a + * system write that carried none. With no data layer (minimal/test stacks) + * we warn and synthesize an id so fan-out can still be exercised. */ private async writeEvent( data: IDataEngine | undefined, input: EmitInput, suppressed: readonly ChannelSuppression[], - ): Promise { + ): Promise<{ id: string; organizationId?: string }> { if (!data) { this.ctx.logger.warn('[messaging] no data engine registered; event not persisted'); - return `evt_${Math.random().toString(36).slice(2)}`; + return { id: `evt_${Math.random().toString(36).slice(2)}` }; } const row: Record = { topic: input.topic, @@ -1422,7 +1440,14 @@ export class MessagingService { await assertActorReferenceResolves(data, NOTIFICATION_EVENT_OBJECT, row.actor_id); const created = await data.insert(NOTIFICATION_EVENT_OBJECT, row, { context: FAN_OUT_SYSTEM_CONTEXT }); const id = Array.isArray(created) ? created[0]?.id : created?.id ?? created; - return id != null ? String(id) : `evt_${Math.random().toString(36).slice(2)}`; + const stored: unknown = Array.isArray(created) ? created[0] : created; + const organizationId = stored !== null && typeof stored === 'object' + ? str((stored as Record).organization_id) + : undefined; + return { + id: id != null ? String(id) : `evt_${Math.random().toString(36).slice(2)}`, + ...(organizationId !== undefined ? { organizationId } : {}), + }; } /** From e82f9daebf2b7c6d8098cab1362d97c74d165507 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 20:44:47 +0000 Subject: [PATCH 2/5] test: pin the organization every notification row carries Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- .../tenancy-by-object-classification.test.ts | 53 +++- ...oval-notification-body.integration.test.ts | 39 ++- ...ify-organization-stamp.integration.test.ts | 273 ++++++++++++++++-- ...nization-follows-event.integration.test.ts | 135 +++++++++ 4 files changed, 468 insertions(+), 32 deletions(-) create mode 100644 packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts diff --git a/packages/objectql/src/tenancy-by-object-classification.test.ts b/packages/objectql/src/tenancy-by-object-classification.test.ts index ed6d68fef2f..65195991ab1 100644 --- a/packages/objectql/src/tenancy-by-object-classification.test.ts +++ b/packages/objectql/src/tenancy-by-object-classification.test.ts @@ -105,13 +105,15 @@ const NO_TENANT_FIELD = { } as any; const ORG_OBJECT = { name: 'sys_organization', fields: { name: { type: 'text' } } } as any; +/** ADMITTED by #23284 — the notification family's event row. */ +const SYS_NOTIFICATION = { name: 'sys_notification', fields: { topic: { type: 'text' } } } as any; async function makeEngine(opts: { posture?: string; organizations?: string[] } = {}) { const observed: ObservedCall[] = []; const engine = new ObjectQL(); engine.registerDriver(makeDriver(observed, opts.organizations ?? [ORG_ID]), true); await engine.init(); - for (const o of [SYS_FILE, SYS_PERMISSION_SET, SYS_UNADJUDICATED, DISPATCH_ORDER, NO_TENANT_FIELD, ORG_OBJECT]) { + for (const o of [SYS_FILE, SYS_PERMISSION_SET, SYS_UNADJUDICATED, DISPATCH_ORDER, NO_TENANT_FIELD, ORG_OBJECT, SYS_NOTIFICATION]) { engine.registry.registerObject(o, PACKAGE_ID); } if (opts.posture) engine.setTenancyPostureProvider(() => opts.posture); @@ -164,12 +166,19 @@ describe('#13491 the inventory — a verdict per object, never a namespace', () // Pinned as a LIST, not a count: the blast radius of the reclassification // IS this set, and a silent arrival here is a behaviour change nobody // adjudicated. Growing it is a maintainer decision, not a refactor. + // + // `sys_notification` (#23284) arrives on the maintainer's 2026-08-24 + // ruling that names the table — 「11303 + // sys_inbox_message/sys_notification/sys_email 应该写 organization_id。」 — + // once its producers fill the column; its own describe block below pins + // what the admission moves. expect(tenantScopedPlatformObjects()).toEqual([ 'sys_approval_action', 'sys_approval_approver', 'sys_approval_request', 'sys_automation_run', 'sys_file', + 'sys_notification', 'sys_notification_delivery', 'sys_record_share', 'sys_upload_session', @@ -232,6 +241,48 @@ describe('#13491 gate 1 — an admitted platform object reaches #8844 derive/ref }); }); +describe('#23284 — a notification no producer could attribute takes the owner ADR-0131 D9 derives', () => { + // The producers thread an organization when they have one (the run's, the + // record the notification is about, the approval request's). What is left — + // a notification about no record, from a run with no organization — reaches + // the engine with none, and admission is what puts it under the system-write + // rule instead of landing it with `organization_id` NULL. + it('single posture, one organization: the org-less notification takes the sole organization', async () => { + const { engine, observed } = await makeEngine({ posture: 'single' }); + await engine.insert('sys_notification', { topic: 'sweep.breach', organization_id: null }, { context: SYSTEM_CTX } as any); + expect(lastWrite(observed, 'sys_notification')?.options?.tenantId).toBe(ORG_ID); + }); + + it('single posture, two organizations: the org-less notification is REFUSED, loudly, and never written', async () => { + const { engine, observed } = await makeEngine({ posture: 'single', organizations: [ORG_ID, 'org_second'] }); + await expect( + engine.insert('sys_notification', { topic: 'sweep.breach', organization_id: null }, { context: SYSTEM_CTX } as any), + ).rejects.toMatchObject({ code: 'ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED', status: 500 }); + expect(lastWrite(observed, 'sys_notification')).toBeUndefined(); + }); + + it.each(['isolated', 'group'] as const)( + '%s posture: the org-less notification is REFUSED, loudly, and never written', + async (posture) => { + const { engine, observed } = await makeEngine({ posture }); + await expect( + engine.insert('sys_notification', { topic: 'sweep.breach', organization_id: null }, { context: SYSTEM_CTX } as any), + ).rejects.toMatchObject({ code: 'ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED', status: 500 }); + expect(lastWrite(observed, 'sys_notification')).toBeUndefined(); + }, + ); + + it('a notification that CARRIES its organization is written as it came, on a walled posture too', async () => { + const { engine, observed } = await makeEngine({ posture: 'isolated', organizations: [ORG_ID, 'org_second'] }); + await engine.insert('sys_notification', { topic: 'sweep.breach', organization_id: 'org_second' }, { context: SYSTEM_CTX } as any); + const write = lastWrite(observed, 'sys_notification'); + expect(write?.method).toBe('create'); + // Nothing was derived over the producer's answer: the engine resolved no + // organization of its own for a row that already names one. + expect(write?.options?.tenantId).toBeUndefined(); + }); +}); + describe('#13491 gate 2 — the isSystem mute narrows by the SAME classification', () => { it('an elevated write on an admitted platform object is no longer auto-muted', async () => { const { engine, observed } = await makeEngine({ posture: 'single' }); diff --git a/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts b/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts index 6ed03380be6..4c2c225b613 100644 --- a/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts +++ b/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts @@ -36,6 +36,16 @@ * The end-to-end half — the text read back from the recipient's * `GET /api/v1/notifications` through the real messaging service — is the * dogfood pin `packages/qa/dogfood/test/approval-notification-body.dogfood.test.ts`. + * + * ## And whose notification it is (#23284) + * + * The same battery pins the organization every call site hands messaging: the + * request's own `organization_id`, read back from the stored request row. + * Before it, `notify()` passed none, so every approval notification — + * `approval.requested` from a user's own submission included — reached + * `sys_notification` with `organization_id` NULL. The deals live in two + * organizations, so a site that passed one fixed organization, or the actor's, + * would not satisfy the pin. */ import { describe, it, expect, beforeAll, afterAll } from 'vitest'; @@ -63,6 +73,15 @@ const COMMENT = 'The signed contract is attached to the record now.'; const QUESTION = 'Which cost centre should this be booked against?'; const SEND_BACK = 'Split the total by quarter, please.'; +/** + * [#23284] The two organizations the battery's deals live in. The SLA pair sits + * in the second, every other request in the first — so "the request's + * organization" cannot be satisfied by a constant. + */ +const ORG_A = 'org_battery_a'; +const ORG_B = 'org_battery_b'; +const orgOfDeal = (key: string): string => (key.startsWith('sla_') ? ORG_B : ORG_A); + /** The fields `notify()` declares for a notification's payload. */ const PAYLOAD_FIELDS = new Set(['title', 'body', 'actionUrl', 'actions']); @@ -113,6 +132,7 @@ interface Emitted { audience: string[]; source?: { object: string; id: string }; payload?: Record; + organizationId?: string; } describe('every approval notification carries its text in body', () => { @@ -123,7 +143,7 @@ describe('every approval notification carries its text in body', () => { const ids: Record = {}; const open = async (key: string, approvers: any[], extra: Record = {}) => { - await engine.insert('crm_deal', { id: `D_${key}`, title: key }, { context: SYSTEM } as any); + await engine.insert('crm_deal', { id: `D_${key}`, title: key, organization_id: orgOfDeal(key) }, { context: SYSTEM } as any); const row: any = await svc.openNodeRequest({ object: 'crm_deal', recordId: `D_${key}`, runId: `run_${key}`, nodeId: 'review', flowName: FLOW.name, config: { approvers, behavior: 'first_response', ...extra } as any, @@ -256,4 +276,21 @@ describe('every approval notification carries its text in body', () => { } expect(offenders).toEqual([]); }); + + it("[#23284] every notification carries its request's organization, read back from the stored request", async () => { + const stored = new Map(); + for (const id of Object.values(ids)) { + const row: any = await engine.findOne('sys_approval_request', { where: { id }, fields: ['id', 'organization_id'] } as any, { context: SYSTEM } as any); + stored.set(id, row?.organization_id ?? null); + } + // The fixture really is two organizations, and each request landed in its + // deal's — otherwise the identity below could hold for a constant. + expect(new Set(stored.values())).toEqual(new Set([ORG_A, ORG_B])); + for (const [key, id] of Object.entries(ids)) expect(stored.get(id), key).toBe(orgOfDeal(key)); + + const offenders = emits + .filter((e) => e.organizationId !== stored.get(e.source?.id ?? '')) + .map((e) => `${siteOf(e)}: organizationId ${JSON.stringify(e.organizationId)}, request ${JSON.stringify(stored.get(e.source?.id ?? ''))}`); + expect(offenders).toEqual([]); + }); }); diff --git a/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts b/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts index 8849129ca80..f8a581005b8 100644 --- a/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts +++ b/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts @@ -38,6 +38,20 @@ import { assertEngineFindOnePredicate, type EngineFindOneQueryInput } from '@obj * `organization_id` is worse than a null, because a null is visibly missing * while a wrong value is silently authoritative to every report, export and * cleanup script that filters by organization. + * + * [#23284] A run that carries NO organization — a schedule tick, a record + * change a system context wrote — takes the organization of the record the + * notification is about: the declared click-through record, else the run's + * trigger record, each read back from the STORED row (PIN E–H). Measured on + * HotCRM at 18.0.0-next.0, every such emit landed `sys_notification` with + * `organization_id` NULL. A notification about no record, from such a run, + * reaches `emit()` with no organization and the engine decides its owner + * (ADR-0131 D9, now that `sys_notification` is admitted): that half is pinned + * where the rule lives — `tenancy-by-object-classification.test.ts` in + * `@objectstack/objectql` and `notification-organization-follows-event.integration.test.ts` + * in `@objectstack/service-messaging`. Here: the node hands it on with no + * organization (PIN C) and fails the step with the engine's code when the + * engine refuses it (PIN D). */ function silentLogger(): any { @@ -62,13 +76,28 @@ function recordingLogger(): { logger: any; warnings: string[] } { /** Every row this stack wrote, in insertion order, with the object it landed in. */ interface WrittenRow { object: string; row: Record } +/** One business record the data engine holds, as it is STORED. */ +interface StoredRecord { object: string; row: Record & { id: string } } + +/** One `findOne` the stack issued, verbatim. */ +interface ObservedRead { object: string; query: unknown; options: unknown } + /** - * A capturing data engine. Reads answer empty (no preference rows, no dedup - * hit) so the default always-on `inbox` channel is the one that runs; writes - * are recorded verbatim, which is the only thing these pins assert on. + * A capturing data engine. Reads of the notification family answer empty (no + * preference rows, no dedup hit) so the default always-on `inbox` channel is + * the one that runs; writes are recorded verbatim, which is what these pins + * assert on. + * + * [#23284] It also holds the business records a notification can be about + * (`records`), answers `findOne` by id for them, and describes their schema + * through `getSchema` — the one member the wall-column resolver reads — with + * the kernel's `organization_id` column, as a registered object carries it. + * Every `findOne` is recorded, so a pin can say which read happened, and that + * none did. */ -function capturingEngine(): { engine: any; written: WrittenRow[] } { +function capturingEngine(records: readonly StoredRecord[] = []): { engine: any; written: WrittenRow[]; reads: ObservedRead[] } { const written: WrittenRow[] = []; + const reads: ObservedRead[] = []; let seq = 0; const engine = { async insert(object: string, row: Record) { @@ -77,10 +106,27 @@ function capturingEngine(): { engine: any; written: WrittenRow[] } { return { ...row, id }; }, async find() { return []; }, - async findOne(object: string, query?: EngineFindOneQueryInput) { - assertEngineFindOnePredicate(object, query); return undefined; }, + async findOne(object: string, query?: EngineFindOneQueryInput, options?: unknown) { + assertEngineFindOnePredicate(object, query); + reads.push({ object, query, options }); + const id = (query as { where?: { id?: unknown } } | undefined)?.where?.id; + const hit = records.find((r) => r.object === object && r.row.id === id); + return hit ? { ...hit.row } : undefined; + }, + getSchema(object: string) { + if (!records.some((r) => r.object === object)) return undefined; + return { + name: object, + fields: { name: { type: 'text' }, organization_id: { type: 'lookup', reference: 'sys_organization' } }, + }; + }, }; - return { engine, written }; + return { engine, written, reads }; +} + +/** Only the reads a pin is about — the business records, not the family's own lookups. */ +function recordReads(reads: readonly ObservedRead[]): ObservedRead[] { + return reads.filter((r) => !NOTIFICATION_FAMILY.has(r.object)); } /** The four tables the ruling names for the notification family. */ @@ -102,7 +148,7 @@ function orgIdentities(written: WrittenRow[]): string[] { .map((w) => `${w.object}:${w.row.organization_id ?? 'NULL'}`); } -function notifyFlow(): any { +function notifyFlow(about?: { sourceObject: string; sourceId: string }): any { return { name: 'nudge', label: 'Nudge', @@ -119,6 +165,7 @@ function notifyFlow(): any { title: 'Renewal due', message: 'Ping', channels: ['inbox'], + ...(about ?? {}), }, }, { id: 'end', type: 'end' as const, label: 'End' }, @@ -135,20 +182,65 @@ function notifyFlow(): any { * node — the seam under test is precisely the handoff between them, so a fake * that answers `emit()` in one shot could not express it. */ -function bootInlineStack(logger: any = silentLogger()) { - const { engine: data, written } = capturingEngine(); +function bootInlineStack( + logger: any = silentLogger(), + opts: { records?: readonly StoredRecord[]; about?: { sourceObject: string; sourceId: string } } = {}, +) { + const { engine: data, written, reads } = capturingEngine(opts.records); const messaging = new MessagingService({ logger, getData: () => data }); messaging.registerChannel(createInboxChannel({ getData: () => data })); const engine = new AutomationEngine(logger); registerNotifyNode(engine, { logger, - getService: (name: string) => (name === 'messaging' ? messaging : undefined), + getService: (name: string) => (name === 'messaging' ? messaging : name === 'data' ? data : undefined), + } as any); + engine.registerFlow('nudge', notifyFlow(opts.about)); + return { engine, messaging, written, reads }; +} + +/** + * A notify node over a messaging service that only RECORDS what it is handed + * (or throws what it is told to), with `data` serving the record reads — for + * the pins that are about the emit input itself. + */ +function bootRecordingStack(opts: { + records?: readonly StoredRecord[]; + about?: { sourceObject: string; sourceId: string }; + emitThrows?: unknown; +} = {}) { + const { engine: data, reads } = capturingEngine(opts.records); + const emitted: any[] = []; + const service: MessagingServiceSurface = { + async emit(n: any) { + emitted.push(n); + if (opts.emitThrows !== undefined) throw opts.emitThrows; + return { notificationId: 'evt_1', delivered: n.audience.length, failed: 0 }; + }, + }; + const engine = new AutomationEngine(silentLogger()); + registerNotifyNode(engine, { + logger: silentLogger(), + getService: (name: string) => (name === 'messaging' ? service : name === 'data' ? data : undefined), } as any); - engine.registerFlow('nudge', notifyFlow()); - return { engine, messaging, written }; + engine.registerFlow('nudge', notifyFlow(opts.about)); + return { engine, emitted, reads }; } +/** The context a schedule tick hands an organization-less flow (`ScheduleTrigger`). */ +const SCHEDULE_TICK = { + event: 'schedule', + params: { jobId: 'tick-1', flowName: 'nudge', schedule: { type: 'cron', expression: '0 * * * *' } }, + callerParamKeys: [], +} as any; + +/** The case a breach notification is about, and the organization it is stored under. */ +const CASE = { object: 'crm_case', row: { id: 'case_1', name: 'Breached', organization_id: 'org_case_owner' } }; +const ABOUT_CASE = { sourceObject: 'crm_case', sourceId: 'case_1' }; + +/** The record a system-written record-change run is about. */ +const TASK = { object: 'crm_task', row: { id: 'task_1', name: 'Urgent', organization_id: 'org_task_owner' } }; + describe('#11303 — the notify producer stamps organization_id on the notification family', () => { it('PIN A: threads the run\'s own organization onto the emit input', async () => { const emitted: any[] = []; @@ -212,13 +304,29 @@ describe('#11303 — the notify producer stamps organization_id on the notificat expect(rows[0].organizationId).toBe('org_pin_alpha'); }); - it('PIN C (over-denial control): a stack with no organization in scope still delivers', async () => { + it('PIN C: a notification about no record, from a run with no organization, reaches emit() with none', async () => { + // [#23284] The owner of what is left is the ENGINE's to decide + // (ADR-0131 D9): the sole organization under `single`, a refusal where + // none is derivable — pinned in objectql and service-messaging. What + // the node owes is to hand it on WITHOUT inventing one: no + // `organizationId` key at all (absent, not null), and no read, since + // nothing names a record. + const { engine, emitted, reads } = bootRecordingStack({ records: [CASE] }); + + const run = await engine.execute('nudge', SCHEDULE_TICK); + + expect(run.success).toBe(true); + expect(emitted).toHaveLength(1); + expect('organizationId' in emitted[0]).toBe(false); + expect(recordReads(reads)).toEqual([]); + }); + + it('PIN C2 (over-denial control): an organization-less stack with no engine rule still delivers', async () => { // The control that stops the fix from degenerating into "refuse unless - // an organization is present". A `single`-posture deployment — and every - // fresh boot before the first organization exists — has no organization - // to thread, and a notify there must still emit and still write its rows. - // ⭐ A suite that only pinned "organization_id is present" would score - // green on an implementation that breaks exactly this deployment. + // an organization is present" inside the NODE: with no rule below it + // (this capturing engine runs none), an organization-less emit is still + // emitted and its rows are still written. Refusing is the engine's + // call, never the node's. const { engine, written } = bootInlineStack(); const run = await engine.execute('nudge'); @@ -231,19 +339,124 @@ describe('#11303 — the notify producer stamps organization_id on the notificat ]); }); - it('PIN D (fail-loud, not fail-guess): an unresolvable organization warns audibly', async () => { - // Fail-LOUD by warning rather than refusing — see PIN C for why a - // refusal is not available here. The warning is what makes the org-less - // row a visible event instead of a silent one, and it must name the - // topic so the operator can find the producer. - const { logger, warnings } = recordingLogger(); - const { engine } = bootInlineStack(logger); + it('PIN D: the engine refusing an owner-less notification FAILS the step, naming the engine code', async () => { + // [#23284] Where ADR-0131 D9 derives no owner the engine refuses the + // `sys_notification` insert with `ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED` + // (500). The node must not swallow that into a logged success: the + // step fails and its failure text carries the code an operator greps. + const refusal = Object.assign( + new Error("Insert on 'sys_notification' was REFUSED: a system-context write on a tenant-scoped object must carry an organization"), + { code: 'ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED', status: 500 }, + ); + const { engine, emitted } = bootRecordingStack({ emitThrows: refusal }); - const run = await engine.execute('nudge'); + const run = await engine.execute('nudge', SCHEDULE_TICK); + + expect(emitted).toHaveLength(1); + expect(run.success).toBe(false); + expect(run.error).toContain('notify failed: ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED: '); + expect(run.error).toContain("Insert on 'sys_notification' was REFUSED"); + }); +}); +describe('#23284 — a run with no organization takes the organization of the record the notification is about', () => { + it('PIN E: a schedule-tick notify about a record stores THAT record\'s organization on all four row kinds', async () => { + // Inline fan-out: the event, the inbox row, the delivered receipt. + const inline = bootInlineStack(silentLogger(), { records: [CASE], about: ABOUT_CASE }); + const run = await inline.engine.execute('nudge', SCHEDULE_TICK); expect(run.success).toBe(true); - const line = warnings.find((w) => w.includes('organization')); - expect(line, `no organization warning in: ${JSON.stringify(warnings)}`).toBeDefined(); - expect(line).toContain('deal.won'); + expect(orgIdentities(inline.written)).toEqual([ + `${NOTIFICATION_EVENT_OBJECT}:org_case_owner`, + `${INBOX_OBJECT}:org_case_owner`, + `${RECEIPT_OBJECT}:org_case_owner`, + ]); + // The organization is read off the STORED row, by id, elevated — the + // one read the node issues, projecting only the wall column. + expect(recordReads(inline.reads)).toEqual([{ + object: 'crm_case', + query: { where: { id: 'case_1' }, fields: ['id', 'organization_id'] }, + options: { context: { isSystem: true } }, + }]); + + // Durable path: the delivery row. + const { engine: data } = capturingEngine([CASE]); + const outbox = new MemoryNotificationOutbox(1); + const messaging = new MessagingService({ logger: silentLogger(), getData: () => data, outbox }); + messaging.registerChannel(createInboxChannel({ getData: () => data })); + const durable = new AutomationEngine(silentLogger()); + registerNotifyNode(durable, { + logger: silentLogger(), + getService: (name: string) => (name === 'messaging' ? messaging : name === 'data' ? data : undefined), + } as any); + durable.registerFlow('nudge', notifyFlow(ABOUT_CASE)); + expect((await durable.execute('nudge', SCHEDULE_TICK)).success).toBe(true); + const rows = await outbox.list(); + expect(rows.map((r) => r.organizationId)).toEqual(['org_case_owner']); + }); + + it('PIN F: a record change written by a system context takes the TRIGGER record\'s stored organization', async () => { + // The snapshot the run carries names a different organization on + // purpose: a trigger door can hand a run a caller-written `record`, so + // the node reads the stored row and never the snapshot's column. + const { engine, emitted, reads } = bootRecordingStack({ records: [TASK] }); + + const run = await engine.execute('nudge', { + event: 'on_create', + object: 'crm_task', + record: { id: 'task_1', name: 'Urgent', organization_id: 'org_from_the_snapshot' }, + } as any); + + expect(run.success).toBe(true); + expect(emitted.map((e) => e.organizationId)).toEqual(['org_task_owner']); + expect(recordReads(reads).map((r) => r.object)).toEqual(['crm_task']); + }); + + it('PIN G: the declared click-through record is asked BEFORE the trigger record', async () => { + const { engine, emitted, reads } = bootRecordingStack({ records: [CASE, TASK], about: ABOUT_CASE }); + + const run = await engine.execute('nudge', { event: 'on_create', object: 'crm_task', record: { id: 'task_1' } } as any); + + expect(run.success).toBe(true); + expect(emitted.map((e) => e.organizationId)).toEqual(['org_case_owner']); + // The first anchor answered, so the second was never read. + expect(recordReads(reads).map((r) => r.object)).toEqual(['crm_case']); + }); + + it('PIN H: a record that names no organization falls through to the next anchor, then to none', async () => { + const orgless = { object: 'crm_case', row: { id: 'case_1', name: 'Unowned', organization_id: null as unknown as string } }; + const { engine, emitted, reads } = bootRecordingStack({ records: [orgless, TASK], about: ABOUT_CASE }); + + await engine.execute('nudge', { event: 'on_create', object: 'crm_task', record: { id: 'task_1' } } as any); + // A schedule tick about a record that is gone: nothing to read, no organization. + const gone = bootRecordingStack({ records: [], about: ABOUT_CASE }); + await gone.engine.execute('nudge', SCHEDULE_TICK); + + expect(emitted.map((e) => e.organizationId)).toEqual(['org_task_owner']); + expect(recordReads(reads).map((r) => r.object)).toEqual(['crm_case', 'crm_task']); + expect(gone.emitted).toHaveLength(1); + expect('organizationId' in gone.emitted[0]).toBe(false); + }); + + it('PIN I (user-driven, unchanged): a run that carries an organization emits exactly as before, and reads nothing', async () => { + // The run's organization wins over every record anchor — here the + // click-through case lives in ANOTHER organization — and the emit input + // is the very object the node built before this card, key for key. + const { engine, emitted, reads } = bootRecordingStack({ records: [CASE], about: ABOUT_CASE }); + + const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha', userId: 'user_9' } as any); + + expect(run.success).toBe(true); + expect(emitted).toHaveLength(1); + expect(emitted[0]).toStrictEqual({ + topic: 'deal.won', + audience: ['user_1'], + payload: { title: 'Renewal due', body: 'Ping', url: undefined }, + severity: undefined, + source: { object: 'crm_case', id: 'case_1' }, + actorId: undefined, + organizationId: 'org_pin_alpha', + channels: ['inbox'], + }); + expect(recordReads(reads)).toEqual([]); }); }); diff --git a/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts b/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts new file mode 100644 index 00000000000..0b76cabe6c0 --- /dev/null +++ b/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts @@ -0,0 +1,135 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#23284] A notification no producer could attribute takes the owner the + * engine derives for a system write (ADR-0131 D9), and every row below it + * follows that event row. + * + * ## What was measured + * + * On HotCRM at 18.0.0-next.0 (one `single`-posture install with its Default + * Organization) every notification emitted from a system-context run — a + * schedule tick, a record change a system context wrote, an approval node — + * stored `sys_notification.organization_id` NULL. The producer passed no + * organization, and the engine's system-write rule never ran for the table: it + * was `unclassified` in the #13491 ledger, so it sat outside the rule that + * derived its own DELIVERY row's organization one write later. On an install + * holding two organizations the same emit stored the NULL event row and had + * every delivery refused, with the notify step reporting success. + * + * ## What this pins, on a real `ObjectQL` over `SqlDriver` (better-sqlite3) + * + * - `single`, one organization: an emit that carries no organization stores + * the sole organization on the event row AND on the inbox row, the receipt + * (inline fan-out) and the delivery row (outbox) — the children follow the + * stored event row, so the family cannot split; + * - `single`, two organizations: the same emit is REFUSED with the engine's + * own envelope and nothing is written, the event row included; + * - an emit that carries its organization is stored with exactly that one. + * + * The producer half — WHICH organization a notify node, or the approvals + * service, threads — is pinned at each producer + * (`notify-organization-stamp.integration.test.ts`, + * `approval-notification-body.integration.test.ts`). + */ + +import { describe, it, expect, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { SysNotification } from '@objectstack/platform-objects'; +import { MessagingService, NOTIFICATION_EVENT_OBJECT } from './messaging-service.js'; +import { createInboxChannel, INBOX_OBJECT, RECEIPT_OBJECT } from './inbox-channel.js'; +import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; +import { InboxMessage, NotificationDelivery, NotificationPreference, NotificationReceipt } from './objects/index.js'; + +const SYSTEM = { context: { isSystem: true } } as const; + +const SYS_ORGANIZATION = { name: 'sys_organization', label: 'Organization', fields: { name: { type: 'text', label: 'Name' } } }; +const SYS_USER = { name: 'sys_user', label: 'User', fields: { name: { type: 'text', label: 'Name' } } }; + +const silentLogger: any = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} }; +silentLogger.child = () => silentLogger; + +let engine: ObjectQL | undefined; + +afterEach(async () => { + try { await engine?.destroy(); } catch { /* noop */ } + engine = undefined; +}); + +/** A real engine on the `single` posture holding `organizations`, and one recipient. */ +async function boot(organizations: readonly string[]): Promise { + const ql = new ObjectQL(); + ql.registerDriver(new SqlDriver({ client: 'better-sqlite3', connection: { filename: ':memory:' }, useNullAsDefault: true }), true); + await ql.init(); + for (const def of [SYS_ORGANIZATION, SYS_USER, SysNotification, InboxMessage, NotificationReceipt, NotificationDelivery, NotificationPreference]) { + ql.registry.registerObject(def as any, '@objectstack/service-messaging'); + } + await ql.syncSchemas(); + ql.setTenancyPostureProvider(() => 'single'); + for (const id of organizations) await ql.insert('sys_organization', { id, name: id }, SYSTEM as any); + await ql.insert('sys_user', { id: 'user_1', name: 'Recipient' }, SYSTEM as any); + engine = ql; + return ql; +} + +async function organizationsOf(ql: ObjectQL, object: string): Promise { + const rows = (await ql.find(object, { fields: ['id', 'organization_id'] } as any, SYSTEM as any)) ?? []; + return rows.map((r: any) => r.organization_id ?? 'NULL'); +} + +const EMIT = { topic: 'sweep.breach', audience: ['user_1'], payload: { title: 'Breach', body: 'An SLA was breached.' } }; + +describe('#23284 — a notification no producer attributed takes the owner ADR-0131 D9 derives', () => { + it('single, one organization: the event row and every row below it carry the sole organization (inline)', async () => { + const ql = await boot(['org_sole']); + const messaging = new MessagingService({ logger: silentLogger, getData: () => ql as any }); + messaging.registerChannel(createInboxChannel({ getData: () => ql as any })); + + const result = await messaging.emit({ ...EMIT, channels: ['inbox'] }); + + expect(result.delivered).toBe(1); + expect(await organizationsOf(ql, NOTIFICATION_EVENT_OBJECT)).toEqual(['org_sole']); + expect(await organizationsOf(ql, INBOX_OBJECT)).toEqual(['org_sole']); + expect(await organizationsOf(ql, RECEIPT_OBJECT)).toEqual(['org_sole']); + }); + + it('single, one organization: the delivery row carries the event row\'s organization (outbox)', async () => { + const ql = await boot(['org_sole']); + const outbox = new SqlNotificationOutbox(ql as any, { partitionCount: 1 }); + const messaging = new MessagingService({ logger: silentLogger, getData: () => ql as any, outbox }); + messaging.registerChannel(createInboxChannel({ getData: () => ql as any })); + + const result = await messaging.emit({ ...EMIT, channels: ['inbox'] }); + + expect(result.enqueued).toBe(1); + expect(await organizationsOf(ql, NOTIFICATION_EVENT_OBJECT)).toEqual(['org_sole']); + expect(await organizationsOf(ql, DELIVERY_OBJECT)).toEqual(['org_sole']); + }); + + it('single, two organizations: the emit is REFUSED with the engine envelope, and nothing is written', async () => { + const ql = await boot(['org_a', 'org_b']); + const messaging = new MessagingService({ logger: silentLogger, getData: () => ql as any }); + messaging.registerChannel(createInboxChannel({ getData: () => ql as any })); + + await expect(messaging.emit({ ...EMIT, channels: ['inbox'] })).rejects.toMatchObject({ + code: 'ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED', + status: 500, + }); + // Before #23284 this stored the event row with organization_id NULL. + expect(await organizationsOf(ql, NOTIFICATION_EVENT_OBJECT)).toEqual([]); + expect(await organizationsOf(ql, INBOX_OBJECT)).toEqual([]); + }); + + it('an emit that carries its organization is stored with exactly that one, on every row', async () => { + const ql = await boot(['org_a', 'org_b']); + const messaging = new MessagingService({ logger: silentLogger, getData: () => ql as any }); + messaging.registerChannel(createInboxChannel({ getData: () => ql as any })); + + await messaging.emit({ ...EMIT, channels: ['inbox'], organizationId: 'org_b' }); + + expect(await organizationsOf(ql, NOTIFICATION_EVENT_OBJECT)).toEqual(['org_b']); + expect(await organizationsOf(ql, INBOX_OBJECT)).toEqual(['org_b']); + expect(await organizationsOf(ql, RECEIPT_OBJECT)).toEqual(['org_b']); + }); +}); From d564b21d57568b904ab7a9c9b5caf1705f60196b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 20:45:28 +0000 Subject: [PATCH 3/5] chore: changeset for the notification organization narrowing Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- ...-notification-takes-record-organization.md | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 .changeset/23284-notification-takes-record-organization.md diff --git a/.changeset/23284-notification-takes-record-organization.md b/.changeset/23284-notification-takes-record-organization.md new file mode 100644 index 00000000000..1d92e934b6e --- /dev/null +++ b/.changeset/23284-notification-takes-record-organization.md @@ -0,0 +1,30 @@ +--- +'@objectstack/objectql': minor +'@objectstack/service-messaging': minor +'@objectstack/service-automation': minor +'@objectstack/plugin-approvals': patch +--- + +fix(objectql,service-messaging,service-automation,plugin-approvals)!: a notification carries the organization of what it is about, and one no organization owns is refused instead of stored with `organization_id` NULL + +Clause-②: no (narrowing) — notification rows gain the organization they should already carry; what narrows is an emit that names no organization on an install where none is derivable, which the engine now refuses where it used to store the event row with `organization_id` NULL. + + + +**BREAKING** (an accept-set narrowing), shipped as `minor` under the launch-window convention for breaking changes. + +Every notification emitted from a system-context run — a scheduled flow, a record change written by a system context, an approval node — was stored in `sys_notification` with `organization_id` NULL, so reports and cleanup that filter by organization did not see it. Each emit also logged a `[notify] no organization in scope` warning. On an install holding several organizations, the deliveries of the same emit were refused while the flow step reported success, so nothing reached the inbox. + +**Who owns a notification now:** + +- **The run's organization.** A run that carries one emits exactly as before. +- **Else the record it is about.** A `notify` node in a run with no organization reads the organization of the record the notification is about. That is the declared click-through record (`sourceObject` + `sourceId`), else the run's trigger record. It is read from the stored row by id, never from the record snapshot the run carries. +- **Approvals.** Every approval notification carries its request's organization: `approval.requested`, reminders, comments, send-backs, reassignments, escalations and out-of-office notices. +- **Else the owner the engine derives for a system write** (ADR-0131 D9). `sys_notification` is now in the platform-object tenancy ledger, so an emit that names no organization gets the install's sole organization under the `single` posture. +- **The rows below follow the event.** `emit()` stamps the stored event row's organization on its delivery, inbox row and receipt, on both the inline and the outbox path. + +**Refused now:** where no organization is derivable (a walled posture, or a `single` install holding more than one organization), an emit that names none is refused with `ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED`. Nothing is written for it. A `notify` step that hits the refusal fails with `notify failed: ERR_SYSTEM_WRITE_ORGANIZATION_REQUIRED: …` instead of succeeding. `MessagingService.emit()` rejects with the engine's error for any other producer. + +**What to do when it refuses:** give the notification an owner. Name the record it is about on the `notify` node (`sourceObject` + `sourceId`), declare the scheduled flow's acting organization (`organization` on its start node), or pass `organizationId` to `emit()` from your own code. + +The `[notify] no organization in scope` warning is gone, because the NULL row it predicted is no longer written. Rows already stored with `organization_id` NULL are not rewritten here. The v18 organization-ownership ceremony attributes them from their source record. From b389aa57dd94d069645d728f9440f0b47819fe6a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 21:11:49 +0000 Subject: [PATCH 4/5] test: type the read options in the new organization pins Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- .../approval-notification-body.integration.test.ts | 2 +- ...ion-organization-follows-event.integration.test.ts | 11 ++++++----- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts b/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts index 4c2c225b613..5fa0a6c0df4 100644 --- a/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts +++ b/packages/plugins/plugin-approvals/src/approval-notification-body.integration.test.ts @@ -280,7 +280,7 @@ describe('every approval notification carries its text in body', () => { it("[#23284] every notification carries its request's organization, read back from the stored request", async () => { const stored = new Map(); for (const id of Object.values(ids)) { - const row: any = await engine.findOne('sys_approval_request', { where: { id }, fields: ['id', 'organization_id'] } as any, { context: SYSTEM } as any); + const row = await engine.findOne('sys_approval_request', { where: { id }, fields: ['id', 'organization_id'] }, { context: SYSTEM }); stored.set(id, row?.organization_id ?? null); } // The fixture really is two organizations, and each request landed in its diff --git a/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts b/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts index 0b76cabe6c0..d4d19d0b2d3 100644 --- a/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts +++ b/packages/services/service-messaging/src/notification-organization-follows-event.integration.test.ts @@ -35,6 +35,7 @@ import { describe, it, expect, afterEach } from 'vitest'; import { ObjectQL } from '@objectstack/objectql'; +import type { ExecutionContext } from '@objectstack/spec/kernel'; import { SqlDriver } from '@objectstack/driver-sql'; import { SysNotification } from '@objectstack/platform-objects'; import { MessagingService, NOTIFICATION_EVENT_OBJECT } from './messaging-service.js'; @@ -42,7 +43,7 @@ import { createInboxChannel, INBOX_OBJECT, RECEIPT_OBJECT } from './inbox-channe import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; import { InboxMessage, NotificationDelivery, NotificationPreference, NotificationReceipt } from './objects/index.js'; -const SYSTEM = { context: { isSystem: true } } as const; +const SYSTEM = { context: { isSystem: true } as ExecutionContext }; const SYS_ORGANIZATION = { name: 'sys_organization', label: 'Organization', fields: { name: { type: 'text', label: 'Name' } } }; const SYS_USER = { name: 'sys_user', label: 'User', fields: { name: { type: 'text', label: 'Name' } } }; @@ -67,15 +68,15 @@ async function boot(organizations: readonly string[]): Promise { } await ql.syncSchemas(); ql.setTenancyPostureProvider(() => 'single'); - for (const id of organizations) await ql.insert('sys_organization', { id, name: id }, SYSTEM as any); - await ql.insert('sys_user', { id: 'user_1', name: 'Recipient' }, SYSTEM as any); + for (const id of organizations) await ql.insert('sys_organization', { id, name: id }, SYSTEM); + await ql.insert('sys_user', { id: 'user_1', name: 'Recipient' }, SYSTEM); engine = ql; return ql; } async function organizationsOf(ql: ObjectQL, object: string): Promise { - const rows = (await ql.find(object, { fields: ['id', 'organization_id'] } as any, SYSTEM as any)) ?? []; - return rows.map((r: any) => r.organization_id ?? 'NULL'); + const rows = (await ql.find(object, { fields: ['id', 'organization_id'] }, SYSTEM)) ?? []; + return rows.map((r: Record) => String(r.organization_id ?? 'NULL')); } const EMIT = { topic: 'sweep.breach', audience: ['user_1'], payload: { title: 'Breach', body: 'An SLA was breached.' } }; From 238d07b36509b71682396b2fcc0451d6206899c9 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 22:11:17 +0000 Subject: [PATCH 5/5] fix: drop the orphaned test helper; grade plugin-approvals minor Claude-Session: https://claude.ai/code/session_01FzQy26KznSa5j2gnf3AzsE Co-authored-by: Claude --- .../23284-notification-takes-record-organization.md | 4 ++-- .../notify-organization-stamp.integration.test.ts | 13 ------------- 2 files changed, 2 insertions(+), 15 deletions(-) diff --git a/.changeset/23284-notification-takes-record-organization.md b/.changeset/23284-notification-takes-record-organization.md index 1d92e934b6e..ac5df187f65 100644 --- a/.changeset/23284-notification-takes-record-organization.md +++ b/.changeset/23284-notification-takes-record-organization.md @@ -2,14 +2,14 @@ '@objectstack/objectql': minor '@objectstack/service-messaging': minor '@objectstack/service-automation': minor -'@objectstack/plugin-approvals': patch +'@objectstack/plugin-approvals': minor --- fix(objectql,service-messaging,service-automation,plugin-approvals)!: a notification carries the organization of what it is about, and one no organization owns is refused instead of stored with `organization_id` NULL Clause-②: no (narrowing) — notification rows gain the organization they should already carry; what narrows is an emit that names no organization on an install where none is derivable, which the engine now refuses where it used to store the event row with `organization_id` NULL. - + **BREAKING** (an accept-set narrowing), shipped as `minor` under the launch-window convention for breaking changes. diff --git a/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts b/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts index f8a581005b8..b73398b34e0 100644 --- a/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts +++ b/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts @@ -60,19 +60,6 @@ function silentLogger(): any { return l; } -/** A logger that records every warning line, for the fail-loud pin. */ -function recordingLogger(): { logger: any; warnings: string[] } { - const warnings: string[] = []; - const l: any = { - info: () => {}, - warn: (...args: unknown[]) => { warnings.push(args.map(String).join(' ')); }, - error: () => {}, - debug: () => {}, - }; - l.child = () => l; - return { logger: l, warnings }; -} - /** Every row this stack wrote, in insertion order, with the object it landed in. */ interface WrittenRow { object: string; row: Record }