Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/extension/src/run-checks.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import { fixtureGithub } from "@corbits/triage-workflows/evals/serve";
import { runChecks } from "./run-checks.ts";
import { outcomeOf, type ViewCheck } from "./view.ts";

const PR = JSON.parse(readFileSync(join(import.meta.dir, "../../../packages/pr-workflow-anthropic/fixtures/godotenv-262.json"), "utf8")) as Snapshot;
const PR = JSON.parse(readFileSync(join(import.meta.dir, "../fixtures/godotenv-262.json"), "utf8")) as Snapshot;

function readOnly(writes: string[]) {
const served = fixtureGithub(PR, {});
Expand Down
2 changes: 1 addition & 1 deletion apps/extension/src/run-checks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ import * as actions from "@corbits/triage-workflows/actions/index";
import type { RulesOutput } from "@corbits/triage-workflows/actions/rules";
import { credentialBindings } from "@corbits/triage-workflows/agents";
import { localActionCredentials } from "@corbits/triage-workflows/local-credentials";
import { workflow } from "@corbits/triage-workflows/pr-triage-github";
import { workflow } from "@corbits/triage-workflows/pr-triage";
import { runLocal, type ActionHandler, type OnTriggerPrimitive, type StepInvoker, type WorkflowAuthorizeFn, type WorkflowRun } from "@intx/workflow";
import { fileNotes, type FileNote } from "./notes.ts";
import type { PrRef } from "./pr.ts";
Expand Down
2 changes: 1 addition & 1 deletion apps/extension/src/session-reads.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import { fixtureGithub } from "@corbits/triage-workflows/evals/serve";
import { runChecks } from "./run-checks.ts";
import { sessionGithub, type Session } from "./session-reads.ts";

const PR = JSON.parse(readFileSync(join(import.meta.dir, "../../../packages/pr-workflow-anthropic/fixtures/godotenv-262.json"), "utf8")) as Snapshot;
const PR = JSON.parse(readFileSync(join(import.meta.dir, "../fixtures/godotenv-262.json"), "utf8")) as Snapshot;
const SHA = "c0ffee".padEnd(40, "0");
// Tag characters spell text no reader sees; only the commit's body, past its headline, carries them.
const HIDDEN = "\u{E0041}\u{E0042}\u{E0043}";
Expand Down
21 changes: 11 additions & 10 deletions apps/hub/src/audit-log.postgres.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,14 @@
import { expect, test } from "bun:test";
import { randomBytes } from "node:crypto";
import { sql } from "drizzle-orm";
import { createDB, runMigrations, schema } from "@intx/db";
import { createDB, runMigrations, schema, type DB } from "@intx/db";
import { createArtifact, runArtifactMigrations } from "@corbits/artifacts";
import { APPROVE_DO_KIND, AuditLog, checkPackName, emptyPack, stockTriggerMail } from "@corbits/triage-contracts";
import type { WorkflowRunEvent } from "@intx/hub-sessions";
import { createAuditLog } from "./audit-log.js";
import { createAuditLog, ROTATION_REASON } from "./audit-log.js";
import { loadCheckPack } from "./github/check-pack-store.js";
import { migrateTriageTables } from "./triage-db.js";
import { ROTATION_REASON } from "./github/deployment.js";
import { createVerdictRecordStore, type RecordedVerdict } from "./github/verdict-record-store.js";
import { type RecordedVerdict, replaceVerdict } from "./github/verdict-record-store.js";

const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
const TENANT = "tnt_audit";
Expand All @@ -35,14 +34,19 @@ const LOGS: Record<string, WorkflowRunEvent[]> = {
wfr_forged: approvalRun({ ...stockTriggerMail(APPROVAL), headers: { from: "mallory@elsewhere.test", to: ["pr-triage@acme.test"] } }),
};

function verdict(runId: string, time: string, state: string, comment: string): RecordedVerdict {
function verdict(runId: string, time: string, state: string, comment: string, headSha = "abcdef123"): RecordedVerdict {
return {
runId, repo: REPO, number: 7, headSha: "abcdef123", startedAt: at(time), verdictAt: at(time), settled: true, state, actor: "maintainer", humanGated: true,
runId, repo: REPO, number: 7, headSha, startedAt: at(time), verdictAt: at(time), settled: true, state, actor: "maintainer", humanGated: true,
degraded: null, checks: [], dos: [], commented: false, runStatus: "completed",
detail: { actions: [{ list: "fail", effectId: "eff_comment", kind: "comment", status: comment }, { list: "pass", effectId: "eff_approve", kind: "approve", held: true, status: "done" }] },
};
}

/** Seeds verdict rows as managers record them. */
async function seed(db: DB["db"], tenantId: string, verdicts: readonly RecordedVerdict[]): Promise<void> {
for (const verdict of verdicts) await replaceVerdict(db, tenantId, verdict);
}

function lines(log: AuditLog): string[] {
return log.entries.map(({ actor, action, target }) => `${actor} | ${action} | ${target}`);
}
Expand Down Expand Up @@ -81,7 +85,7 @@ test("PostgreSQL: the audit log merges verdicts, the Dos they ran, approvals, ch
{ id: "wfr_forged", anchorRunId: "wfr_anchor", definitionId: "wfd_triage", tenantId: TENANT, status: "completed", createdAt: at("10:04") },
{ id: "wfr_approval", anchorRunId: "wfr_anchor", definitionId: "wfd_triage", tenantId: TENANT, status: "completed", createdAt: at("10:05") },
]);
await createVerdictRecordStore(db).record(TENANT, [verdict("wfr_verdict", "10:00", "needs-review", "done"), verdict("wfr_retriage", "10:30", "ready", "failed")]);
await seed(db, TENANT, [verdict("wfr_verdict", "10:00", "needs-review", "done"), verdict("wfr_retriage", "10:30", "ready", "failed", "fedcba321")]);
await db.transaction(async function seedPack(tx) {
await createArtifact(tx, {
scope: { tenantId: TENANT, principalId: "prn_owner" }, ownerPrincipalId: "prn_owner", kind: "document",
Expand All @@ -104,9 +108,6 @@ test("PostgreSQL: the audit log merges verdicts, the Dos they ran, approvals, ch
return LOGS[runId] ?? [];
},
},
async sameSource() {
return true;
},
async authorize(_principalId, _tenantId, resource) {
return !denied.has(resource);
},
Expand Down
10 changes: 5 additions & 5 deletions apps/hub/src/audit-log.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ import { schema, type DB } from "@intx/db";
import { formatRunAddress } from "@intx/types";
import { WORKFLOW_RUN_REF, workflowRunRepoIdForAddress, type WorkflowRunReader } from "@intx/hub-sessions";
import type { CheckPackRead } from "./github/check-pack-store.js";
import { ROTATION_REASON } from "./github/deployment.js";
import { failure } from "./github/portal-credential.js";
import { repoRecords, sourceOf, triageNs } from "./github/tenant-config.js";
import { approvalIn, type Approval } from "./github/triage-runs.js";
Expand All @@ -36,6 +35,9 @@ const DO_OUTCOMES: Record<string, string> = { done: "Ran", satisfied: "Found don

type Entry = Omit<AuditEntry, "time"> & { at: Date };

/** The reason the hub gave when it replaced a busy deployment with a fresh copy, which earlier entries carry. */
export const ROTATION_REASON = "Replaced by a fresh deployment of the same workflow";

/** `people` is whether the caller may read the workspace's principals, which emails and members need. */
type Window = { tenantId: string; repo?: string; since?: Date; limit: number; people: boolean };

Expand All @@ -44,9 +46,8 @@ export type AuditLogDeps = {
artifacts: ArtifactDb;
readCheckPack: (tenantId: string, repo: string, source: RepoSource) => Promise<CheckPackRead>;
runReader: Pick<WorkflowRunReader, "readRunEvents">;
sameSource: (anchorRunId: string, otherAnchorRunId: string) => Promise<boolean>;
authorize: (principalId: string, tenantId: string, resource: string, action: string) => Promise<boolean>;
/** The pr-triage workflows whose approval runs are read, one per pull request host. */
/** The workflows whose approval runs are read. */
workflows: readonly string[];
};

Expand Down Expand Up @@ -227,8 +228,7 @@ export function createAuditLog(deps: AuditLogDeps) {
const previous = rows.slice(index + 1).find((older) => older.name === row.name);
const rotated = previous?.cancellationReason === ROTATION_REASON
&& previous.cancellationRequestedAt !== null
&& row.createdAt <= previous.cancellationRequestedAt
&& await deps.sameSource(previous.id, row.id);
&& row.createdAt <= previous.cancellationRequestedAt;
entries.push({ at: row.createdAt, actor: rotated ? HUB : row.principalId ?? UNKNOWN, action: "Deployed", target: row.name, detail: row.id });
if (row.endedAt === null) continue;
const byHub = row.status === "failed" || row.cancellationReason === ROTATION_REASON;
Expand Down
4 changes: 2 additions & 2 deletions apps/hub/src/bitbucket-datacenter/integration.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,13 @@ import { randomBytes } from "node:crypto";
import { type } from "arktype";
import { BITBUCKET_DATACENTER_HOOK_HEADERS, xHubHookMetadata } from "@corbits/common/hook-metadata";
import type { BitbucketFetch } from "@corbits/bitbucket-datacenter-reads";
import { TRIAGE_LISTENERS } from "@corbits/triage-contracts";
import { ensureDeployerRole } from "../deployer-role.js";
import { ensureTriageRole, grantEnabledKinds } from "../github/automation.js";
import { tenantConfigOf } from "../github/bridge.js";
import { activeCredential, credentialSecret, failure, holds, type PortalCredentialDeps } from "../github/portal-credential.js";
import { patchCorbitsTriage, repoRecords, triageNs, type CorbitsTriageNs, type RepoRecord } from "../github/tenant-config.js";
import type { Member } from "../integrations/app.js";
import { triageWorkflowFor } from "../github/source-workflow.js";
import { asCaller } from "../stock-caller.js";
import {
BitbucketFailed, bitbucketFetch, createWebhook, defaultBranch, deleteWebhook, InstanceRepo, lookupAddresses, pagedValues, readRepo,
Expand All @@ -22,7 +22,7 @@ export const BITBUCKET_PATH = "/api/integrations/bitbucket-datacenter";
const BITBUCKET_HOOK_PATH = "/api/hooks/bitbucket-datacenter";
const NO_STORE = { headers: { "Cache-Control": "no-store" } };
const HOST_REFUSED = "The host must resolve only to public addresses; a private-network instance needs the hub operator to allowlist it.";
const HOOK_METADATA = xHubHookMetadata(triageWorkflowFor("bitbucket-datacenter").id, BITBUCKET_DATACENTER_HOOK_HEADERS);
const HOOK_METADATA = xHubHookMetadata(TRIAGE_LISTENERS["bitbucket-datacenter"], BITBUCKET_DATACENTER_HOOK_HEADERS);

const NewConnection = type({ baseUrl: "string > 0", token: "string > 0" });
/** Connects, or replaces the token in place; `rotateHookSecret` instead signs the webhooks triage created with a new secret. */
Expand Down
68 changes: 6 additions & 62 deletions apps/hub/src/github/deployment.test.ts
Original file line number Diff line number Diff line change
@@ -1,40 +1,7 @@
import { describe, expect, test } from "bun:test";
import { pinOf, resolveLiveDeployments, sameSource } from "./deployment.js";
import { resolveLiveDeployments } from "./deployment.js";

const TENANT_ID = "tnt_acme";
const SOURCE = { kind: "asset", assetId: "ast_triage", package: { format: "source", commitSha: "c0ffee", packageName: "@corbits/pr-triage-workflow" } };
const SPEC = {
anchorRunId: "run_old",
deploymentDomain: "acme.test",
sourceAuthorityPrincipalId: "prn_deployer",
sourceOfferingIds: ["off_a", "off_b"],
defaultSourceOfferingId: "off_a",
deployContent: { systemPrompt: "Triage pull requests." },
frozenApprovalBundle: { source: SOURCE, entry: "./pr-triage.mjs" },
};

const TARBALL_SOURCE = { kind: "asset", assetId: "ast_corbits", package: { format: "tarball" } };

function closurePinning(...topLevel: { name: string; version: string }[]) {
const entries = topLevel.map(({ name, version }) => ({
name,
version,
source: { ...TARBALL_SOURCE, package: { format: "tarball", path: `tarballs/${name.slice(1).replace("/", "-")}-${version}.tgz`, integrity: "sha512-AAAA" } },
}));
return { schemaVersion: "1", topLevel, entries } as never;
}

function tarballSpec(version: string) {
return {
...SPEC,
frozenApprovalBundle: {
source: TARBALL_SOURCE,
entry: "./pr-triage.mjs",
closure: closurePinning({ name: "@corbits/pr-triage-workflow", version }),
},
};
}

function selectReturning(rows: unknown[]) {
const chain = {
from: () => chain,
Expand All @@ -47,38 +14,15 @@ function selectReturning(rows: unknown[]) {
}

describe("deployments", () => {
test("two deployments run the same source only when asset, commit and entry match", async () => {
const moved = { ...SPEC, frozenApprovalBundle: { ...SPEC.frozenApprovalBundle, source: { ...SOURCE, package: { ...SOURCE.package, commitSha: "beef" } } } };
const specs: Record<string, unknown> = { run_a: SPEC, run_b: SPEC, run_moved: moved };
const launchSpecs = { get: async (anchorRunId: string) => (specs[anchorRunId] ?? null) as never };
expect(await sameSource(launchSpecs, "run_a", "run_b")).toBe(true);
expect(await sameSource(launchSpecs, "run_a", "run_moved")).toBe(false);
expect(await sameSource(launchSpecs, "run_a", "run_missing")).toBe(false);
});

test("tarball-sourced deployments with different pins are not the same source", async () => {
const specs: Record<string, unknown> = { run_a: tarballSpec("0.1.0-a"), run_b: tarballSpec("0.1.0-a"), run_bumped: tarballSpec("0.1.0-b") };
const launchSpecs = { get: async (anchorRunId: string) => (specs[anchorRunId] ?? null) as never };
expect(await sameSource(launchSpecs, "run_a", "run_b")).toBe(true);
expect(await sameSource(launchSpecs, "run_a", "run_bumped")).toBe(false);
});

test("a deployment whose cancellation was requested is listed as cancelling", async () => {
const createdAt = new Date("2026-10-07T12:00:00.000Z");
const db = selectReturning([
{ runId: "run_new", workflow: "pr-triage", domain: "acme.test", createdAt, cancellationRequestedAt: null },
{ runId: "run_old", workflow: "pr-triage", domain: "acme.test", createdAt, cancellationRequestedAt: createdAt },
{ runId: "run_new", workflow: "triage-listener-github", domain: "acme.test", createdAt, cancellationRequestedAt: null },
{ runId: "run_old", workflow: "triage-listener-github", domain: "acme.test", createdAt, cancellationRequestedAt: createdAt },
]);
expect(await resolveLiveDeployments(db, TENANT_ID, "pr-triage")).toEqual([
{ runId: "run_new", address: "run_new@acme.test", createdAt, cancelling: false, workflow: "pr-triage" },
{ runId: "run_old", address: "run_old@acme.test", createdAt, cancelling: true, workflow: "pr-triage" },
expect(await resolveLiveDeployments(db, TENANT_ID, "triage-listener-github")).toEqual([
{ runId: "run_new", address: "run_new@acme.test", createdAt, cancelling: false, workflow: "triage-listener-github" },
{ runId: "run_old", address: "run_old@acme.test", createdAt, cancelling: true, workflow: "triage-listener-github" },
]);
});
});

describe("pinOf", () => {
test("pins the frozen closure's one top-level package, and refuses a closure without one", () => {
expect(pinOf(closurePinning({ name: "@corbits/pr-triage-workflow", version: "0.1.0-a" }))).toBe("@corbits/pr-triage-workflow@0.1.0-a");
expect(() => pinOf(closurePinning())).toThrow("frozen closure has 0 top-level packages, expected one");
});
});
53 changes: 6 additions & 47 deletions apps/hub/src/github/deployment.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { and, desc, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm";
import { schema, type DB, type ResolvedSenderPrincipal, type WorkflowRunLaunchSpecStore } from "@intx/db";
import { and, desc, eq, inArray, isNotNull, sql } from "drizzle-orm";
import { schema, type DB, type ResolvedSenderPrincipal } from "@intx/db";
import { MAIL_ADMISSION_DENIED_CODE } from "@intx/hub-sessions";
import type { RunTriggerMaterialize } from "@corbits/webhooks";
import type { ToolPackageManifest } from "@intx/types/tool-packages";
Expand Down Expand Up @@ -55,20 +55,7 @@ export type LiveDeployment = { runId: string; address: string; createdAt: Date;

export type TenantDeployment = LiveDeployment & { workflow: string };

/**
* A worker deployment carries this key in its deploy content, so the hub can tell it from the listener it was copied
* from: a worker ran one mail and was released when that run ended.
*/
const WORKER_MARKER = "corbitsTriageWorker";

export type DeploymentKind = "listener" | "worker";

function markedAs(kind: DeploymentKind) {
const marker = sql`${schema.workflowRunLaunchSpec.deployContent} ->> ${WORKER_MARKER}`;
return kind === "worker" ? eq(marker, "true") : isNull(marker);
}

type DeploymentQuery = { tenantId: string; definitionName?: string; kind: DeploymentKind };
type DeploymentQuery = { tenantId: string; definitionName?: string };

/**
* A workflow the portal publishes per person deploys from an asset named for its publisher and release, which names
Expand Down Expand Up @@ -97,7 +84,6 @@ async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<T
isNotNull(workflowRun.anchorRunId),
eq(workflowRun.id, workflowRun.anchorRunId),
inArray(workflowRun.status, [...liveWorkflowRunStatuses]),
markedAs(query.kind),
))
.orderBy(desc(workflowRun.createdAt), desc(workflowRun.id));
return rows.map((row) => ({
Expand All @@ -109,9 +95,9 @@ async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<T
}));
}

/** Live deployments; every workflow's when no name is given; listeners unless workers are asked for. */
export function resolveLiveDeployments(db: DB["db"], tenantId: string, definitionName?: string, kind: DeploymentKind = "listener"): Promise<TenantDeployment[]> {
return deploymentsWhere(db, { tenantId, definitionName, kind });
/** Live deployments; every workflow's when no name is given. */
export function resolveLiveDeployments(db: DB["db"], tenantId: string, definitionName?: string): Promise<TenantDeployment[]> {
return deploymentsWhere(db, { tenantId, definitionName });
}

export function newestLive(deployments: readonly LiveDeployment[]): LiveDeployment | undefined {
Expand All @@ -134,12 +120,6 @@ function rootOf(closure: ToolPackageManifest): PackageVersion {
return { name: root.name, version: root.version };
}

/** The launch spec does not store the pin, so it is read back from the frozen closure. */
export function pinOf(closure: ToolPackageManifest): string {
const { name, version } = rootOf(closure);
return `${name}@${version}`;
}

export type ClosureVersions = { workflow: PackageVersion; tools: PackageVersion[] };

/** A tarball-format source pins its one top-level package; a source-format one names its package, at the version it built. */
Expand All @@ -159,24 +139,3 @@ export function closureVersions(bundle: FrozenApprovalBundle): ClosureVersions |
.map((entry) => ({ name: entry.name, version: entry.version }));
return { workflow, tools };
}

/** A tarball-format source names its package only through the pin. */
function pinFor(bundle: FrozenApprovalBundle): string | undefined {
if (bundle.source.kind !== "asset" || bundle.source.package.format !== "tarball") return undefined;
return pinOf(bundle.closure);
}

/** The reason the hub gave when it replaced a busy deployment with a fresh copy, which earlier audit entries carry. */
export const ROTATION_REASON = "Replaced by a fresh deployment of the same workflow";

/** Whether two deployments were made from the same request: one source, package, pin and entry. */
export async function sameSource(launchSpecs: Pick<WorkflowRunLaunchSpecStore, "get">, anchorRunId: string, otherAnchorRunId: string): Promise<boolean> {
const [spec, other] = await Promise.all([launchSpecs.get(anchorRunId), launchSpecs.get(otherAnchorRunId)]);
if (!spec || !other) return false;
const a = spec.frozenApprovalBundle;
const b = other.frozenApprovalBundle;
if (a.source.kind !== "asset" || b.source.kind !== "asset" || a.entry !== b.entry) return false;
return a.source.assetId === b.source.assetId
&& JSON.stringify(a.source.package) === JSON.stringify(b.source.package)
&& pinFor(a) === pinFor(b);
}
Loading
Loading