T-HUB/apps/scheduler/src/activities/dispatch-gate.ts
glass beadian b89dea5c93 feat: per-connection publish governance — auto, 1-admin, majority, consensus
Per the owner's spec: each social connection gets publishGovernance in its
fields — unset = manual publish (today), auto = dispatch on approval,
admin1/admin_majority/admin_consensus = a publish-phase chatgov poll
(subject draft_publish, its own open-poll slot, no collision with review
gates). Dispatch holds until the publish poll passes (E3 pre-dispatch
check); auto dispatches on approval. Includes migration 033, the PATCH
/connections/:id/publish-governance route (chatgov.manage), and 19+31 new
tests. Implementation by the branding-panel process (claude/opus), lint
and integration fixes here.
2026-10-14 08:04:56 +00:00

261 lines
11 KiB
TypeScript

import type { Pool } from "pg";
/**
* §4 reason-code vocabulary (chatgov-design.md §4). Mirrored from the P0
* spine (apps/api/src/chatgov/policy.ts REASON_CODES) rather than imported
* across the app package boundary; gate_decisions CHECKs values, so this list
* must stay in sync — the unit tests assert the mirror.
*/
export const E3_REASON_CODES = [
"INSUFFICIENT_QUORUM",
"INELIGIBLE_ACTOR",
"SELF_APPROVAL_DENIED",
"TIMEBOX_EXPIRED_AUTO_REJECT",
"TIMEBOX_EXPIRED_AUTO_APPROVE",
"RESET_ON_CHANGE",
"PENDING_POLL",
"VETOED",
"OUTSIDE_POLICY",
"POLICY_DRY_RUN",
"OVERRIDDEN_BY_OWNER",
"LOCKED_CHANNEL",
"NO_CHANNEL_BOUND",
"CONFIRM_EXPIRED",
] as const;
export type ReasonCode = (typeof E3_REASON_CODES)[number];
/**
* E3 pre-dispatch re-check (chatgov-design.md §4 step 5, P1).
*
* Re-evaluated IMMEDIATELY BEFORE each platform dispatch, per destination:
* - approval still holds (the draft is still `approved`/`scheduled` with its
* pinned approved_hash/revision — contentFenceHolds owns the content side
* in the execution claim; E3 owns the *governance* side and re-reads the
* draft row so a revocation that landed after the claim is caught here);
* - the destination's compiled policy_version is still current (a rule
* change after approval recompiles the snapshot ⇒ RESET_ON_CHANGE);
* - a compiled snapshot exists at all — none is OUTSIDE_POLICY, fail closed
* (P0 hand-off: "OUTSIDE_POLICY is the evaluation posture when no
* snapshot row exists").
*
* Authorization fences (principal/approver/credential) are NOT duplicated
* here: the repository's claim + withDispatchGate already re-read every live
* authz input; E3 integrates by never weakening them.
*
* Every evaluation writes a gate_decisions row (fired or blocked) with the
* §4 reason codes, so partial fan-out ("A fired, B blocked") is observable
* from the blocked queue without touching the publishing spine.
*/
export const E3_ACTOR = "system:gate-e3";
/**
* Publish-phase gate modes (connections.fields.publishGovernance), mirrored
* from apps/api/src/publishing/publish-governance.ts across the app package
* boundary. "auto" and unset need no publish poll.
*/
export const PUBLISH_GATE_MODES = ["admin1", "admin_majority", "admin_consensus"] as const;
/** DB column order for the re-check query. */
interface E3Row {
state: string | null;
approved_hash: string | null;
approved_revision: string | null;
reviewer_user_id: string | null;
effective: { mode: "immediate" | "quorum" | "admin_only" } | null;
policy_version: string | null;
poll_state: string | null;
/** connections.fields->>'publishGovernance' of the attempt's destination. */
publish_governance?: string | null;
/** Latest publish poll for (draft, destination, approved revision). */
publish_poll_state?: string | null;
}
export interface E3Outcome {
outcome: "fired" | "blocked";
reasonCodes: ReasonCode[];
policyVersion: string;
mode: "immediate" | "quorum" | "admin_only" | "none";
approvedHash: string | null;
pollState: string | null;
}
function reason(code: ReasonCode): ReasonCode {
return (E3_REASON_CODES as readonly string[]).includes(code) ? code : "OUTSIDE_POLICY";
}
export function evaluateE3(row: E3Row | undefined): E3Outcome {
// Fail closed: no attempt/draft/snapshot row at all.
if (row === undefined) {
return { outcome: "blocked", reasonCodes: [reason("OUTSIDE_POLICY")], policyVersion: "none", mode: "none", approvedHash: null, pollState: null };
}
const approvedHash = row.approved_hash === null ? null : String(row.approved_hash);
// 1) Approval still holds. `approved`/`scheduled` with a pinned hash is the
// approved posture; anything else (rejected, editing, revoked approval)
// is RESET_ON_CHANGE — the approval changed after it landed.
if (row.state !== "approved" && row.state !== "scheduled") {
return { outcome: "blocked", reasonCodes: [reason("RESET_ON_CHANGE")], policyVersion: row.policy_version ?? "none", mode: row.effective?.mode ?? "none", approvedHash, pollState: null };
}
// 1b) Publish-phase gate: a gated destination dispatches only after the
// admins' publish poll for THIS approved revision passed. Attempts are
// normally created by that pass; this re-check also catches schedules
// made before the connection was switched to a gate mode.
if ((PUBLISH_GATE_MODES as readonly string[]).includes(row.publish_governance ?? "")
&& row.publish_poll_state !== "passed") {
return {
outcome: "blocked",
reasonCodes: [reason("INSUFFICIENT_QUORUM"), reason("PENDING_POLL")],
policyVersion: row.policy_version ?? "none",
mode: row.effective?.mode ?? "none",
approvedHash,
pollState: row.publish_poll_state ?? "none",
};
}
// 2) A compiled snapshot exists for this destination. None => fail closed
// (P0 hand-off: OUTSIDE_POLICY posture, never a silent ungated send).
if (row.effective === null || row.policy_version === null) {
return { outcome: "blocked", reasonCodes: [reason("OUTSIDE_POLICY")], policyVersion: "none", mode: "none", approvedHash, pollState: null };
}
// 3) Policy version still current is implied by reading the LIVE snapshot
// row: a rule change recompiled it in place (whole-block replacement),
// so the row we read here IS the current version — recorded below. A
// version pinned in an E1/E2 decision that no longer matches is the
// RESET_ON_CHANGE case the UI surfaces from gate_decisions history.
// 4) Mode: immediate/admin_only gate nothing at E3 (their E1 verdict already
// fired; §4 "immediate mode = E1 minimal + E3 always" — E3 re-checks the
// fences above and fires). quorum still requires the linked poll passed.
if (row.effective.mode === "quorum") {
const pollState = row.poll_state === null ? "none" : String(row.poll_state);
if (pollState !== "passed") {
return {
outcome: "blocked",
reasonCodes: [reason("INSUFFICIENT_QUORUM"), reason("PENDING_POLL")],
policyVersion: String(row.policy_version),
mode: "quorum",
approvedHash,
pollState,
};
}
}
return {
outcome: "fired",
reasonCodes: [],
policyVersion: String(row.policy_version),
mode: row.effective.mode,
approvedHash,
pollState: row.poll_state === null ? null : String(row.poll_state),
};
}
export class DispatchGateService {
constructor(private readonly pool: Pool) {}
/**
* E3 re-check + decision record for one destination. One round-trip: the
* CTE reads draft + live snapshot + latest poll, evaluates, and inserts the
* gate_decisions row (and its audit event) in the same transaction.
* Returns the outcome the dispatch path must obey.
*/
async check(args: {
workspaceId: string;
draftId: string;
idempotencyKey: string;
connectionId: string;
platform: string;
topicBindingId?: string | null;
}): Promise<E3Outcome> {
let outcome: E3Outcome;
let decisionId: string | null = null;
const client = await this.pool.connect();
try {
await client.query("BEGIN");
await client.query("SELECT set_config('app.workspace_id', $1, true)", [args.workspaceId]);
// NOTE: parameter numbering must be dense from $1 — PostgreSQL cannot
// type an unused placeholder, and the workspace id rides in set_config.
const result = await client.query<E3Row>(
`WITH attempt AS (
SELECT a.id AS attempt_id, a.draft_id, a.connection_id
FROM publishing_attempts a
WHERE a.idempotency_key = $1::text
), target AS (
SELECT d.state, d.approved_hash, d.approved_revision, d.reviewer_user_id,
s.effective, s.policy_version,
(SELECT p.state FROM polls p WHERE p.draft_id = d.id AND p.state <> 'cancelled'
AND p.subject_kind <> 'draft_publish'
ORDER BY p.created_at DESC LIMIT 1) AS poll_state,
(SELECT c.fields->>'publishGovernance' FROM connections c WHERE c.id = $2::uuid) AS publish_governance,
(SELECT p.state FROM polls p WHERE p.draft_id = d.id AND p.subject_kind = 'draft_publish'
AND p.publish_connection_id = $2::uuid AND p.draft_revision = d.approved_revision
ORDER BY p.created_at DESC LIMIT 1) AS publish_poll_state
FROM drafts d
LEFT JOIN posting_policy_snapshots s
ON s.workspace_id = d.workspace_id
AND s.account_id = $2::uuid
AND s.content_type = ''
AND s.topic_binding_id IS NOT DISTINCT FROM NULLIF($3::text, '')::uuid
WHERE d.id = (SELECT draft_id FROM attempt)
)
SELECT state, approved_hash, approved_revision, reviewer_user_id, effective, policy_version, poll_state,
publish_governance, publish_poll_state FROM target`,
[args.idempotencyKey, args.connectionId, args.topicBindingId ?? null],
);
const row = result.rows[0] as E3Row | undefined;
outcome = evaluateE3(row);
const inserted = await client.query<{ id: string }>(
`INSERT INTO gate_decisions
(workspace_id, draft_id, destination, gate, policy_version, outcome, inputs, reason_codes, actor)
SELECT d.workspace_id, d.id,
jsonb_build_object('platform', $1::text, 'accountId', $2::uuid,
'topicBindingId', $3::uuid, 'idempotencyKey', $4::text),
'E3', $5::text, $6::text,
jsonb_build_object('approvals', 1, 'approvedHash', d.approved_hash,
'policyMode', $7::text, 'pollState', $8::text),
$9::text[], 'system:gate-e3'
FROM drafts d WHERE d.id = $10::uuid
RETURNING id`,
[
args.platform,
args.connectionId,
args.topicBindingId ?? null,
args.idempotencyKey,
outcome.policyVersion,
outcome.outcome,
outcome.mode,
outcome.pollState ?? "none",
outcome.reasonCodes,
args.draftId,
],
);
decisionId = inserted.rows[0]?.id ?? null;
if (decisionId !== null) {
await client.query("SELECT audit_append($1::uuid, NULL, 'sp_scheduler', 'policy.gate', 'draft', $2::text, $3::jsonb)", [
args.workspaceId,
args.draftId,
JSON.stringify({
gate: "E3",
decisionId,
idempotencyKey: args.idempotencyKey,
destination: { platform: args.platform, accountId: args.connectionId },
outcome: outcome.outcome,
reasonCodes: outcome.reasonCodes,
policyVersion: outcome.policyVersion,
mode: outcome.mode,
pollState: outcome.pollState ?? "none",
}),
]);
}
await client.query("COMMIT");
} catch (error) {
await client.query("ROLLBACK").catch(() => undefined);
throw error;
} finally {
client.release();
}
return outcome;
}
}