Compare commits

..

9 commits

Author SHA1 Message Date
Scott Werner
9bd499cdbe
Merge pull request #817 from fabro-sh/codex/run-record-sql-foundation
Some checks are pending
Rust / Test (Linux) (push) Waiting to run
Rust / Test (macOS) (push) Waiting to run
Rust / Format (push) Waiting to run
Rust / Clippy (push) Waiting to run
Rust / Generated Docs (push) Waiting to run
TypeScript / Typecheck (push) Waiting to run
TypeScript / Test (push) Waiting to run
TypeScript / Build (push) Waiting to run
Add inactive SQL run event storage foundation
2026-08-27 16:30:05 -04:00
Scott Werner
57bbb923c2 Keep the RunSummaryStore name until the SQL cutover
Revert the run summary -> run record rename. The SQLite store is still
the summary read model today; it only grows an inactive events table
here. Renaming it now made the store file show as a delete plus add and
touched nine unrelated files. The final rename happens once, when the
SQL store becomes the run authority.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-27 15:43:56 -04:00
Scott Werner
ff2aef4564 Simplify SQL run record store write path and test fixtures
Share one bind helper across the runs insert/upsert/update statements,
compute the next event sequence once per append, and decode stored
sequence columns through a single helper. Check the run head before
decoding events, rewrite the first-visit stage listing as a UNION ALL so
each arm uses its partial index, and share the run_events insert SQL
with the test seeder.

Collapse the duplicated in-memory pool fixture, remove two tests that
only asserted Arc sharing, and fold the fabro-db test row helpers.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-27 14:59:34 -04:00
Scott Werner
dc55183468
Merge pull request #814 from fabro-sh/codex/automation-run-target
Migrate automations to canonical run targets
2026-08-27 13:50:37 -04:00
Scott Werner
b62e458289 Add inactive SQL run storage foundation 2026-08-27 13:32:01 -04:00
Bryan Helmkamp
039517a6a5
Merge pull request #816 from fabro-sh/reject-tool-call-index-gaps
Reject tool call index gaps in Chat Completions streams
2026-08-27 10:56:56 -04:00
Bryan Helmkamp
5ad2817da8
Reject tool call index gaps in Chat Completions streams
The openai_compatible stream decoder grew its tool call accumulator with
empty placeholder entries whenever a delta arrived with a sparse index,
then emitted every slot as a real tool call at finish. A provider that
numbers tool_calls[].index wrongly (Venice's Anthropic translation
passes through content-block positions, so a first tool call after text
arrives with index 1) therefore produced a phantom tool call with an
empty id and name. The phantom poisoned the conversation: the agent
answered it with a tool error, and the next request was rejected by the
provider (400: tool_use.id must match '^[a-zA-Z0-9_-]+$'), failing the
run as a non-retryable deterministic error.

A gap in the index sequence is indistinguishable from lost chunks, so
the decoder now fails the stream with a clear error naming the provider
and index instead of fabricating a tool call. Error::Stream is
classified retryable, so stage retries resample the turn rather than
replaying a poisoned history.

Observed on run 01M11JZVT7V507R56BCJJHZB1B; reproduced against the live
Venice API on claude-opus-5 and claude-sonnet-5 (four non-Claude models
stream index 0 correctly) and reported to Venice.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01C6JBmbpi2NeZXNEsftAhzd
2026-08-27 09:58:53 -04:00
Scott Werner
50fed849f3 Simplify automation run-target plumbing
- Materializer derives the manifest GitContext from RunTarget::validate()
  instead of hand-building it and re-parsing the repository slug
- Drop parse_github_repository_slug and InvalidRepositorySlug, now unused
- Store reuses Automation::git_target() instead of a private duplicate
- Legacy TOML import returns the target directly rather than a tuple
- Automation target migration updates columns with a single UPDATE ... FROM
- Web: share gitTarget(), targetFromFormValues(), and one SHA validator
  across the automation form, list, detail, new, and edit views

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-26 18:17:22 -04:00
Scott Werner
a65c4ff779 Migrate automations to canonical run targets 2026-08-26 17:35:36 -04:00
44 changed files with 2829 additions and 559 deletions

View file

@ -4,10 +4,16 @@ import type {
Automation,
AutomationTrigger,
Run,
RunProjection,
WorkflowSettings,
} from "@qltysh/fabro-api-client";
import { findApiTrigger, findScheduleTrigger } from "../lib/automation";
import {
findApiTrigger,
findScheduleTrigger,
gitTarget,
type GitRunTarget,
} from "../lib/automation";
import { Panel, Row } from "./settings-panel";
import { INPUT_CLASS } from "./ui";
import { sandboxRuntime } from "../lib/run-sandbox-lifecycle";
@ -17,7 +23,9 @@ export interface AutomationFormValues {
name: string;
description: string;
repository: string;
ref: string;
branch: string;
tag: string;
sha: string;
workflow: string;
manualEnabled: boolean;
scheduleEnabled: boolean;
@ -29,7 +37,9 @@ export const EMPTY_AUTOMATION_FORM: AutomationFormValues = {
name: "",
description: "",
repository: "",
ref: "main",
branch: "main",
tag: "",
sha: "",
workflow: "",
manualEnabled: true,
scheduleEnabled: false,
@ -46,13 +56,16 @@ const CRON_PRESETS: ReadonlyArray<{ label: string; value: string }> = [
export function automationToFormValues(automation: Automation): AutomationFormValues {
const apiTrigger = findApiTrigger(automation);
const scheduleTrigger = findScheduleTrigger(automation);
const target = gitTarget(automation.target);
return {
id: automation.id,
name: automation.name,
description: automation.description ?? "",
repository: automation.target.repository,
ref: automation.target.ref,
workflow: automation.target.workflow,
repository: target?.repo ?? "",
branch: target?.branch ?? EMPTY_AUTOMATION_FORM.branch,
tag: target?.tag ?? "",
sha: target?.sha ?? "",
workflow: automation.workflow,
manualEnabled: apiTrigger?.enabled ?? false,
scheduleEnabled: scheduleTrigger?.enabled ?? false,
cron: scheduleTrigger?.expression ?? "0 9 * * 1-5",
@ -61,6 +74,7 @@ export function automationToFormValues(automation: Automation): AutomationFormVa
export function automationFormValuesFromRun(
run: Run,
runState?: RunProjection | null,
settings?: WorkflowSettings | null,
): AutomationFormValues {
const name = firstPresentString(
@ -75,7 +89,9 @@ export function automationFormValuesFromRun(
run.workflow.graph_name,
name,
);
const repository = githubRepositoryFromSettings(settings)
const canonicalTarget = gitTarget(runState?.spec.target);
const repository = canonicalTarget?.repo
?? githubRepositoryFromSettings(settings)
?? githubRepositoryName(run.repository?.name)
?? githubRepositoryFromOriginUrl(run.repository?.origin_url)
?? "";
@ -85,7 +101,11 @@ export function automationFormValuesFromRun(
id: kebabify(name),
name,
repository,
ref: cloneBranch ?? EMPTY_AUTOMATION_FORM.ref,
branch: canonicalTarget?.branch
?? cloneBranch
?? EMPTY_AUTOMATION_FORM.branch,
tag: canonicalTarget?.tag ?? "",
sha: canonicalTarget?.sha ?? "",
workflow: run.workflow.slug?.trim() || kebabify(workflowName),
};
}
@ -111,11 +131,31 @@ export function isFormValid(values: AutomationFormValues): boolean {
values.id.trim() !== "" &&
values.name.trim() !== "" &&
values.repository.trim() !== "" &&
values.ref.trim() !== "" &&
values.branch.trim() !== "" &&
isOptionalShaValid(values.sha) &&
values.workflow.trim() !== ""
);
}
const GIT_SHA_RE = /^[0-9a-fA-F]{40}$/;
/** An empty SHA means "no pin"; anything else must be a full 40-hex commit id. */
function isOptionalShaValid(sha: string): boolean {
const trimmed = sha.trim();
return trimmed === "" || GIT_SHA_RE.test(trimmed);
}
/** Canonical Git target sent in create/replace requests. */
export function targetFromFormValues(values: AutomationFormValues): GitRunTarget {
return {
kind: "git",
repo: values.repository.trim(),
branch: values.branch.trim(),
tag: values.tag.trim() || undefined,
sha: values.sha.trim().toLowerCase() || undefined,
};
}
function kebabify(value: string): string {
return value
.toLowerCase()
@ -194,6 +234,7 @@ export function AutomationFormFields({
lockIdAndTarget = false,
}: AutomationFormFieldsProps) {
const slugTouchedRef = useRef(values.id.length > 0);
const shaValid = isOptionalShaValid(values.sha);
function patch(partial: Partial<AutomationFormValues>) {
onChange({ ...values, ...partial });
@ -277,19 +318,59 @@ export function AutomationFormFields({
className={`${INPUT_CLASS} font-mono`}
/>
</Row>
<Row title={<Label required>Branch</Label>} help="Default branch to run against.">
<Row
title={<Label required>Working branch</Label>}
help="Attached branch retained with the run, including when a tag or exact commit is selected."
>
<input
type="text"
name="branch"
aria-label="Default branch"
value={values.ref}
onChange={(e) => patch({ ref: e.target.value })}
aria-label="Working branch"
value={values.branch}
onChange={(e) => patch({ branch: e.target.value })}
placeholder="main"
autoComplete="off"
spellCheck={false}
className={`${INPUT_CLASS} font-mono`}
/>
</Row>
<Row
title={<Label optional>Tag</Label>}
help="Bare tag name resolved when the automation fires. Used only when exact SHA is empty."
>
<input
type="text"
name="tag"
aria-label="Tag"
value={values.tag}
onChange={(e) => patch({ tag: e.target.value })}
placeholder="v1.2.3"
autoComplete="off"
spellCheck={false}
className={`${INPUT_CLASS} font-mono`}
/>
</Row>
<Row
title={<Label optional>Exact SHA</Label>}
help={
shaValid
? "A 40-character commit SHA pins exact content and takes precedence over branch and tag."
: <span className="text-coral">Enter exactly 40 hexadecimal characters.</span>
}
>
<input
type="text"
name="sha"
aria-label="Exact commit SHA"
aria-invalid={!shaValid}
value={values.sha}
onChange={(e) => patch({ sha: e.target.value })}
placeholder="0123456789abcdef0123456789abcdef01234567"
autoComplete="off"
spellCheck={false}
className={`${INPUT_CLASS} font-mono`}
/>
</Row>
<Row
title={<Label required>Workflow slug</Label>}
help="Dash-separated identifier matching the workflow directory name (e.g. patch-cves)."

View file

@ -1,4 +1,13 @@
import type { Automation, AutomationTrigger } from "@qltysh/fabro-api-client";
import type { Automation, AutomationTrigger, RunTarget } from "@qltysh/fabro-api-client";
export type GitRunTarget = Extract<RunTarget, { kind: "git" }>;
/** Label shown in place of a repository when an automation's target is not Git-backed. */
export const UNSUPPORTED_TARGET_LABEL = "Unsupported target";
export function gitTarget(target: RunTarget | null | undefined): GitRunTarget | null {
return target?.kind === "git" ? target : null;
}
type TriggerOfType<K extends AutomationTrigger["type"]> = Extract<
AutomationTrigger,

View file

@ -18,7 +18,12 @@ import type {
import { toRunWithStatus } from "../data/runs";
import { ApiError, apiData, automationsApi } from "../lib/api-client";
import { findApiTrigger, findScheduleTrigger } from "../lib/automation";
import {
UNSUPPORTED_TARGET_LABEL,
findApiTrigger,
findScheduleTrigger,
gitTarget,
} from "../lib/automation";
import { useAutomation, useAutomationRuns } from "../lib/queries";
import { queryKeys } from "../lib/query-keys";
import { useDataUpdatedAt } from "../hooks/use-data-updated-at";
@ -93,6 +98,7 @@ function AutomationHeader({ automation }: { automation: Automation }) {
const scheduleTrigger = findScheduleTrigger(automation);
const apiTrigger = findApiTrigger(automation);
const target = gitTarget(automation.target);
const canRun = apiTrigger?.enabled === true;
async function onRun() {
@ -139,10 +145,16 @@ function AutomationHeader({ automation }: { automation: Automation }) {
</div>
<div className="mt-2 flex flex-wrap items-center gap-x-5 gap-y-2 text-sm">
<Chip icon={FolderIcon}>
{automation.target.repository}
<span className="text-fg-muted/70"> · {automation.target.ref}</span>
{target?.repo ?? UNSUPPORTED_TARGET_LABEL}
{target ? (
<span className="text-fg-muted/70">
{" · "}{target.branch}
{target.tag ? ` · ${target.tag}` : ""}
{target.sha ? ` · ${target.sha.slice(0, 8)}` : ""}
</span>
) : null}
</Chip>
<Chip icon={RectangleStackIcon}>{automation.target.workflow}</Chip>
<Chip icon={RectangleStackIcon}>{automation.workflow}</Chip>
{scheduleTrigger ? (
<Chip icon={ClockIcon}>{scheduleTrigger.expression}</Chip>
) : null}

View file

@ -11,6 +11,7 @@ import {
AutomationFormFields,
automationToFormValues,
isFormValid,
targetFromFormValues,
triggersFromFormValues,
type AutomationFormValues,
} from "../components/automation-form";
@ -85,11 +86,8 @@ function EditAutomationForm({ automation }: { automation: Automation }) {
automationsApi.replaceAutomation(automation.id, automation.revision, {
name: trimmedName,
description: values.description.trim() || null,
target: {
repository: values.repository.trim(),
ref: values.ref.trim(),
workflow: values.workflow.trim(),
},
target: targetFromFormValues(values),
workflow: values.workflow.trim(),
triggers: triggersFromFormValues(values),
}),
);

View file

@ -10,6 +10,8 @@ import { setupReactTestEnv } from "../lib/test-utils";
let currentRun: any = null;
let currentRunError: unknown = null;
let currentRunLoading = false;
let currentRunState: any = null;
let currentRunStateLoading = false;
let currentRunSettings: any = null;
const queryCalls: Array<{ hook: string; id: string | undefined }> = [];
const mountedRenderers: TestRenderer.ReactTestRenderer[] = [];
@ -58,6 +60,14 @@ mock.module("../lib/queries", () => ({
isLoading: false,
};
},
useRunState: (id: string | undefined) => {
queryCalls.push({ hook: "useRunState", id });
return {
data: currentRunState,
error: null,
isLoading: currentRunStateLoading,
};
},
}));
mock.module("../lib/api-client", () => ({
@ -253,6 +263,8 @@ beforeEach(() => {
currentRun = null;
currentRunError = null;
currentRunLoading = false;
currentRunState = null;
currentRunStateLoading = false;
currentRunSettings = null;
queryCalls.length = 0;
createAutomationMock.mockClear();
@ -274,7 +286,9 @@ describe("AutomationsNew", () => {
expect(fieldValue(renderer, "Automation name")).toBe("");
expect(fieldValue(renderer, "Automation slug")).toBe("");
expect(fieldValue(renderer, "Repository")).toBe("");
expect(fieldValue(renderer, "Default branch")).toBe("main");
expect(fieldValue(renderer, "Working branch")).toBe("main");
expect(fieldValue(renderer, "Tag")).toBe("");
expect(fieldValue(renderer, "Exact commit SHA")).toBe("");
expect(fieldValue(renderer, "Workflow slug")).toBe("");
expect(switchChecked(renderer, "Enable manual and API triggers")).toBe(true);
expect(switchChecked(renderer, "Enable scheduled triggers")).toBe(false);
@ -299,7 +313,9 @@ describe("AutomationsNew", () => {
expect(fieldValue(renderer, "Automation name")).toBe("Fix failing tests");
expect(fieldValue(renderer, "Automation slug")).toBe("fix-failing-tests");
expect(fieldValue(renderer, "Repository")).toBe("qltysh/fabro");
expect(fieldValue(renderer, "Default branch")).toBe("feature/from-run");
expect(fieldValue(renderer, "Working branch")).toBe("feature/from-run");
expect(fieldValue(renderer, "Tag")).toBe("");
expect(fieldValue(renderer, "Exact commit SHA")).toBe("");
expect(fieldValue(renderer, "Workflow slug")).toBe("fix-ci");
expect(switchChecked(renderer, "Enable manual and API triggers")).toBe(true);
expect(switchChecked(renderer, "Enable scheduled triggers")).toBe(false);
@ -307,9 +323,35 @@ describe("AutomationsNew", () => {
renderer.root.findAllByProps({ "aria-label": "Cron expression" }),
).toHaveLength(0);
expect(queryCalls).toContainEqual({ hook: "useRun", id: "run_1" });
expect(queryCalls).toContainEqual({ hook: "useRunState", id: "run_1" });
expect(queryCalls).toContainEqual({ hook: "useRunSettings", id: "run_1" });
});
test("canonical run target wins over legacy run, settings, and sandbox projections", async () => {
currentRun = makeRun();
currentRunSettings = makeRunSettings();
currentRunState = {
spec: {
target: {
kind: "git",
repo: "canonical/repo",
branch: "release",
tag: "v2.0.0",
sha: "0123456789abcdef0123456789abcdef01234567",
},
},
};
const { renderer } = await renderAutomationsNew("/automations/new?from_run=run_1");
expect(fieldValue(renderer, "Repository")).toBe("canonical/repo");
expect(fieldValue(renderer, "Working branch")).toBe("release");
expect(fieldValue(renderer, "Tag")).toBe("v2.0.0");
expect(fieldValue(renderer, "Exact commit SHA")).toBe(
"0123456789abcdef0123456789abcdef01234567",
);
});
test("automationFormValuesFromRun kebab-cases the workflow name fallback", () => {
const run = makeRun({
workflow: {
@ -334,7 +376,7 @@ describe("AutomationsNew", () => {
expect(textFromNode(renderer.toJSON())).toContain("fill it out manually");
expect(fieldValue(renderer, "Automation name")).toBe("");
expect(fieldValue(renderer, "Repository")).toBe("");
expect(fieldValue(renderer, "Default branch")).toBe("main");
expect(fieldValue(renderer, "Working branch")).toBe("main");
expect(fieldValue(renderer, "Workflow slug")).toBe("");
});
});

View file

@ -5,12 +5,13 @@ import { ChevronRightIcon } from "@heroicons/react/20/solid";
import { ApiError, apiData, automationsApi } from "../lib/api-client";
import { queryKeys } from "../lib/query-keys";
import { useRun, useRunSettings } from "../lib/queries";
import { useRun, useRunSettings, useRunState } from "../lib/queries";
import {
AutomationFormFields,
EMPTY_AUTOMATION_FORM,
automationFormValuesFromRun,
isFormValid,
targetFromFormValues,
triggersFromFormValues,
type AutomationFormValues,
} from "../components/automation-form";
@ -31,6 +32,7 @@ export default function AutomationsNew() {
const [searchParams] = useSearchParams();
const fromRunId = searchParams.get("from_run")?.trim() || undefined;
const runQuery = useRun(fromRunId);
const runStateQuery = useRunState(fromRunId);
const settingsQuery = useRunSettings(fromRunId);
if (!fromRunId) {
@ -45,8 +47,9 @@ export default function AutomationsNew() {
// Wait for both queries to settle before mounting the form, so the user's
// edits aren't blown away when settings arrive after the run.
const runPending = runQuery.isLoading && !runQuery.data;
const runStatePending = runStateQuery.isLoading && !runStateQuery.data;
const settingsPending = settingsQuery.isLoading && !settingsQuery.data;
if (runPending || settingsPending) {
if (runPending || runStatePending || settingsPending) {
return (
<div className="space-y-6">
<PageHeader />
@ -69,6 +72,7 @@ export default function AutomationsNew() {
const initialValues = automationFormValuesFromRun(
runQuery.data,
runStateQuery.data ?? null,
settingsQuery.data ?? null,
);
@ -108,11 +112,8 @@ function AutomationCreateForm({
id: values.id.trim(),
name: trimmedName,
description: values.description.trim() || null,
target: {
repository: values.repository.trim(),
ref: values.ref.trim(),
workflow: values.workflow.trim(),
},
target: targetFromFormValues(values),
workflow: values.workflow.trim(),
triggers: triggersFromFormValues(values),
}),
);

View file

@ -18,7 +18,12 @@ import { FilterButton } from "../components/runs-list/filter-button";
import type { Automation, AutomationListResponse } from "@qltysh/fabro-api-client";
import { Link, useNavigate } from "react-router";
import { ApiError, apiData, automationsApi } from "../lib/api-client";
import { findScheduleTrigger, hasEnabledApiTrigger } from "../lib/automation";
import {
UNSUPPORTED_TARGET_LABEL,
findScheduleTrigger,
gitTarget,
hasEnabledApiTrigger,
} from "../lib/automation";
import { useAutomations } from "../lib/queries";
import { queryKeys } from "../lib/query-keys";
import { ConfirmDialog, PRIMARY_BUTTON_CLASS } from "../components/ui";
@ -81,17 +86,20 @@ const MENU_ITEM_DANGER_CLASS =
function mapAutomations(result: AutomationListResponse | undefined): AutomationRow[] {
const automations = result?.data ?? [];
return automations.map((a) => ({
id: a.id,
revision: a.revision,
name: a.name,
workflow: a.target.workflow,
repository: a.target.repository,
schedule: findScheduleTrigger(a)?.expression,
apiEnabled: hasEnabledApiTrigger(a),
icon: slugIconMap[a.target.workflow] ?? CodeBracketIcon,
color: slugColorMap[a.target.workflow] ?? "var(--color-teal-500)",
}));
return automations.map((a) => {
const target = gitTarget(a.target);
return {
id: a.id,
revision: a.revision,
name: a.name,
workflow: a.workflow,
repository: target?.repo ?? UNSUPPORTED_TARGET_LABEL,
schedule: findScheduleTrigger(a)?.expression,
apiEnabled: hasEnabledApiTrigger(a),
icon: slugIconMap[a.workflow] ?? CodeBracketIcon,
color: slugColorMap[a.workflow] ?? "var(--color-teal-500)",
};
});
}
function PlayIcon({ className }: { className?: string }) {

View file

@ -6738,6 +6738,7 @@ components:
- name
- description
- target
- workflow
- triggers
properties:
id:
@ -6756,34 +6757,16 @@ components:
type: ["string", "null"]
example: Keeps dependencies fresh.
target:
$ref: "#/components/schemas/AutomationTarget"
$ref: "#/components/schemas/RunTarget"
workflow:
type: string
description: Workflow slug or path resolved in the selected repository checkout.
example: dependency-update
triggers:
type: array
items:
$ref: "#/components/schemas/AutomationTrigger"
AutomationTarget:
description: Repository and workflow selected by an automation.
type: object
additionalProperties: false
required:
- repository
- ref
- workflow
properties:
repository:
type: string
description: GitHub repository slug in `owner/repo` form.
example: fabro-sh/fabro
ref:
type: string
description: Branch, tag, or SHA selector resolved when materializing a run.
example: main
workflow:
type: string
description: Workflow slug or path resolved in the target repository.
example: dependency-update
AutomationTrigger:
description: |
Automation trigger configuration. Unknown `type` discriminator values
@ -6850,6 +6833,7 @@ components:
- id
- name
- target
- workflow
- triggers
properties:
id:
@ -6863,7 +6847,11 @@ components:
type: ["string", "null"]
example: Keeps dependencies fresh.
target:
$ref: "#/components/schemas/AutomationTarget"
$ref: "#/components/schemas/RunTarget"
workflow:
type: string
description: Workflow slug or path resolved in the selected repository checkout.
example: dependency-update
triggers:
type: array
items:
@ -6876,6 +6864,7 @@ components:
required:
- name
- target
- workflow
- triggers
properties:
name:
@ -6885,7 +6874,11 @@ components:
type: ["string", "null"]
example: Keeps dependencies fresh.
target:
$ref: "#/components/schemas/AutomationTarget"
$ref: "#/components/schemas/RunTarget"
workflow:
type: string
description: Workflow slug or path resolved in the selected repository checkout.
example: dependency-update
triggers:
type: array
items:

View file

@ -3,13 +3,37 @@ title: "Automations"
description: "Named, repeatable run configurations with API and schedule triggers"
---
An **automation** is a saved run configuration — a repository, ref, and workflow — plus the triggers that may start it. Every trigger fire creates and starts a normal Fabro run through the same pipeline as `POST /api/v1/runs`, so automation runs get the same lifecycle, events, and observability as manually created runs. Each run records the automation and trigger that created it.
An **automation** is a saved run configuration — a Git repository, working branch, optional tag or exact commit, and workflow — plus the triggers that may start it. Every trigger fire creates and starts a normal Fabro run through the same pipeline as `POST /api/v1/runs`, so automation runs get the same lifecycle, events, and observability as manually created runs. Each run records the automation and trigger that created it.
## Defining automations
The server stores automations in its SQLite database. Manage them in the web UI at `/automations` or through the `/api/v1/automations` REST API.
When upgrading from file-backed automation storage, startup imports every valid `automations/*.toml` file next to the active `settings.toml`. Existing SQLite definitions win on ID conflicts. After a successful import, Fabro renames the directory to a timestamped backup such as `automations.imported-20260711T180000000000Z.bak`. Invalid TOML leaves the original directory untouched for operator repair.
New definitions use Fabro's canonical Git run target. The working branch is always required. An optional tag selects that tag when no exact commit is present, and an optional 40-character commit SHA pins the run exactly. The exact commit wins when both a tag and SHA are present; the branch is retained as the run's working branch in every case.
```json title="Create automation request"
{
"name": "Nightly release",
"description": "Cut a nightly build from main",
"target": {
"kind": "git",
"repo": "fabro-sh/fabro",
"branch": "main",
"tag": "v1.2.3",
"sha": "0123456789abcdef0123456789abcdef01234567"
},
"workflow": "release",
"triggers": [
{ "type": "api", "id": "manual", "enabled": true }
]
}
```
Automations currently support Git targets only. Folder and empty run targets are rejected during validation.
### Upgrading legacy targets
When upgrading from file-backed automation storage, startup imports every valid `automations/*.toml` file next to the active `settings.toml`. Existing SQLite definitions win on ID conflicts. After a successful import, Fabro renames the directory to a timestamped backup such as `automations.imported-20260711T180000000000Z.bak`. Invalid TOML or an invalid target leaves the original directory untouched for operator repair.
The legacy files use this shape:
@ -34,7 +58,19 @@ enabled = true
expression = "0 0 * * *"
```
The target names a GitHub repository as an `owner/repo` slug, the ref to run against, and a project workflow defined in that repository. When a trigger fires, Fabro clones the repository at the ref, resolves the workflow, and creates and starts the run. Repositories are cached server-side as bare clones, so repeat fires fetch only what changed.
Fabro converts legacy refs deterministically:
- A 40-character hexadecimal SHA becomes an exact commit on working branch `main`.
- `refs/tags/<name>` and `tags/<name>` become a tag on working branch `main`.
- `refs/heads/<name>` and `heads/<name>` become a working branch.
- `HEAD` becomes working branch `main`.
- Any other bare value becomes a working branch.
The `main` default is only a migration assumption. If the repository uses another working branch, edit the imported automation before running it.
The same conversion runs transactionally for automations already in SQLite. An unsupported `refs/*` selector or an invalid branch or tag name aborts startup with an actionable error instead of guessing. The database remains on its previous schema and data, and the migration snapshot remains available. Edit the unsupported legacy `target_ref` to a branch, head selector, tag selector, `HEAD`, or exact SHA, then restart Fabro.
When a trigger fires, Fabro clones the repository at the selected branch, tag, or exact commit, resolves the workflow, and creates and starts the run. The created run records the exact checked-out commit in its canonical target, so later inspection and automation creation preserve the revision that actually ran. Repositories are cached server-side as bare clones, so repeat fires fetch only what changed.
## Triggers

View file

@ -125,12 +125,14 @@ pub(crate) async fn activate_blob_storage(
);
let blob_store = Arc::new(fabro_store::BlobStore::new(database.clone_pool()));
let run_summary_store = Arc::new(fabro_store::RunSummaryStore::new(database.clone_pool()));
let store = Arc::new(fabro_store::Database::new(
object_store,
slatedb_prefix,
flush_interval,
cache_path,
Arc::clone(&blob_store),
run_summary_store,
));
let inventory = store

View file

@ -1,14 +1,12 @@
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::Context as _;
use async_trait::async_trait;
use fabro_api::types::RunManifest;
use fabro_automation::{AutomationId, AutomationTarget};
use fabro_automation::AutomationId;
use fabro_config::{EnvironmentLayer, MergeMap};
use fabro_manifest::ManifestBuildInput;
use fabro_types::{DirtyStatus, GitContext, GitHubRepositorySlug, RunId};
use fabro_util::error::collect_chain;
use fabro_types::{GitHubRepositorySlug, GitRunTarget, RunId, RunTarget, TargetValidationError};
use tokio::{fs, task};
use crate::git_checkout::{
@ -18,7 +16,8 @@ use crate::git_checkout::{
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct AutomationRunMaterializeInput {
pub automation_id: AutomationId,
pub target: AutomationTarget,
pub target: GitRunTarget,
pub workflow: String,
pub run_id: RunId,
pub user_settings_path: PathBuf,
pub temp_root: PathBuf,
@ -28,28 +27,52 @@ pub(crate) struct AutomationRunMaterializeInput {
pub(crate) struct AutomationRunMaterialized {
pub manifest: RunManifest,
pub submitted_manifest_bytes: Vec<u8>,
pub target: GitRunTarget,
}
#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq)]
#[derive(thiserror::Error, Debug)]
pub(crate) enum RunMaterializeError {
#[error("invalid repository target: {0}")]
InvalidTarget(String),
#[error("failed to clone repository: {0}")]
CloneFailed(String),
#[error("failed to resolve workflow: {0}")]
WorkflowNotFound(String),
#[error("failed to build run manifest: {0}")]
Manifest(String),
#[error("failed to load GitHub credentials: {0}")]
Credentials(String),
}
impl From<GitCheckoutError> for RunMaterializeError {
fn from(value: GitCheckoutError) -> Self {
match value {
GitCheckoutError::CloneFailed(message) => Self::CloneFailed(message),
}
}
#[error("invalid automation Git target")]
InvalidTarget {
#[source]
source: TargetValidationError,
},
#[error("failed to prepare automation checkout")]
Checkout {
#[from]
source: GitCheckoutError,
},
#[error("failed to prepare automation temporary directory {path}")]
TempDirectory {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("failed to resolve automation workflow")]
WorkflowNotFound {
#[source]
source: anyhow::Error,
},
#[error("failed to build run manifest")]
Manifest {
#[source]
source: anyhow::Error,
},
#[error("manifest build task failed")]
ManifestTask {
#[source]
source: task::JoinError,
},
#[error("failed to serialize materialized run manifest")]
SerializeManifest {
#[source]
source: serde_json::Error,
},
#[error("failed to load GitHub credentials")]
Credentials {
#[source]
source: anyhow::Error,
},
}
#[async_trait]
@ -93,13 +116,17 @@ impl AutomationRunMaterializer for ProductionAutomationRunMaterializer {
&self,
input: AutomationRunMaterializeInput,
) -> Result<AutomationRunMaterialized, RunMaterializeError> {
let repo = parse_target_repository(&input.target.repository)?;
fs::create_dir_all(&input.temp_root).await.map_err(|err| {
RunMaterializeError::CloneFailed(format!(
"failed to create temp root {}: {err}",
input.temp_root.display()
))
})?;
let repo = GitHubRepositorySlug::try_new(&input.target.repo).ok_or(
RunMaterializeError::InvalidTarget {
source: TargetValidationError::Repository,
},
)?;
fs::create_dir_all(&input.temp_root)
.await
.map_err(|source| RunMaterializeError::TempDirectory {
path: input.temp_root.clone(),
source,
})?;
let temp_dir = tempfile::Builder::new()
.prefix(&format!(
"automation-{}-{}-",
@ -107,11 +134,9 @@ impl AutomationRunMaterializer for ProductionAutomationRunMaterializer {
input.run_id
))
.tempdir_in(&input.temp_root)
.map_err(|err| {
RunMaterializeError::CloneFailed(format!(
"failed to create per-run temp directory under {}: {err}",
input.temp_root.display()
))
.map_err(|source| RunMaterializeError::TempDirectory {
path: input.temp_root.clone(),
source,
})?;
let checkout_dir = temp_dir.path().join("repo");
let auth = resolve_git_auth_config(
@ -121,62 +146,43 @@ impl AutomationRunMaterializer for ProductionAutomationRunMaterializer {
self.http_client.clone(),
)
.await
.map_err(|err| RunMaterializeError::CloneFailed(render_error_chain(err.as_ref())))?;
.map_err(|source| RunMaterializeError::Credentials { source })?;
let checked_out_sha = self
.repo_cache
.prepare_worktree(WorktreePrepareInput {
repo: &repo,
ref_selector: &input.target.ref_selector,
target: &input.target,
auth: auth.as_ref(),
worktree_dir: &checkout_dir,
})
.await?;
let mut exact_target = input.target;
exact_target.sha = Some(checked_out_sha);
let manifest_input = ManifestFromCheckoutInput {
workflow: input.target.workflow,
workflow: input.workflow,
user_settings_path: input.user_settings_path,
checkout_dir,
git_context: ManifestGitContextInput {
repo,
ref_selector: input.target.ref_selector,
checked_out_sha,
},
target: exact_target,
environment_defaults: self.environment_defaults.clone(),
};
task::spawn_blocking(move || build_manifest_from_checkout(manifest_input))
.await
.map_err(|err| {
RunMaterializeError::Manifest(format!("manifest build task failed: {err}"))
})?
.map_err(|source| RunMaterializeError::ManifestTask { source })?
}
}
fn render_error_chain(error: &(dyn std::error::Error + 'static)) -> String {
collect_chain(error).join(": ")
}
fn parse_target_repository(value: &str) -> Result<GitHubRepositorySlug, RunMaterializeError> {
fabro_automation::parse_github_repository_slug(value)
.map_err(|err| RunMaterializeError::InvalidTarget(err.to_string()))
}
#[derive(Debug)]
pub(crate) struct ManifestFromCheckoutInput {
workflow: String,
user_settings_path: PathBuf,
checkout_dir: PathBuf,
git_context: ManifestGitContextInput,
target: GitRunTarget,
environment_defaults: MergeMap<EnvironmentLayer>,
}
#[derive(Debug)]
pub(crate) struct ManifestGitContextInput {
repo: GitHubRepositorySlug,
ref_selector: String,
checked_out_sha: String,
}
fn build_manifest_from_checkout(
args: ManifestFromCheckoutInput,
) -> Result<AutomationRunMaterialized, RunMaterializeError> {
@ -184,9 +190,17 @@ fn build_manifest_from_checkout(
workflow,
user_settings_path,
checkout_dir,
git_context,
target,
environment_defaults,
} = args;
// Re-validating the exact target (now carrying the checked-out SHA) yields
// the same `GitContext` projection the run-intent path uses.
let validated = RunTarget::Git(target)
.validate()
.map_err(|source| RunMaterializeError::InvalidTarget { source })?;
let RunTarget::Git(target) = validated.target else {
unreachable!("validating a Git target yields a Git target");
};
let built = fabro_manifest::build_run_manifest(ManifestBuildInput {
workflow: workflow.into(),
cwd: checkout_dir,
@ -194,33 +208,28 @@ fn build_manifest_from_checkout(
environment_defaults,
..ManifestBuildInput::default()
})
.map_err(|err| manifest_build_error(&err))?;
.map_err(manifest_build_error)?;
let mut manifest = built.manifest;
manifest.git = Some(GitContext {
origin_url: git_context.repo.https_url(),
branch: git_context.ref_selector,
sha: Some(git_context.checked_out_sha),
dirty: DirtyStatus::Clean,
});
manifest.git = validated.git;
let submitted_manifest_bytes = serde_json::to_vec(&manifest)
.context("failed to serialize materialized run manifest")
.map_err(|err| RunMaterializeError::Manifest(err.to_string()))?;
.map_err(|source| RunMaterializeError::SerializeManifest { source })?;
Ok(AutomationRunMaterialized {
manifest,
submitted_manifest_bytes,
target,
})
}
fn manifest_build_error(error: &anyhow::Error) -> RunMaterializeError {
fn manifest_build_error(error: anyhow::Error) -> RunMaterializeError {
if error.chain().any(|source| {
source
.downcast_ref::<fabro_config::Error>()
.is_some_and(|err| matches!(err, fabro_config::Error::WorkflowNotFound(_)))
}) {
RunMaterializeError::WorkflowNotFound(render_error_chain(error.as_ref()))
RunMaterializeError::WorkflowNotFound { source: error }
} else {
RunMaterializeError::Manifest(render_error_chain(error.as_ref()))
RunMaterializeError::Manifest { source: error }
}
}
@ -233,23 +242,28 @@ pub struct TestAutomationRunMaterializer {
#[cfg(any(test, feature = "test-support"))]
struct TestAutomationRunMaterializerState {
captured_inputs: Vec<AutomationRunMaterializeInput>,
response: Result<AutomationRunMaterialized, RunMaterializeError>,
response: Result<Box<AutomationRunMaterialized>, TargetValidationError>,
}
#[cfg(any(test, feature = "test-support"))]
impl TestAutomationRunMaterializer {
pub fn succeed(manifest: RunManifest, submitted_manifest_bytes: Vec<u8>) -> Self {
Self::new(Ok(AutomationRunMaterialized {
pub fn succeed(
manifest: RunManifest,
submitted_manifest_bytes: Vec<u8>,
target: GitRunTarget,
) -> Self {
Self::new(Ok(Box::new(AutomationRunMaterialized {
manifest,
submitted_manifest_bytes,
}))
target,
})))
}
pub fn fail_invalid_target(message: impl Into<String>) -> Self {
Self::new(Err(RunMaterializeError::InvalidTarget(message.into())))
pub fn fail_invalid_target() -> Self {
Self::new(Err(TargetValidationError::Repository))
}
fn new(response: Result<AutomationRunMaterialized, RunMaterializeError>) -> Self {
fn new(response: Result<Box<AutomationRunMaterialized>, TargetValidationError>) -> Self {
Self {
inner: std::sync::Arc::new(std::sync::Mutex::new(TestAutomationRunMaterializerState {
captured_inputs: Vec::new(),
@ -283,7 +297,11 @@ impl AutomationRunMaterializer for TestAutomationRunMaterializer {
.lock()
.expect("test automation materializer lock poisoned");
guard.captured_inputs.push(input);
guard.response.clone()
guard
.response
.clone()
.map(|materialized| *materialized)
.map_err(|source| RunMaterializeError::InvalidTarget { source })
}
}
@ -328,17 +346,17 @@ mod tests {
.unwrap();
let user_settings_path = temp.path().join("settings.toml");
fs::write(&user_settings_path, "_version = 1\n").unwrap();
let repo = parse_target_repository("workspace-org/app").unwrap();
let sha = "0123456789abcdef0123456789abcdef01234567".to_string();
let materialized = build_manifest_from_checkout(ManifestFromCheckoutInput {
workflow: "demo".to_string(),
user_settings_path: user_settings_path.clone(),
checkout_dir: checkout.clone(),
git_context: ManifestGitContextInput {
repo,
ref_selector: "release".to_string(),
checked_out_sha: sha.clone(),
target: GitRunTarget {
repo: "workspace-org/app".to_string(),
branch: "release".to_string(),
tag: Some("v1".to_string()),
sha: Some(sha.clone()),
},
environment_defaults: test_environment_defaults(),
})
@ -365,6 +383,8 @@ mod tests {
assert_eq!(git.branch, "release");
assert_eq!(git.sha.as_deref(), Some(sha.as_str()));
assert_eq!(git.dirty, DirtyStatus::Clean);
assert_eq!(materialized.target.tag.as_deref(), Some("v1"));
assert_eq!(materialized.target.sha.as_deref(), Some(sha.as_str()));
let submitted_manifest: serde_json::Value =
serde_json::from_slice(&materialized.submitted_manifest_bytes)
.expect("submitted bytes should be a manifest");

View file

@ -1,10 +1,11 @@
use std::borrow::Cow;
use std::path::{Path, PathBuf};
use std::time::Duration;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use fabro_store::KeyedMutex;
use fabro_types::GitHubRepositorySlug;
use fabro_types::{GitHubRepositorySlug, GitRunTarget};
use tokio::process::Command;
use tokio::{fs, time};
@ -15,10 +16,64 @@ const GIT_WORKTREE_PRUNE_TIMEOUT: Duration = Duration::from_secs(10);
const GIT_REV_PARSE_TIMEOUT: Duration = Duration::from_secs(10);
/// Error returned while preparing a checkout from a git source.
#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq)]
#[derive(thiserror::Error, Debug)]
pub(crate) enum GitCheckoutError {
#[error("failed to clone repository: {0}")]
CloneFailed(String),
#[error("failed to create Git cache directory {path}")]
CacheDirectory {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("failed to clone repository")]
Clone {
#[source]
source: GitCommandError,
},
#[error("failed to fetch branch {branch:?}")]
FetchBranch {
branch: String,
#[source]
source: GitCommandError,
},
#[error("failed to fetch tag {tag:?}")]
FetchTag {
tag: String,
#[source]
source: GitCommandError,
},
#[error("failed to fetch exact commit {sha}")]
FetchCommit {
sha: String,
#[source]
source: GitCommandError,
},
#[error("failed to resolve fetched Git target to a commit")]
ResolveCommit {
#[source]
source: GitCommandError,
},
#[error("failed to add Git worktree")]
AddWorktree {
#[source]
source: GitCommandError,
},
}
#[derive(thiserror::Error, Debug)]
pub(crate) enum GitCommandError {
#[error("{command} timed out after {timeout_secs}s")]
Timeout {
command: String,
timeout_secs: u64,
},
#[error("failed to run {command}")]
Spawn {
command: String,
#[source]
source: std::io::Error,
},
#[error("{message}")]
Exit { message: String },
}
/// Persistent on-disk cache of bare GitHub clones, one per `(owner, repo)`.
@ -113,25 +168,30 @@ impl GitRepoCache {
if !bare_exists {
if let Some(parent) = bare_dir.parent() {
fs::create_dir_all(parent).await.map_err(|err| {
GitCheckoutError::CloneFailed(format!(
"failed to create cache dir {}: {err}",
parent.display()
))
GitCheckoutError::CacheDirectory {
path: parent.to_path_buf(),
source: err,
}
})?;
}
run_git_plan(build_bare_clone_plan(clone_url, bare_dir, args.auth)).await?;
run_git_plan(build_bare_clone_plan(clone_url, bare_dir, args.auth))
.await
.map_err(|source| GitCheckoutError::Clone { source })?;
}
let fetch_target = GitFetchTarget::from(args.target);
run_git_plan(build_bare_fetch_plan(
bare_dir,
clone_url,
args.ref_selector,
&fetch_target.selector(),
args.auth,
))
.await?;
.await
.map_err(|source| fetch_target.checkout_error(source))?;
let checked_out_sha = run_git_plan(build_rev_parse_fetch_head_plan(bare_dir))
.await
.map_err(|source| GitCheckoutError::ResolveCommit { source })
.map(|stdout| String::from_utf8_lossy(&stdout).trim().to_string())?;
add_worktree_with_stale_retry(bare_dir, args.worktree_dir, &checked_out_sha).await?;
@ -142,11 +202,55 @@ impl GitRepoCache {
pub(crate) struct WorktreePrepareInput<'a> {
pub repo: &'a GitHubRepositorySlug,
pub ref_selector: &'a str,
pub target: &'a GitRunTarget,
pub auth: Option<&'a GitAuthConfig>,
pub worktree_dir: &'a Path,
}
enum GitFetchTarget<'a> {
Branch(&'a str),
Tag(&'a str),
Commit(&'a str),
}
impl<'a> From<&'a GitRunTarget> for GitFetchTarget<'a> {
fn from(target: &'a GitRunTarget) -> Self {
if let Some(sha) = target.sha.as_deref() {
Self::Commit(sha)
} else if let Some(tag) = target.tag.as_deref() {
Self::Tag(tag)
} else {
Self::Branch(&target.branch)
}
}
}
impl GitFetchTarget<'_> {
fn selector(&self) -> Cow<'_, str> {
match self {
Self::Branch(selector) | Self::Commit(selector) => Cow::Borrowed(selector),
Self::Tag(tag) => Cow::Owned(format!("refs/tags/{tag}")),
}
}
fn checkout_error(&self, source: GitCommandError) -> GitCheckoutError {
match self {
Self::Branch(branch) => GitCheckoutError::FetchBranch {
branch: (*branch).to_string(),
source,
},
Self::Tag(tag) => GitCheckoutError::FetchTag {
tag: (*tag).to_string(),
source,
},
Self::Commit(sha) => GitCheckoutError::FetchCommit {
sha: (*sha).to_string(),
source,
},
}
}
}
async fn bare_clone_may_be_corrupt(bare_dir: &Path) -> bool {
match fs::metadata(&bare_dir.join("HEAD")).await {
Ok(meta) => meta.len() == 0,
@ -348,7 +452,8 @@ fn build_worktree_prune_plan(bare_dir: &Path) -> GitCommandPlan {
}
fn build_rev_parse_fetch_head_plan(bare_dir: &Path) -> GitCommandPlan {
GitCommandPlan::new(["rev-parse", "FETCH_HEAD"], GIT_REV_PARSE_TIMEOUT).current_dir(bare_dir)
GitCommandPlan::new(["rev-parse", "FETCH_HEAD^{commit}"], GIT_REV_PARSE_TIMEOUT)
.current_dir(bare_dir)
}
async fn add_worktree_with_stale_retry(
@ -360,14 +465,14 @@ async fn add_worktree_with_stale_retry(
Ok(_) => Ok(()),
Err(first_err) => {
tracing::warn!(
%first_err,
error = ?first_err,
bare_dir = %bare_dir.display(),
worktree_dir = %worktree_dir.display(),
"git worktree add failed; pruning stale worktree entries and retrying"
);
if let Err(prune_err) = run_git_plan(build_worktree_prune_plan(bare_dir)).await {
tracing::warn!(
%prune_err,
error = ?prune_err,
bare_dir = %bare_dir.display(),
"failed to prune stale git worktree entries"
);
@ -375,11 +480,12 @@ async fn add_worktree_with_stale_retry(
run_git_plan(build_worktree_add_plan(bare_dir, worktree_dir, target))
.await
.map(|_| ())
.map_err(|source| GitCheckoutError::AddWorktree { source })
}
}
}
async fn run_git_plan(plan: GitCommandPlan) -> Result<Vec<u8>, GitCheckoutError> {
async fn run_git_plan(plan: GitCommandPlan) -> Result<Vec<u8>, GitCommandError> {
let mut command = Command::new(&plan.program);
command.args(&plan.args);
command.envs(plan.env.iter().map(|(key, value)| (key, value)));
@ -390,18 +496,13 @@ async fn run_git_plan(plan: GitCommandPlan) -> Result<Vec<u8>, GitCheckoutError>
let output = time::timeout(plan.timeout, command.output())
.await
.map_err(|_| {
GitCheckoutError::CloneFailed(format!(
"{} timed out after {}s",
safe_command_label(&plan),
plan.timeout.as_secs()
))
.map_err(|_| GitCommandError::Timeout {
command: safe_command_label(&plan),
timeout_secs: plan.timeout.as_secs(),
})?
.map_err(|err| {
GitCheckoutError::CloneFailed(format!(
"failed to run {}: {err}",
safe_command_label(&plan)
))
.map_err(|err| GitCommandError::Spawn {
command: safe_command_label(&plan),
source: err,
})?;
if output.status.success() {
@ -422,10 +523,9 @@ async fn run_git_plan(plan: GitCommandPlan) -> Result<Vec<u8>, GitCheckoutError>
message.push_str(": ");
message.push_str(stdout.trim());
}
Err(GitCheckoutError::CloneFailed(redact_git_output(
&message,
&plan.sensitive_values,
)))
Err(GitCommandError::Exit {
message: redact_git_output(&message, &plan.sensitive_values),
})
}
fn safe_command_label(plan: &GitCommandPlan) -> String {
@ -466,6 +566,15 @@ mod tests {
GitHubRepositorySlug::try_new(value).expect("slug should parse")
}
fn git_target(branch: &str, tag: Option<&str>, sha: Option<&str>) -> GitRunTarget {
GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: branch.to_string(),
tag: tag.map(str::to_string),
sha: sha.map(str::to_string),
}
}
#[test]
fn target_repository_urls_are_github_metadata_urls_without_credentials() {
let repo = repository_slug("fabro-sh/fabro");
@ -546,7 +655,7 @@ mod tests {
assert_eq!(prune.timeout, Duration::from_secs(10));
let rev_parse = build_rev_parse_fetch_head_plan(&bare_dir);
assert_eq!(rev_parse.args, vec!["rev-parse", "FETCH_HEAD"]);
assert_eq!(rev_parse.args, vec!["rev-parse", "FETCH_HEAD^{commit}"]);
assert_eq!(rev_parse.current_dir.as_deref(), Some(bare_dir.as_path()));
assert_eq!(rev_parse.timeout, Duration::from_secs(10));
}
@ -638,11 +747,16 @@ mod tests {
.args(["-C", work.to_str().unwrap(), "commit", "-m", "seed"])
.status()
.expect("git commit seed");
std::process::Command::new("git")
.args(["-C", work.to_str().unwrap(), "tag", "-a", "v1", "-m", "v1"])
.status()
.expect("git tag seed");
std::process::Command::new("git")
.args([
"-C",
work.to_str().unwrap(),
"push",
"--follow-tags",
upstream.to_str().unwrap(),
"main",
])
@ -689,13 +803,14 @@ mod tests {
let cache = GitRepoCache::new(temp.path().join("cache"));
let repo = repository_slug("fabro-sh/fabro");
let upstream_url = upstream.to_str().unwrap().to_string();
let target = git_target("main", None, None);
let worktree_a = temp.path().join("wt-a");
let sha_a = cache
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
ref_selector: "main",
target: &target,
auth: None,
worktree_dir: &worktree_a,
},
@ -717,7 +832,7 @@ mod tests {
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
ref_selector: "main",
target: &target,
auth: None,
worktree_dir: &worktree_b,
},
@ -741,13 +856,14 @@ mod tests {
let cache = GitRepoCache::new(temp.path().join("cache"));
let repo = repository_slug("fabro-sh/fabro");
let upstream_url = upstream.to_str().unwrap().to_string();
let target = git_target("main", None, None);
let worktree_a = temp.path().join("wt-a");
cache
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
ref_selector: "main",
target: &target,
auth: None,
worktree_dir: &worktree_a,
},
@ -765,7 +881,7 @@ mod tests {
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
ref_selector: "main",
target: &target,
auth: None,
worktree_dir: &worktree_b,
},
@ -781,4 +897,84 @@ mod tests {
.is_empty()
);
}
#[tokio::test]
async fn tag_and_exact_commit_modes_return_the_peeled_sha() {
let temp = TempDir::new().unwrap();
let upstream = temp.path().join("upstream.git");
let expected_sha = seed_upstream(&upstream);
let cache = GitRepoCache::new(temp.path().join("cache"));
let repo = repository_slug("fabro-sh/fabro");
let upstream_url = upstream.to_str().unwrap().to_string();
for (name, target) in [
("tag", git_target("main", Some("v1"), None)),
(
"pinned-tag",
git_target("main", Some("v1"), Some(&expected_sha)),
),
("commit", git_target("main", None, Some(&expected_sha))),
] {
let sha = cache
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
target: &target,
auth: None,
worktree_dir: &temp.path().join(name),
},
&upstream_url,
)
.await
.expect("target should materialize");
assert_eq!(sha, expected_sha, "{name}");
}
}
#[tokio::test]
async fn missing_tag_and_unavailable_commit_are_distinct_errors() {
let temp = TempDir::new().unwrap();
let upstream = temp.path().join("upstream.git");
seed_upstream(&upstream);
let cache = GitRepoCache::new(temp.path().join("cache"));
let repo = repository_slug("fabro-sh/fabro");
let upstream_url = upstream.to_str().unwrap().to_string();
let missing_tag = git_target("main", Some("missing"), None);
let unavailable_sha = "ffffffffffffffffffffffffffffffffffffffff";
let unavailable_commit = git_target("main", None, Some(unavailable_sha));
let tag_error = cache
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
target: &missing_tag,
auth: None,
worktree_dir: &temp.path().join("missing-tag"),
},
&upstream_url,
)
.await
.expect_err("missing tag should fail");
assert!(matches!(
tag_error,
GitCheckoutError::FetchTag { tag, .. } if tag == "missing"
));
let commit_error = cache
.prepare_worktree_with_clone_url(
WorktreePrepareInput {
repo: &repo,
target: &unavailable_commit,
auth: None,
worktree_dir: &temp.path().join("missing-commit"),
},
&upstream_url,
)
.await
.expect_err("unavailable commit should fail");
assert!(matches!(
commit_error,
GitCheckoutError::FetchCommit { sha, .. } if sha == unavailable_sha
));
}
}

View file

@ -1205,7 +1205,7 @@ impl AppState {
let credentials = self
.github_credentials(&settings.server.integrations.github)
.await
.map_err(|err| RunMaterializeError::Credentials(err.to_string()))?;
.map_err(|source| RunMaterializeError::Credentials { source })?;
ProductionAutomationRunMaterializer::new(
credentials,
self.github_api_base_url.clone(),
@ -2447,8 +2447,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
})
.context("load environments")?,
);
let run_summaries =
store.attach_run_summary_store(Arc::new(RunSummaryStore::new(db_pool.clone())));
let run_summaries = store.run_summary_store();
let auth_codes = Arc::new(AuthCodeStore::new(db_pool.clone()));
let auth_sessions = Arc::new(AuthSessionStore::new(db_pool.clone()));
let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);

View file

@ -8,7 +8,7 @@ use croner::errors::CronError;
use fabro_automation::{
Automation, AutomationId, AutomationRevision, AutomationTriggerId, parse_schedule_expression,
};
use fabro_types::{AutomationRef, Principal, RunId, SystemActorKind};
use fabro_types::{AutomationRef, Principal, RunId, RunTarget, SystemActorKind};
use tokio::time::sleep;
use tracing::{Instrument, error, info, info_span, warn};
@ -229,10 +229,18 @@ async fn fire_scheduled_automation_run(
) {
let automation_id = automation.id.clone();
let run_id = RunId::new();
let Some(target) = automation.git_target().cloned() else {
error!(
automation_id = %automation_id,
"Stored automation target is not Git-backed",
);
return;
};
let materialized = match state
.materialize_automation_run(AutomationRunMaterializeInput {
automation_id: automation_id.clone(),
target: automation.target.clone(),
target,
workflow: automation.workflow.clone(),
run_id,
user_settings_path: state.active_config_path().to_path_buf(),
temp_root: state.automation_temp_root(),
@ -243,7 +251,7 @@ async fn fire_scheduled_automation_run(
Err(err) => {
error!(
due_at = %due_at,
error = %err,
error = ?err,
"Failed to materialize scheduled automation run",
);
return;
@ -271,6 +279,7 @@ async fn fire_scheduled_automation_run(
actor: actor.clone(),
headers: HeaderMap::new(),
automation: Some(automation_ref),
target: Some(RunTarget::Git(materialized.target)),
},
))
.await;
@ -335,10 +344,10 @@ fn run_due_schedules_once<'a>(
#[cfg(test)]
mod tests {
use fabro_api::types::RunManifest;
use fabro_automation::{AutomationDraft, AutomationTarget, AutomationTrigger, ScheduleTrigger};
use fabro_automation::{AutomationDraft, AutomationTrigger, ScheduleTrigger};
use fabro_static::EnvVars;
use fabro_store::ListRunsQuery;
use fabro_types::RunStatus;
use fabro_types::{GitRunTarget, RunStatus};
use serde_json::json;
use super::*;
@ -350,14 +359,19 @@ mod tests {
.with_timezone(&Utc)
}
fn target() -> AutomationTarget {
AutomationTarget {
repository: "fabro-sh/fabro".to_string(),
ref_selector: "main".to_string(),
workflow: "workflow.fabro".to_string(),
fn git_target() -> GitRunTarget {
GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: None,
sha: None,
}
}
fn target() -> RunTarget {
RunTarget::Git(git_target())
}
fn schedule_trigger(id: &str, expression: &str, enabled: bool) -> AutomationTrigger {
AutomationTrigger::Schedule(ScheduleTrigger {
id: AutomationTriggerId::new(id).expect("test trigger id should be valid"),
@ -373,6 +387,7 @@ mod tests {
name: name.to_string(),
description: None,
target: target(),
workflow: "workflow.fabro".to_string(),
triggers,
}
}
@ -390,6 +405,7 @@ mod tests {
name: name.to_string(),
description: None,
target: target(),
workflow: "workflow.fabro".to_string(),
triggers,
})
.await
@ -422,7 +438,9 @@ mod tests {
let manifest = minimal_manifest();
let submitted_manifest_bytes =
serde_json::to_vec(&manifest).expect("manifest should serialize");
TestAutomationRunMaterializer::succeed(manifest, submitted_manifest_bytes)
let mut exact_target = git_target();
exact_target.sha = Some("0123456789abcdef0123456789abcdef01234567".to_string());
TestAutomationRunMaterializer::succeed(manifest, submitted_manifest_bytes, exact_target)
}
fn test_state_with_materializer(materializer: TestAutomationRunMaterializer) -> Arc<AppState> {
@ -696,7 +714,7 @@ mod tests {
#[tokio::test]
async fn failing_materializer_waits_until_next_cron_occurrence() {
let materializer = TestAutomationRunMaterializer::fail_invalid_target("boom");
let materializer = TestAutomationRunMaterializer::fail_invalid_target();
let state = test_state_with_materializer(materializer.clone());
create_automation(state.as_ref(), "nightly", "Nightly", vec![
schedule_trigger("schedule", "* * * * *", true),

View file

@ -6,7 +6,8 @@ use fabro_automation::{
Automation, AutomationDraft, AutomationId, AutomationReplace, AutomationStoreError,
};
use fabro_store::{RunSummaryListQuery, RunSummaryVisibility};
use fabro_types::{AutomationRef, RunId};
use fabro_types::{AutomationRef, RunId, RunTarget};
use fabro_util::error as error_util;
use serde::Serialize;
use super::super::{
@ -116,12 +117,20 @@ async fn create_automation_run(
.into_response();
};
let api_trigger_id = api_trigger.id.to_string();
let Some(target) = automation.git_target().cloned() else {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"Stored automation target is not Git-backed",
)
.into_response();
};
let run_id = RunId::new();
let materialized = match state
.materialize_automation_run(AutomationRunMaterializeInput {
automation_id: automation.id.clone(),
target: automation.target.clone(),
target,
workflow: automation.workflow.clone(),
run_id,
user_settings_path: state.active_config_path().to_path_buf(),
temp_root: state.automation_temp_root(),
@ -130,8 +139,8 @@ async fn create_automation_run(
{
Ok(materialized) => materialized,
Err(err) => {
return ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, err.to_string())
.into_response();
let message = error_util::collect_chain(&err).join(": ");
return ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, message).into_response();
}
};
let explicit_title_supplied = materialized.manifest.title.is_some();
@ -151,6 +160,7 @@ async fn create_automation_run(
actor: actor.clone(),
headers,
automation: Some(automation_ref),
target: Some(RunTarget::Git(materialized.target)),
},
))
.await;

View file

@ -562,6 +562,7 @@ async fn create_run(
actor,
headers,
automation: None,
target: None,
},
))
.await
@ -1142,6 +1143,9 @@ pub(crate) struct CreateRunFromManifestRequest {
pub(crate) actor: Principal,
pub(crate) headers: HeaderMap,
pub(crate) automation: Option<AutomationRef>,
/// Trusted canonical target supplied by an internal manifest producer.
/// Public legacy manifest requests always leave this absent.
pub(crate) target: Option<RunTarget>,
}
struct ManifestRunCompilerAdapter {
@ -1287,6 +1291,7 @@ pub(crate) async fn create_run_from_manifest(
actor,
headers,
automation,
target,
} = request;
let manifest_run_defaults = state.manifest_run_defaults();
let manifest_environment_defaults = state.environment_store().catalog_layer();
@ -1318,7 +1323,7 @@ pub(crate) async fn create_run_from_manifest(
storage_root: state.server_storage_dir(),
workflow_slug: None,
workflow_version_id: None,
target: None,
target,
provenance: run_provenance(&headers, &actor),
web_url: None,
submitted_manifest_bytes: Some(submitted_manifest_bytes),

View file

@ -11,7 +11,7 @@ use async_zip::base::read::mem::ZipFileReader;
use axum::body::Body;
use axum::http::{Method, Request, header};
use chrono::{Duration as ChronoDuration, SubsecRound as _, Utc};
use fabro_automation::{AutomationId, AutomationTarget};
use fabro_automation::AutomationId;
use fabro_config::bind::Bind;
use fabro_config::{
EnvironmentLayer, MergeMap, RunLayer, ServerSettingsBuilder, WorkflowSettingsBuilder,
@ -27,8 +27,8 @@ use fabro_types::settings::ServerAuthMethod;
use fabro_types::settings::run::EnvironmentProvider;
use fabro_types::{
AgentBackend, AttrValue, AuthMethod, BlobHash, CommandTermination, FailureCategory,
FailureDetail, Graph, InterviewQuestionRecord, Node, Outcome, ParallelBranchId, QuestionType,
RunId, RunSpec, SandboxProviderKind, StageContextWindowBreakdownItem,
FailureDetail, GitRunTarget, Graph, InterviewQuestionRecord, Node, Outcome, ParallelBranchId,
QuestionType, RunId, RunSpec, RunTarget, SandboxProviderKind, StageContextWindowBreakdownItem,
StageContextWindowCategory, StageContextWindowCountMethod, StageContextWindowProjection,
StageContextWindowStaleness, StageContextWindowWarning, StageModelUsage, StageTiming,
SuccessReason, SystemActorKind, WorkflowSettings, fixtures, test_support,
@ -4422,6 +4422,7 @@ async fn create_run_from_manifest_helper_persists_without_automation_metadata()
},
headers: HeaderMap::new(),
automation: None,
target: None,
},
))
.await;
@ -4437,10 +4438,12 @@ async fn create_run_from_manifest_helper_persists_without_automation_metadata()
.unwrap()
.unwrap();
assert!(summary.automation.is_none());
let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
assert!(run_store.state().await.unwrap().spec.target.is_none());
}
#[tokio::test]
async fn create_run_from_manifest_helper_persists_automation_metadata() {
async fn create_run_from_manifest_helper_persists_automation_metadata_and_exact_target() {
let state = TestAppStateBuilder::new()
.env_lookup(|_| None)
.vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
@ -4453,6 +4456,12 @@ async fn create_run_from_manifest_helper_persists_automation_metadata() {
name: Some("Nightly".to_string()),
trigger_id: Some("schedule".to_string()),
};
let target = RunTarget::Git(GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: Some("v1.2.3".to_string()),
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
});
let response = Box::pin(handler::runs::create_run_from_manifest(
Arc::clone(&state),
@ -4466,6 +4475,7 @@ async fn create_run_from_manifest_helper_persists_automation_metadata() {
},
headers: HeaderMap::new(),
automation: Some(automation.clone()),
target: Some(target.clone()),
},
))
.await;
@ -4488,6 +4498,8 @@ async fn create_run_from_manifest_helper_persists_automation_metadata() {
.unwrap()
.unwrap();
assert_eq!(summary.automation, Some(automation));
let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
assert_eq!(run_store.state().await.unwrap().spec.target, Some(target));
}
#[tokio::test]
@ -4565,6 +4577,7 @@ layer = "project"
},
headers,
automation: None,
target: None,
},
))
.await;
@ -4585,6 +4598,10 @@ layer = "project"
);
let run_state = run_store.state().await.unwrap();
let spec = &run_state.spec;
assert!(
spec.target.is_none(),
"legacy manifest GitContext must not become canonical target authority"
);
assert_eq!(spec.run_id, run_id);
assert_eq!(spec.graph.goal(), "Inline release goal");
assert_eq!(
@ -4723,6 +4740,7 @@ async fn create_run_from_manifest_pins_compiler_http_error_mappings() {
},
headers: HeaderMap::new(),
automation: None,
target: None,
},
))
.await;
@ -4777,6 +4795,7 @@ async fn create_run_from_manifest_preserves_competing_preparation_error_preceden
},
headers: HeaderMap::new(),
automation: None,
target: None,
},
))
.await;
@ -4822,6 +4841,7 @@ async fn create_run_from_manifest_resolves_generated_id_after_variable_snapshot(
},
headers: HeaderMap::new(),
automation: None,
target: None,
},
))
.await;
@ -4840,6 +4860,12 @@ async fn fake_automation_materializer_injection_captures_input_and_returns_manif
let fake = TestAutomationRunMaterializer::succeed(
materialized_manifest.clone(),
b"{\"fake\":true}".to_vec(),
GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: None,
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
},
);
let state = TestAppStateBuilder::new()
.automation_materializer(fake.clone())
@ -4847,16 +4873,18 @@ async fn fake_automation_materializer_injection_captures_input_and_returns_manif
let run_id = RunId::new();
let user_settings_path = PathBuf::from("/tmp/fabro/settings.toml");
let temp_root = PathBuf::from("/tmp/fabro/automation");
let target = AutomationTarget {
repository: "fabro-sh/fabro".to_string(),
ref_selector: "main".to_string(),
workflow: "demo".to_string(),
let target = GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: None,
sha: None,
};
let output = state
.materialize_automation_run(AutomationRunMaterializeInput {
automation_id: AutomationId::new("nightly").unwrap(),
target: target.clone(),
workflow: "demo".to_string(),
run_id,
user_settings_path: user_settings_path.clone(),
temp_root: temp_root.clone(),
@ -4873,6 +4901,7 @@ async fn fake_automation_materializer_injection_captures_input_and_returns_manif
assert_eq!(captured.len(), 1);
assert_eq!(captured[0].automation_id.as_str(), "nightly");
assert_eq!(captured[0].target, target);
assert_eq!(captured[0].workflow, "demo");
assert_eq!(captured[0].run_id, run_id);
assert_eq!(captured[0].user_settings_path, user_settings_path);
assert_eq!(captured[0].temp_root, temp_root);

View file

@ -8,6 +8,7 @@ use fabro_server::test_support::{
TestAppStateBuilder, TestAutomationRunMaterializer, build_test_router, test_auth_mode,
};
use fabro_static::EnvVars;
use fabro_types::GitRunTarget;
use serde_json::{Value, json};
use sqlx::Row as _;
use tower::ServiceExt;
@ -23,10 +24,11 @@ fn automation_body(id: &str, name: &str) -> Value {
"name": name,
"description": "Runs on a schedule.",
"target": {
"repository": "fabro-sh/fabro",
"ref": "main",
"workflow": "release"
"kind": "git",
"repo": "fabro-sh/fabro",
"branch": "main"
},
"workflow": "release",
"triggers": [
{
"type": "api",
@ -48,10 +50,11 @@ fn replacement_body(name: &str) -> Value {
"name": name,
"description": null,
"target": {
"repository": "fabro-sh/fabro",
"ref": "main",
"workflow": "release"
"kind": "git",
"repo": "fabro-sh/fabro",
"branch": "main"
},
"workflow": "release",
"triggers": [
{
"type": "api",
@ -91,6 +94,12 @@ fn automation_app_with_fake_materializer() -> (axum::Router, tempfile::TempDir,
.automation_materializer(TestAutomationRunMaterializer::succeed(
materialized_manifest,
submitted_manifest_bytes,
GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: None,
sha: Some("0123456789abcdef0123456789abcdef01234567".to_string()),
},
))
.build();
(build_test_router(state), temp_dir, sqlite_path)

View file

@ -6,10 +6,15 @@ use std::path::{Path, PathBuf};
use chrono::Utc;
use fabro_db::{DbPool, ImportReport};
use fabro_types::{GitRunTarget, RunTarget, repository};
use serde::Deserialize;
use tokio::fs;
use tracing::info;
use crate::{Automation, AutomationId, AutomationStoreError, store};
use crate::{
Automation, AutomationId, AutomationReplace, AutomationRevision, AutomationStoreError,
AutomationTrigger, store,
};
pub(crate) const REMOVAL_DEADLINE: &str = "2026-10-11";
@ -26,7 +31,7 @@ pub async fn import_legacy_directory_once(
let bytes = fs::read(&path)
.await
.map_err(|source| AutomationStoreError::io(&path, source))?;
automations.push(Automation::from_persisted_path(id, &bytes, path)?);
automations.push(parse_legacy_automation(id, &bytes, &path)?);
}
let mut transaction = pool.begin().await?;
@ -61,6 +66,88 @@ pub async fn import_legacy_directory_once(
Ok(Some(report))
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyPersistedAutomation {
name: String,
#[serde(default)]
description: Option<String>,
target: LegacyAutomationTarget,
#[serde(default)]
triggers: Vec<AutomationTrigger>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyAutomationTarget {
repository: String,
#[serde(rename = "ref")]
selector: String,
workflow: String,
}
fn parse_legacy_automation(
id: AutomationId,
bytes: &[u8],
path: &Path,
) -> Result<Automation, AutomationStoreError> {
let revision = AutomationRevision::from_bytes(bytes);
let content = std::str::from_utf8(bytes)
.map_err(|source| AutomationStoreError::invalid_utf8(path, source))?;
let legacy: LegacyPersistedAutomation =
toml::from_str(content).map_err(|source| AutomationStoreError::parse(path, source))?;
let LegacyAutomationTarget {
repository,
selector,
workflow,
} = legacy.target;
let target = legacy_target(repository, &selector, path)?;
Automation::from_stored(id.clone(), revision, AutomationReplace {
name: legacy.name,
description: legacy.description,
target,
workflow,
triggers: legacy.triggers,
})
.map_err(|source| AutomationStoreError::StoredValidation { id, source })
}
fn legacy_target(
repository: String,
selector: &str,
path: &Path,
) -> Result<RunTarget, AutomationStoreError> {
let (branch, tag, sha) = if let Some(sha) = repository::normalize_git_commit_sha(selector) {
("main".to_string(), None, Some(sha))
} else if let Some(tag) = selector
.strip_prefix("refs/tags/")
.or_else(|| selector.strip_prefix("tags/"))
{
("main".to_string(), Some(tag.to_string()), None)
} else if let Some(branch) = selector
.strip_prefix("refs/heads/")
.or_else(|| selector.strip_prefix("heads/"))
{
(branch.to_string(), None, None)
} else if selector == "HEAD" {
("main".to_string(), None, None)
} else {
(selector.to_string(), None, None)
};
RunTarget::Git(GitRunTarget {
repo: repository,
branch,
tag,
sha,
})
.validate()
.map(|validated| validated.target)
.map_err(|source| AutomationStoreError::LegacyTarget {
path: path.to_path_buf(),
source,
})
}
async fn legacy_automation_paths(
source_dir: &Path,
) -> Result<Option<Vec<(AutomationId, PathBuf)>>, AutomationStoreError> {

View file

@ -1,6 +1,7 @@
use std::path::PathBuf;
use croner::errors::CronError;
use fabro_types::TargetValidationError;
use toml::de::Error as TomlDeError;
use toml::ser::Error as TomlSerError;
@ -14,10 +15,13 @@ pub enum AutomationValidationError {
InvalidAutomationTriggerId { value: String },
#[error("automation name must not be empty")]
EmptyName,
#[error("repository slug {value:?} must be a GitHub owner/repo slug")]
InvalidRepositorySlug { value: String },
#[error("git ref selector {value:?} is not safe")]
InvalidGitRefSelector { value: String },
#[error("automation target kind {kind:?} is not supported; only Git targets are accepted")]
UnsupportedTarget { kind: String },
#[error("automation Git target is invalid")]
InvalidTarget {
#[source]
source: TargetValidationError,
},
#[error("workflow selector {value:?} is not safe")]
InvalidWorkflowSelector { value: String },
#[error("duplicate automation trigger id {id:?}")]
@ -114,6 +118,14 @@ pub enum AutomationStoreError {
#[source]
source: std::io::Error,
},
#[error(
"legacy automation target at {path:?} cannot be migrated; edit target.ref to a branch, supported heads/tags selector, HEAD, or 40-hex SHA and restart"
)]
LegacyTarget {
path: PathBuf,
#[source]
source: TargetValidationError,
},
}
impl AutomationStoreError {
@ -156,6 +168,7 @@ impl AutomationStoreError {
Self::Serialize { .. } => "serialize",
Self::Io { .. } => "io",
Self::LegacyBackup { .. } => "legacy_backup",
Self::LegacyTarget { .. } => "legacy_target",
}
}
}

View file

@ -9,7 +9,7 @@ pub use fabro_types::GitHubRepositorySlug;
pub use id::{AutomationId, AutomationRevision, AutomationRevisionParseError, AutomationTriggerId};
pub use migrations::{ImportReport, import_legacy_directory_once};
pub use model::{
ApiTrigger, Automation, AutomationDraft, AutomationReplace, AutomationTarget,
AutomationTrigger, ScheduleTrigger, parse_github_repository_slug, parse_schedule_expression,
ApiTrigger, Automation, AutomationDraft, AutomationReplace, AutomationTrigger, ScheduleTrigger,
parse_schedule_expression,
};
pub use store::AutomationStore;

View file

@ -4,7 +4,7 @@ use std::sync::LazyLock;
use croner::Cron;
use croner::errors::CronError;
use croner::parser::{CronParser, Seconds, Year};
use fabro_types::{GitHubRepositorySlug, repository};
use fabro_types::{GitRunTarget, RunTarget};
use serde::{Deserialize, Serialize};
use crate::{
@ -38,7 +38,8 @@ pub struct Automation {
pub revision: AutomationRevision,
pub name: String,
pub description: Option<String>,
pub target: AutomationTarget,
pub target: RunTarget,
pub workflow: String,
pub triggers: Vec<AutomationTrigger>,
}
@ -49,17 +50,6 @@ impl Automation {
Self::from_persisted(id, revision, persisted).map_err(AutomationStoreError::from)
}
pub(crate) fn from_persisted_path(
id: AutomationId,
bytes: &[u8],
path: impl Into<std::path::PathBuf>,
) -> Result<Self, AutomationStoreError> {
let path = path.into();
let revision = AutomationRevision::from_bytes(bytes);
let persisted = parse_persisted(bytes, Some(path))?;
Self::from_persisted(id, revision, persisted).map_err(AutomationStoreError::from)
}
pub(crate) fn from_replace(
id: AutomationId,
draft: AutomationReplace,
@ -112,6 +102,15 @@ impl Automation {
self.enabled_api_trigger().is_some()
}
/// Returns the validated Git target owned by this automation.
#[must_use]
pub fn git_target(&self) -> Option<&GitRunTarget> {
match &self.target {
RunTarget::Git(target) => Some(target),
RunTarget::None {} | RunTarget::Folder { .. } => None,
}
}
fn from_persisted(
id: AutomationId,
revision: AutomationRevision,
@ -132,20 +131,12 @@ impl Automation {
name: replace.name,
description: replace.description,
target: replace.target,
workflow: replace.workflow,
triggers: replace.triggers,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AutomationTarget {
pub repository: String,
#[serde(rename = "ref")]
pub ref_selector: String,
pub workflow: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
pub enum AutomationTrigger {
@ -205,7 +196,8 @@ pub struct AutomationDraft {
pub name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
pub target: AutomationTarget,
pub target: RunTarget,
pub workflow: String,
pub triggers: Vec<AutomationTrigger>,
}
@ -215,6 +207,7 @@ impl From<AutomationDraft> for (AutomationId, AutomationReplace) {
name: value.name,
description: value.description,
target: value.target,
workflow: value.workflow,
triggers: value.triggers,
})
}
@ -226,7 +219,8 @@ pub struct AutomationReplace {
pub name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
pub target: AutomationTarget,
pub target: RunTarget,
pub workflow: String,
pub triggers: Vec<AutomationTrigger>,
}
@ -236,7 +230,8 @@ pub(crate) struct PersistedAutomation {
name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
description: Option<String>,
target: AutomationTarget,
target: RunTarget,
workflow: String,
#[serde(default)]
triggers: Vec<AutomationTrigger>,
}
@ -247,6 +242,7 @@ impl From<AutomationReplace> for PersistedAutomation {
name: value.name,
description: value.description,
target: value.target,
workflow: value.workflow,
triggers: value.triggers,
}
}
@ -258,6 +254,7 @@ impl From<PersistedAutomation> for AutomationReplace {
name: value.name,
description: value.description,
target: value.target,
workflow: value.workflow,
triggers: value.triggers,
}
}
@ -288,15 +285,14 @@ fn validate_fields(value: &AutomationReplace) -> Result<(), AutomationValidation
if value.name.trim().is_empty() {
return Err(AutomationValidationError::EmptyName);
}
validate_repository_slug(&value.target.repository)?;
validate_git_ref_selector(&value.target.ref_selector)?;
validate_workflow_selector(&value.target.workflow)?;
validate_workflow_selector(&value.workflow)?;
validate_triggers(&value.triggers)
}
fn normalize_replace(
mut value: AutomationReplace,
) -> Result<AutomationReplace, AutomationValidationError> {
value.target = validate_target(value.target)?;
validate_fields(&value)?;
let api_enabled = value
@ -334,28 +330,16 @@ fn normalize_replace(
Ok(value)
}
pub fn parse_github_repository_slug(
value: &str,
) -> Result<GitHubRepositorySlug, AutomationValidationError> {
GitHubRepositorySlug::try_new(value).ok_or_else(|| {
AutomationValidationError::InvalidRepositorySlug {
value: value.to_string(),
}
})
}
fn validate_repository_slug(value: &str) -> Result<(), AutomationValidationError> {
parse_github_repository_slug(value).map(|_| ())
}
fn validate_git_ref_selector(value: &str) -> Result<(), AutomationValidationError> {
if repository::is_valid_github_ref_selector(value) {
Ok(())
} else {
Err(AutomationValidationError::InvalidGitRefSelector {
value: value.to_string(),
})
fn validate_target(target: RunTarget) -> Result<RunTarget, AutomationValidationError> {
if !matches!(&target, RunTarget::Git(_)) {
return Err(AutomationValidationError::UnsupportedTarget {
kind: target.kind_name().to_string(),
});
}
target
.validate()
.map(|validated| validated.target)
.map_err(|source| AutomationValidationError::InvalidTarget { source })
}
fn validate_workflow_selector(value: &str) -> Result<(), AutomationValidationError> {
@ -419,17 +403,20 @@ fn validate_triggers(triggers: &[AutomationTrigger]) -> Result<(), AutomationVal
#[cfg(test)]
mod tests {
use fabro_types::{GitRunTarget, RunTarget, TargetValidationError};
use crate::{
ApiTrigger, Automation, AutomationId, AutomationReplace, AutomationTarget,
AutomationTrigger, AutomationTriggerId, AutomationValidationError, ScheduleTrigger,
ApiTrigger, Automation, AutomationId, AutomationReplace, AutomationTrigger,
AutomationTriggerId, AutomationValidationError, ScheduleTrigger,
};
fn target() -> AutomationTarget {
AutomationTarget {
repository: "fabro-sh/fabro".to_string(),
ref_selector: "main".to_string(),
workflow: ".fabro/workflows/test/workflow.toml".to_string(),
}
fn target() -> RunTarget {
RunTarget::Git(GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: None,
sha: None,
})
}
fn api_trigger(id: &str) -> AutomationTrigger {
@ -455,11 +442,12 @@ mod tests {
fn persisted_toml_applies_defaults_and_canonicalizes_without_id_or_revision() {
let bytes = br#"
name = "Nightly"
workflow = "release"
[target]
repository = "fabro-sh/fabro"
ref = "main"
workflow = "release"
kind = "git"
repo = "fabro-sh/fabro"
branch = "main"
[[triggers]]
type = "api"
@ -492,6 +480,7 @@ expression = "0 0 * * *"
let bytes = br#"
name = "Legacy"
enabled = false
workflow = "release"
[target]
repository = "fabro-sh/fabro"
@ -516,6 +505,7 @@ enabled = true
name: "Nightly".to_string(),
description: None,
target: target(),
workflow: ".fabro/workflows/test/workflow.toml".to_string(),
triggers: vec![
api_trigger("manual"),
schedule_trigger_with_enabled("nightly", "0 0 * * *", true),
@ -533,41 +523,29 @@ enabled = true
}
#[test]
fn repository_slug_parser_returns_the_shared_type() {
let slug: fabro_types::GitHubRepositorySlug =
crate::parse_github_repository_slug("owner/.github").unwrap();
fn invalid_git_target_preserves_the_shared_validation_error() {
let error = super::validate_target(RunTarget::Git(GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main;rm".to_string(),
tag: None,
sha: None,
}))
.unwrap_err();
assert_eq!(slug.owner(), "owner");
assert_eq!(slug.repo(), ".github");
assert!(matches!(&error, AutomationValidationError::InvalidTarget {
source: TargetValidationError::Branch,
}));
assert_eq!(error.to_string(), "automation Git target is invalid");
}
#[test]
fn invalid_repository_slug_preserves_the_automation_error() {
let error = crate::parse_github_repository_slug("not/github/slug").unwrap_err();
fn non_git_targets_are_rejected_with_their_kind() {
let error = super::validate_target(RunTarget::None {}).unwrap_err();
assert!(matches!(
&error,
AutomationValidationError::InvalidRepositorySlug { value }
if value == "not/github/slug"
error,
AutomationValidationError::UnsupportedTarget { kind } if kind == "none"
));
assert_eq!(
error.to_string(),
"repository slug \"not/github/slug\" must be a GitHub owner/repo slug"
);
}
#[test]
fn invalid_git_ref_selector_preserves_the_automation_error() {
let error = super::validate_git_ref_selector("main;rm").unwrap_err();
assert!(matches!(
&error,
AutomationValidationError::InvalidGitRefSelector { value } if value == "main;rm"
));
assert_eq!(
error.to_string(),
"git ref selector \"main;rm\" is not safe"
);
}
#[test]
@ -577,42 +555,45 @@ enabled = true
name: " ".to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![api_trigger("manual")],
},
AutomationReplace {
name: "Bad repo".to_string(),
description: None,
target: AutomationTarget {
repository: "not/github/slug".to_string(),
ref_selector: "main".to_string(),
workflow: "release".to_string(),
},
target: RunTarget::Git(GitRunTarget {
repo: "not/github/slug".to_string(),
branch: "main".to_string(),
tag: None,
sha: None,
}),
workflow: "release".to_string(),
triggers: vec![api_trigger("manual")],
},
AutomationReplace {
name: "Bad ref".to_string(),
description: None,
target: AutomationTarget {
repository: "fabro-sh/fabro".to_string(),
ref_selector: "main;rm".to_string(),
workflow: "release".to_string(),
},
target: RunTarget::Git(GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main;rm".to_string(),
tag: None,
sha: None,
}),
workflow: "release".to_string(),
triggers: vec![api_trigger("manual")],
},
AutomationReplace {
name: "Bad workflow".to_string(),
description: None,
target: AutomationTarget {
repository: "fabro-sh/fabro".to_string(),
ref_selector: "main".to_string(),
workflow: "../release".to_string(),
},
target: target(),
workflow: "../release".to_string(),
triggers: vec![api_trigger("manual")],
},
AutomationReplace {
name: "Duplicate trigger".to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![
api_trigger("manual"),
schedule_trigger("manual", "0 0 * * *"),
@ -622,18 +603,21 @@ enabled = true
name: "Two API triggers".to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![api_trigger("one"), api_trigger("two")],
},
AutomationReplace {
name: "Six field cron".to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![schedule_trigger("nightly", "0 0 0 * * *")],
},
AutomationReplace {
name: "Bad cron".to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![schedule_trigger("nightly", "99 0 * * *")],
},
];

View file

@ -1,13 +1,13 @@
use std::str::FromStr as _;
use fabro_db::DbPool;
use fabro_types::{GitRunTarget, RunTarget};
use sqlx::sqlite::SqliteRow;
use sqlx::{Row as _, Sqlite, Transaction};
use crate::{
ApiTrigger, Automation, AutomationDraft, AutomationId, AutomationReplace, AutomationRevision,
AutomationStoreError, AutomationTarget, AutomationTrigger, AutomationTriggerId,
ScheduleTrigger,
AutomationStoreError, AutomationTrigger, AutomationTriggerId, ScheduleTrigger,
};
/// Shared projection for loading automations with their schedule triggers.
@ -22,7 +22,9 @@ macro_rules! select_automations_sql {
a.description,
a.api_enabled,
a.target_repository,
a.target_ref,
a.target_branch,
a.target_tag,
a.target_sha,
a.target_workflow,
t.id AS trigger_id,
t.enabled AS trigger_enabled,
@ -93,6 +95,7 @@ impl AutomationStore {
draft: AutomationReplace,
) -> Result<Automation, AutomationStoreError> {
let (automation, _) = Automation::from_replace(id.clone(), draft)?;
let target = stored_git_target(&automation);
let mut transaction = self.pool.begin().await?;
let result = sqlx::query(
r"
@ -102,7 +105,9 @@ impl AutomationStore {
description = ?,
api_enabled = ?,
target_repository = ?,
target_ref = ?,
target_branch = ?,
target_tag = ?,
target_sha = ?,
target_workflow = ?
WHERE id = ? AND revision = ?
",
@ -111,9 +116,11 @@ impl AutomationStore {
.bind(&automation.name)
.bind(automation.description.as_deref())
.bind(automation.api_enabled())
.bind(&automation.target.repository)
.bind(&automation.target.ref_selector)
.bind(&automation.target.workflow)
.bind(&target.repo)
.bind(&target.branch)
.bind(target.tag.as_deref())
.bind(target.sha.as_deref())
.bind(&automation.workflow)
.bind(id.as_str())
.bind(expected.as_str())
.execute(&mut *transaction)
@ -156,7 +163,8 @@ struct StoredAutomation {
name: String,
description: Option<String>,
api_enabled: bool,
target: AutomationTarget,
target: RunTarget,
workflow: String,
schedule_triggers: Vec<ScheduleTrigger>,
}
@ -180,11 +188,13 @@ impl StoredAutomation {
name: row.try_get("name")?,
description: row.try_get("description")?,
api_enabled: row.try_get("api_enabled")?,
target: AutomationTarget {
repository: row.try_get("target_repository")?,
ref_selector: row.try_get("target_ref")?,
workflow: row.try_get("target_workflow")?,
},
target: RunTarget::Git(GitRunTarget {
repo: row.try_get("target_repository")?,
branch: row.try_get("target_branch")?,
tag: row.try_get("target_tag")?,
sha: row.try_get("target_sha")?,
}),
workflow: row.try_get("target_workflow")?,
schedule_triggers: Vec::new(),
})
}
@ -230,6 +240,7 @@ impl StoredAutomation {
name: self.name,
description: self.description,
target: self.target,
workflow: self.workflow,
triggers,
})
.map_err(|source| AutomationStoreError::StoredValidation { id, source })
@ -272,6 +283,7 @@ pub(crate) async fn insert_automation_ignoring_conflict(
transaction: &mut Transaction<'_, Sqlite>,
automation: &Automation,
) -> Result<bool, AutomationStoreError> {
let target = stored_git_target(automation);
let result = sqlx::query(
r"
INSERT INTO automations (
@ -281,9 +293,11 @@ pub(crate) async fn insert_automation_ignoring_conflict(
description,
api_enabled,
target_repository,
target_ref,
target_branch,
target_tag,
target_sha,
target_workflow
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO NOTHING
",
)
@ -292,9 +306,11 @@ pub(crate) async fn insert_automation_ignoring_conflict(
.bind(&automation.name)
.bind(automation.description.as_deref())
.bind(automation.api_enabled())
.bind(&automation.target.repository)
.bind(&automation.target.ref_selector)
.bind(&automation.target.workflow)
.bind(&target.repo)
.bind(&target.branch)
.bind(target.tag.as_deref())
.bind(target.sha.as_deref())
.bind(&automation.workflow)
.execute(&mut **transaction)
.await?;
if result.rows_affected() == 0 {
@ -304,6 +320,12 @@ pub(crate) async fn insert_automation_ignoring_conflict(
Ok(true)
}
fn stored_git_target(automation: &Automation) -> &GitRunTarget {
automation
.git_target()
.expect("stored automations have already passed Git-only validation")
}
async fn insert_schedule_triggers(
transaction: &mut Transaction<'_, Sqlite>,
automation: &Automation,

View file

@ -6,11 +6,11 @@
use std::path::Path;
use fabro_automation::{
ApiTrigger, AutomationDraft, AutomationId, AutomationReplace, AutomationStore,
AutomationStoreError, AutomationTarget, AutomationTrigger, AutomationTriggerId,
ScheduleTrigger,
ApiTrigger, AutomationDraft, AutomationId, AutomationReplace, AutomationRevision,
AutomationStore, AutomationStoreError, AutomationTrigger, AutomationTriggerId, ScheduleTrigger,
};
use fabro_db::Database;
use fabro_types::{GitRunTarget, RunTarget};
use tokio::fs;
async fn test_database() -> (tempfile::TempDir, Database) {
@ -22,12 +22,13 @@ async fn test_database() -> (tempfile::TempDir, Database) {
(dir, database)
}
fn target() -> AutomationTarget {
AutomationTarget {
repository: "fabro-sh/fabro".to_string(),
ref_selector: "main".to_string(),
workflow: "release".to_string(),
}
fn target() -> RunTarget {
RunTarget::Git(GitRunTarget {
repo: "fabro-sh/fabro".to_string(),
branch: "main".to_string(),
tag: None,
sha: None,
})
}
fn schedule(id: &str, expression: &str, enabled: bool) -> AutomationTrigger {
@ -44,6 +45,7 @@ fn draft(id: &str, api_enabled: bool) -> AutomationDraft {
name: "Nightly".to_string(),
description: Some("Runs every night".to_string()),
target: target(),
workflow: "release".to_string(),
triggers: vec![
schedule("z-last", "0 2 * * *", false),
AutomationTrigger::Api(ApiTrigger {
@ -60,6 +62,7 @@ fn replacement(name: &str, expression: &str) -> AutomationReplace {
name: name.to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![
schedule("nightly", expression, true),
AutomationTrigger::Api(ApiTrigger {
@ -218,6 +221,7 @@ async fn failed_schedule_insert_rolls_back_parent_replace() {
name: "Should roll back".to_string(),
description: None,
target: target(),
workflow: "release".to_string(),
triggers: vec![schedule("blocked", "0 7 * * *", true)],
};
@ -254,7 +258,8 @@ async fn legacy_import_is_transactional_and_sql_wins() {
let source_dir = dir.path().join("automations");
fs::create_dir_all(&source_dir).await.unwrap();
write_legacy_automation(&source_dir, "existing", "Legacy existing").await;
write_legacy_automation(&source_dir, "imported", "Imported").await;
let imported_bytes = write_legacy_automation(&source_dir, "imported", "Imported").await;
let expected_revision = AutomationRevision::from_bytes(&imported_bytes);
fs::write(source_dir.join("notes.txt"), "ignored")
.await
.unwrap();
@ -278,15 +283,23 @@ async fn legacy_import_is_transactional_and_sql_wins() {
.name,
"Nightly"
);
assert_eq!(
store
.get(&AutomationId::new("imported").unwrap())
.await
.unwrap()
.unwrap()
.name,
"Imported"
);
let imported = store
.get(&AutomationId::new("imported").unwrap())
.await
.unwrap()
.unwrap();
assert_eq!(imported.name, "Imported");
assert_eq!(imported.revision, expected_revision);
assert_eq!(imported.workflow, "release");
assert!(matches!(
imported.target,
RunTarget::Git(GitRunTarget {
branch,
tag: None,
sha: None,
..
}) if branch == "main"
));
fs::create_dir_all(&source_dir).await.unwrap();
write_legacy_automation(&source_dir, "existing", "Legacy existing").await;
@ -340,15 +353,47 @@ async fn invalid_legacy_file_leaves_directory_and_database_unchanged() {
);
}
async fn write_legacy_automation(dir: &Path, id: &str, name: &str) {
fs::write(
dir.join(format!("{id}.toml")),
format!(
r#"name = "{name}"
#[tokio::test]
async fn unsupported_legacy_target_leaves_directory_and_database_unchanged() {
let (dir, database) = test_database().await;
let source_dir = dir.path().join("automations");
fs::create_dir_all(&source_dir).await.unwrap();
let bytes = legacy_automation_bytes("Unsupported", "refs/pull/123/head");
fs::write(source_dir.join("unsupported.toml"), bytes)
.await
.unwrap();
let err = fabro_automation::import_legacy_directory_once(database.pool(), &source_dir)
.await
.unwrap_err();
assert!(matches!(err, AutomationStoreError::LegacyTarget { .. }));
assert!(err.to_string().contains("edit target.ref"));
assert!(source_dir.exists());
assert!(
AutomationStore::new(database.clone_pool())
.list()
.await
.unwrap()
.is_empty()
);
}
async fn write_legacy_automation(dir: &Path, id: &str, name: &str) -> Vec<u8> {
let bytes = legacy_automation_bytes(name, "main");
fs::write(dir.join(format!("{id}.toml")), &bytes)
.await
.unwrap();
bytes
}
fn legacy_automation_bytes(name: &str, ref_selector: &str) -> Vec<u8> {
format!(
r#"name = "{name}"
[target]
repository = "fabro-sh/fabro"
ref = "main"
ref = "{ref_selector}"
workflow = "release"
[[triggers]]
@ -362,8 +407,6 @@ type = "schedule"
enabled = true
expression = "0 3 * * *"
"#
),
)
.await
.unwrap();
.into_bytes()
}

View file

@ -56,7 +56,7 @@ impl StreamState {
}
/// Process a parsed SSE chunk and return events to emit, if any.
fn process_chunk(&mut self, mut chunk: StreamChunk) -> Option<Vec<StreamEvent>> {
fn process_chunk(&mut self, mut chunk: StreamChunk) -> Result<Option<Vec<StreamEvent>>, Error> {
// Capture response metadata from the first chunk.
if let Some(id) = &chunk.id {
if self.response_id.is_empty() {
@ -80,8 +80,12 @@ impl StreamState {
.or_else(|| chunk.cost.as_ref().and_then(|cost| cost.usd));
self.cost_usd = cost_usd.or(self.cost_usd);
let choices = chunk.choices.as_mut()?;
let choice = choices.first_mut()?;
let Some(choices) = chunk.choices.as_mut() else {
return Ok(None);
};
let Some(choice) = choices.first_mut() else {
return Ok(None);
};
let mut events = Vec::new();
@ -90,7 +94,9 @@ impl StreamState {
self.finish_reason = map_finish_reason(Some(reason.as_str()));
}
let delta = choice.delta.as_mut()?;
let Some(delta) = choice.delta.as_mut() else {
return Ok(None);
};
// Accumulate reasoning/thinking content (Kimi, etc.).
if let Some(reasoning) = delta.reasoning() {
@ -121,8 +127,22 @@ impl StreamState {
for tc in tool_calls {
let index = tc.index;
// Grow the accumulated tool calls vector if needed.
while self.tool_calls.len() <= index {
// A delta may only continue an already-started tool call or
// open the next slot. Padding a skipped slot would materialize
// a phantom tool call with an empty id and name, which poisons
// the conversation once echoed back to the provider.
if index > self.tool_calls.len() {
return Err(Error::Stream {
message: format!(
"malformed tool call stream from {}: delta for tool_calls[{index}] \
arrived before tool_calls[{}] was started",
self.provider_name,
self.tool_calls.len()
),
source: None,
});
}
if index == self.tool_calls.len() {
self.tool_calls.push(AccumulatedToolCall {
id: String::new(),
name: String::new(),
@ -163,9 +183,9 @@ impl StreamState {
}
if events.is_empty() {
None
Ok(None)
} else {
Some(events)
Ok(Some(events))
}
}
@ -262,7 +282,7 @@ impl StreamDecoder for StreamState {
let chunk: StreamChunk = serde_json::from_str(ev.data)
.map_err(|e| Error::stream_error(format!("failed to parse SSE chunk: {e}"), e))?;
Ok(self.process_chunk(chunk).unwrap_or_default())
Ok(self.process_chunk(chunk)?.unwrap_or_default())
}
fn finish(&mut self) -> Vec<StreamEvent> {
@ -373,7 +393,7 @@ mod tests {
let chunk1: StreamChunk = serde_json::from_str(
r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":"Hello"},"finish_reason":null}]}"#,
).unwrap();
let events1 = state.process_chunk(chunk1).unwrap();
let events1 = state.process_chunk(chunk1).unwrap().unwrap();
assert_eq!(events1.len(), 2);
assert!(matches!(events1[0], StreamEvent::TextStart { .. }));
assert!(matches!(events1[1], StreamEvent::TextDelta { .. }));
@ -381,7 +401,7 @@ mod tests {
let chunk2: StreamChunk = serde_json::from_str(
r#"{"id":"c1","model":"m1","choices":[{"delta":{"content":" world"},"finish_reason":null}]}"#,
).unwrap();
let events2 = state.process_chunk(chunk2).unwrap();
let events2 = state.process_chunk(chunk2).unwrap().unwrap();
assert_eq!(events2.len(), 1);
assert!(matches!(events2[0], StreamEvent::TextDelta { .. }));
@ -395,14 +415,14 @@ mod tests {
let chunk1: StreamChunk = serde_json::from_str(
r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"fn1","arguments":"{\"k"}}]},"finish_reason":null}]}"#,
).unwrap();
let events1 = state.process_chunk(chunk1).unwrap();
let events1 = state.process_chunk(chunk1).unwrap().unwrap();
assert_eq!(events1.len(), 1);
assert!(matches!(events1[0], StreamEvent::ToolCallStart { .. }));
let chunk2: StreamChunk = serde_json::from_str(
r#"{"id":"c1","model":"m1","choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ey\"}"}}]},"finish_reason":null}]}"#,
).unwrap();
let events2 = state.process_chunk(chunk2).unwrap();
let events2 = state.process_chunk(chunk2).unwrap().unwrap();
assert_eq!(events2.len(), 1);
assert!(matches!(events2[0], StreamEvent::ToolCallDelta { .. }));
@ -479,6 +499,37 @@ mod tests {
}
}
// Reproduces run 01M11JZVT7V507R56BCJJHZB1B: venice (proxying Anthropic)
// numbered tool_calls[].index by content block, so the first tool call
// arrived with index 1 when text preceded it. Padding the skipped slot
// used to materialize a phantom tool call with an empty id and name that
// the provider rejected once echoed back (tool_use.id must match
// '^[a-zA-Z0-9_-]+$'). A gap in the index sequence is indistinguishable
// from lost chunks, so the stream must fail instead.
#[test]
fn sparse_tool_call_index_is_a_stream_error() {
let mut state = test_state("venice", "claude-opus-5");
let text_chunk: StreamChunk = serde_json::from_str(
r#"{"id":"c1","model":"claude-opus-5","choices":[{"delta":{"content":"I'll start by reading the state file."},"finish_reason":null}]}"#,
)
.unwrap();
state.process_chunk(text_chunk).unwrap();
let tool_chunk: StreamChunk = serde_json::from_str(
r#"{"id":"c1","model":"claude-opus-5","choices":[{"delta":{"tool_calls":[{"index":1,"id":"toolu_01EgMidFVtGhitWE22jXQ9Eo","function":{"name":"Read","arguments":"{\"file_path\":\"state.json\"}"}}]},"finish_reason":null}]}"#,
)
.unwrap();
let err = state.process_chunk(tool_chunk).unwrap_err();
assert!(err.retryable(), "malformed stream should be retryable");
let message = err.to_string();
assert!(
message.contains("tool_calls[1]") && message.contains("venice"),
"unexpected error message: {message}"
);
}
#[test]
fn uses_request_model_as_fallback() {
let mut state = test_state("test", "fallback-model");

View file

@ -56,6 +56,18 @@ pub enum Error {
run_id: String,
field: &'static str,
},
#[error("run {run_id} head mismatch: expected {expected_last_seq}, stored {actual_last_seq:?}")]
RunHeadMismatch {
run_id: String,
expected_last_seq: u32,
actual_last_seq: Option<u32>,
},
#[error("stored run event {run_id} sequence {seq} has inconsistent field {field}")]
RunEventMismatch {
run_id: String,
seq: u32,
field: &'static str,
},
#[error(transparent)]
InvalidTransition(#[from] fabro_types::InvalidTransition),
#[error("{0}")]

View file

@ -1128,7 +1128,7 @@ mod tests {
PASSIVE_CHECKPOINT_BYTES, set_automatic_checkpoint,
};
use crate::keys::SlateKey;
use crate::{BlobStore, Database};
use crate::{BlobStore, Database, test_support as store_test_support};
type TestResult<T> = std::result::Result<T, Box<dyn std::error::Error>>;
@ -1153,6 +1153,7 @@ mod tests {
Duration::from_millis(1),
None,
Arc::clone(&target),
store_test_support::test_run_summary_store(),
);
let source_db = source.open_db().await?;
Ok(Self {
@ -1853,6 +1854,7 @@ mod tests {
Duration::from_millis(1),
None,
Arc::clone(&target),
store_test_support::test_run_summary_store(),
);
let mut connection = pool.acquire().await?;

File diff suppressed because it is too large Load diff

View file

@ -4,7 +4,7 @@ mod run_store;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, Utc};
@ -45,7 +45,7 @@ pub struct Database {
catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
projection_cache: Arc<RunProjectionCache>,
projection_cache_warmed: Arc<OnceCell<()>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
run_summary_store: Arc<RunSummaryStore>,
}
impl std::fmt::Debug for Database {
@ -65,6 +65,7 @@ impl Database {
flush_interval: Duration,
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
run_summary_store: Arc<RunSummaryStore>,
) -> Self {
Self {
object_store,
@ -77,16 +78,13 @@ impl Database {
catalog_index: Arc::new(OnceCell::new()),
projection_cache: Arc::new(RunProjectionCache::default()),
projection_cache_warmed: Arc::new(OnceCell::new()),
run_summary_store: Arc::new(OnceLock::new()),
run_summary_store,
}
}
pub fn attach_run_summary_store(&self, store: Arc<RunSummaryStore>) -> Arc<RunSummaryStore> {
Arc::clone(self.run_summary_store.get_or_init(|| store))
}
fn run_summary_store(&self) -> Option<Arc<RunSummaryStore>> {
self.run_summary_store.get().cloned()
#[must_use]
pub fn run_summary_store(&self) -> Arc<RunSummaryStore> {
Arc::clone(&self.run_summary_store)
}
fn shared_db_prefix(&self) -> String {
@ -142,7 +140,7 @@ impl Database {
read_only,
self.blobs(),
Arc::clone(&self.projection_cache),
Arc::clone(&self.run_summary_store),
self.run_summary_store(),
)
.await
}
@ -240,9 +238,7 @@ impl Database {
}
}
}
if let Some(store) = self.run_summary_store() {
store.reconcile(&entries).await?;
}
self.run_summary_store.reconcile(&entries).await?;
self.projection_cache.replace_all(entries).await;
Ok::<_, Error>(())
})
@ -389,9 +385,7 @@ impl Database {
self.delete_session_indexes_for_run(run_id).await?;
self.catalog_index().await?.remove(run_id).await?;
self.remove_cached_run(run_id).await;
if let Some(store) = self.run_summary_store() {
store.delete(run_id).await?;
}
self.run_summary_store.delete(run_id).await?;
Ok(())
}
@ -558,6 +552,21 @@ mod tests {
(object_store, store)
}
fn make_store_with_run_summaries(
run_summaries: Arc<RunSummaryStore>,
) -> (Arc<dyn ObjectStore>, Database) {
let object_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
let store = store_test_support::test_database_with_stores(
object_store.clone(),
"runs/",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
run_summaries,
);
(object_store, store)
}
#[tokio::test]
async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() {
let (_object_store, store) = make_store();
@ -596,8 +605,8 @@ mod tests {
);
}
async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
let (directory, store) = store_test_support::sqlite_summary_store().await;
async fn make_run_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
let (directory, store) = store_test_support::sqlite_run_summary_store().await;
(directory, Arc::new(store))
}
@ -972,9 +981,8 @@ mod tests {
#[tokio::test]
async fn rejected_transition_leaves_reconciled_summary_present() {
let (_object_store, store) = make_store();
let (_directory, summaries) = make_summary_store().await;
store.attach_run_summary_store(Arc::clone(&summaries));
let (_directory, summaries) = make_run_summary_store().await;
let (_object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_runnable(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
@ -995,10 +1003,9 @@ mod tests {
}
#[tokio::test]
async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() {
let (object_store, store) = make_store();
let (directory, summaries) = make_summary_store().await;
store.attach_run_summary_store(Arc::clone(&summaries));
async fn best_effort_run_summary_update_failure_keeps_slate_append_repairable() {
let (directory, summaries) = make_run_summary_store().await;
let (object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run_id = test_run_id("run-1");
let run = store.create_run(&run_id).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
@ -1022,7 +1029,7 @@ mod tests {
assert_eq!(stored.event, result.unwrap().event);
let repaired_summaries =
Arc::new(store_test_support::sqlite_summary_store_at(directory.path()).await);
Arc::new(store_test_support::sqlite_run_summary_store_at(directory.path()).await);
let stale = repaired_summaries
.get(&run_id, Utc::now())
.await
@ -1030,13 +1037,14 @@ mod tests {
.unwrap();
assert_ne!(stale.title, "Committed title");
let reopened = store_test_support::test_database(
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs/",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
Arc::clone(&repaired_summaries),
);
reopened.attach_run_summary_store(Arc::clone(&repaired_summaries));
reopened.warm_projection_cache().await.unwrap();
let repaired = repaired_summaries
.get(&run_id, Utc::now())
@ -1657,10 +1665,9 @@ mod tests {
}
#[tokio::test]
async fn append_event_refreshes_projection_cache_and_delete_removes_it() {
let (_object_store, store) = make_store();
let (_directory, summaries) = make_summary_store().await;
store.attach_run_summary_store(Arc::clone(&summaries));
async fn required_run_summary_append_refreshes_cache_and_delete_removes_rows() {
let (_directory, summaries) = make_run_summary_store().await;
let (_object_store, store) = make_store_with_run_summaries(Arc::clone(&summaries));
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
store.warm_projection_cache().await.unwrap();
@ -1834,15 +1841,20 @@ mod tests {
}
#[tokio::test]
async fn projection_cache_warmup_backfills_sqlite_run_summaries() {
async fn required_run_summary_warmup_backfills_sqlite_run_summaries() {
let (object_store, store) = make_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
let reopened =
store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
let (_directory, summaries) = make_summary_store().await;
reopened.attach_run_summary_store(Arc::clone(&summaries));
let (_directory, summaries) = make_run_summary_store().await;
let reopened = store_test_support::test_database_with_stores(
object_store,
"runs",
Duration::from_millis(1),
None,
store_test_support::test_blob_store(),
Arc::clone(&summaries),
);
reopened.warm_projection_cache().await.unwrap();
let summary = summaries

View file

@ -1,6 +1,6 @@
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, OnceLock};
use bytes::Bytes;
use chrono::Utc;
@ -45,9 +45,7 @@ pub(crate) struct RunDatabaseInner {
state_lock: Mutex<()>,
projection_cache: Mutex<EventProjectionCache>,
shared_projection_cache: Arc<RunProjectionCache>,
// Shared cell rather than a snapshot so a summary store attached after
// this writer opened is still picked up by later appends.
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
run_summary_store: Arc<RunSummaryStore>,
recent_events: Mutex<VecDeque<EventEnvelope>>,
recent_event_limit: usize,
event_tx: broadcast::Sender<EventEnvelope>,
@ -60,7 +58,7 @@ impl RunDatabase {
read_only: bool,
blob_store: Arc<BlobStore>,
shared_projection_cache: Arc<RunProjectionCache>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
run_summary_store: Arc<RunSummaryStore>,
) -> Result<Self> {
let cached_projection = shared_projection_cache.projection_snapshot(&run_id).await;
let projection_cache = cached_projection.as_ref().map_or_else(
@ -231,15 +229,13 @@ impl RunDatabase {
}
async fn update_summary_after_committed_append(&self, cached: &CachedRunProjection) {
if let Some(store) = self.inner.run_summary_store.get() {
if let Err(err) = store.upsert_projection(cached).await {
warn!(
run_id = %self.inner.run_id,
source_last_seq = cached.last_seq,
error = ?err,
"failed to update SQLite run summary after committed append"
);
}
if let Err(err) = self.inner.run_summary_store.upsert_projection(cached).await {
warn!(
run_id = %self.inner.run_id,
source_last_seq = cached.last_seq,
error = ?err,
"failed to update SQLite run summary after committed append"
);
}
}

View file

@ -8,8 +8,8 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use crate::keys::SlateKey;
#[cfg(test)]
use crate::{AuthCodeStore, AuthSessionStore, RunSummaryStore};
use crate::{BlobStore, Database, Result};
use crate::{AuthCodeStore, AuthSessionStore};
use crate::{BlobStore, Database, Result, RunSummaryStore};
/// Returns an isolated SQLite blob authority backed by its own in-memory
/// database.
@ -18,32 +18,48 @@ use crate::{BlobStore, Database, Result};
/// by other tests in the same process. Reopen-style tests that model one
/// process-wide blob authority across several store handles should call this
/// once and share the result through [`test_database_with_blobs`].
///
/// The pool connects lazily so synchronous fixture builders can remain
/// synchronous. Its single connection installs the production blob schema on
/// first use.
#[must_use]
pub fn test_blob_store() -> Arc<BlobStore> {
Arc::new(BlobStore::new(lazy_in_memory_pool(&[
fabro_db::BLOBS_MIGRATION_SQL,
])))
}
/// Returns an isolated SQLite run-summary store backed by its own in-memory
/// database and the production `runs` and `run_events` schemas.
#[must_use]
pub fn test_run_summary_store() -> Arc<RunSummaryStore> {
Arc::new(RunSummaryStore::new(lazy_in_memory_pool(&[
fabro_db::RUNS_MIGRATION_SQL,
fabro_db::RUN_EVENTS_MIGRATION_SQL,
])))
}
/// Builds a single-connection in-memory SQLite pool that installs
/// `migrations` on first use.
///
/// The pool connects lazily so synchronous fixture builders can remain
/// synchronous.
fn lazy_in_memory_pool(migrations: &'static [&'static str]) -> sqlx::SqlitePool {
let options = SqliteConnectOptions::new()
.filename(":memory:")
.foreign_keys(true);
let pool = SqlitePoolOptions::new()
SqlitePoolOptions::new()
.max_connections(1)
// A single in-memory test connection never needs reaping. Disabling
// both timers also keeps this lazy fixture constructible from sync
// tests, where SQLx has no Tokio runtime for maintenance tasks.
.max_lifetime(None)
.idle_timeout(None)
.after_connect(|connection, _metadata| {
.after_connect(move |connection, _metadata| {
Box::pin(async move {
sqlx::query(fabro_db::BLOBS_MIGRATION_SQL)
.execute(&mut *connection)
.await?;
for migration in migrations {
sqlx::raw_sql(*migration).execute(&mut *connection).await?;
}
Ok(())
})
})
.connect_lazy_with(options);
Arc::new(BlobStore::new(pool))
.connect_lazy_with(options)
}
/// Returns the SQLite file backing [`test_blob_store_at`] for `store_dir`.
@ -119,7 +135,37 @@ pub fn test_database_with_blobs(
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
) -> Database {
Database::new(object_store, base_prefix, flush_interval, cache_path, blobs)
test_database_with_stores(
object_store,
base_prefix,
flush_interval,
cache_path,
blobs,
test_run_summary_store(),
)
}
/// Builds a Slate-backed run database with explicit shared SQLite stores.
///
/// Use this only when a test needs a failing, persistent, or shared store;
/// ordinary fixtures should use [`test_database`].
#[must_use]
pub fn test_database_with_stores(
object_store: Arc<dyn ObjectStore>,
base_prefix: impl Into<String>,
flush_interval: Duration,
cache_path: Option<PathBuf>,
blobs: Arc<BlobStore>,
run_summaries: Arc<RunSummaryStore>,
) -> Database {
Database::new(
object_store,
base_prefix,
flush_interval,
cache_path,
blobs,
run_summaries,
)
}
/// Seeds one canonical row in the legacy SlateDB blob keyspace.
@ -171,13 +217,13 @@ pub(crate) async fn sqlite_auth_code_store() -> (tempfile::TempDir, AuthCodeStor
}
#[cfg(test)]
pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
pub(crate) async fn sqlite_run_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
let directory = tempfile::tempdir().unwrap();
let store = sqlite_summary_store_at(directory.path()).await;
let store = sqlite_run_summary_store_at(directory.path()).await;
(directory, store)
}
#[cfg(test)]
pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore {
pub(crate) async fn sqlite_run_summary_store_at(directory: &Path) -> RunSummaryStore {
RunSummaryStore::new(sqlite_test_pool(directory).await)
}

View file

@ -689,8 +689,6 @@ fn main() {
("AskFabro", "fabro_types::AskFabro", &[]),
("Automation", "fabro_automation::Automation", &[]),
("AutomationRef", "fabro_types::AutomationRef", &[]),
("AutomationTarget", "fabro_automation::AutomationTarget", &[
]),
(
"AutomationTrigger",
"fabro_automation::AutomationTrigger",

View file

@ -16,7 +16,7 @@ mod generated {
pub mod types {
pub use fabro_automation::{
Automation, AutomationDraft as CreateAutomationRequest,
AutomationReplace as ReplaceAutomationRequest, AutomationTarget, AutomationTrigger,
AutomationReplace as ReplaceAutomationRequest, AutomationTrigger,
};
pub use fabro_environment::Environment;
pub use fabro_model::{

View file

@ -1,12 +1,9 @@
use fabro_api::types::{
Automation as ApiAutomation, AutomationTarget as ApiAutomationTarget,
AutomationTrigger as ApiAutomationTrigger,
Automation as ApiAutomation, AutomationTrigger as ApiAutomationTrigger,
CreateAutomationRequest as ApiCreateAutomationRequest,
ReplaceAutomationRequest as ApiReplaceAutomationRequest,
};
use fabro_automation::{
Automation, AutomationDraft, AutomationReplace, AutomationTarget, AutomationTrigger,
};
use fabro_automation::{Automation, AutomationDraft, AutomationReplace, AutomationTrigger};
use serde_json::json;
// Compile-time witnesses that the generated API types resolve to the same
@ -14,7 +11,6 @@ use serde_json::json;
// If progenitor stops reusing the domain type, these functions stop type-
// checking and the build fails.
const _: fn(ApiAutomation) -> Automation = |value| value;
const _: fn(ApiAutomationTarget) -> AutomationTarget = |value| value;
const _: fn(ApiAutomationTrigger) -> AutomationTrigger = |value| value;
const _: fn(ApiCreateAutomationRequest) -> AutomationDraft = |value| value;
const _: fn(ApiReplaceAutomationRequest) -> AutomationReplace = |value| value;
@ -27,10 +23,13 @@ fn automation_response_round_trips_public_json_shape() {
"name": "Nightly dependency update",
"description": null,
"target": {
"repository": "fabro-sh/fabro",
"ref": "main",
"workflow": "dependency-update"
"kind": "git",
"repo": "fabro-sh/fabro",
"branch": "main",
"tag": "v1.2.3",
"sha": "0123456789abcdef0123456789abcdef01234567"
},
"workflow": "dependency-update",
"triggers": [
{
"id": "manual",
@ -57,10 +56,11 @@ fn create_automation_request_round_trips_public_json_shape() {
"name": "Nightly dependency update",
"description": "Keep dependencies fresh",
"target": {
"repository": "fabro-sh/fabro",
"ref": "main",
"workflow": "dependency-update"
"kind": "git",
"repo": "fabro-sh/fabro",
"branch": "main"
},
"workflow": "dependency-update",
"triggers": [
{
"id": "manual",
@ -80,10 +80,12 @@ fn replace_automation_request_round_trips_public_json_shape() {
"name": "Nightly dependency update",
"description": "Keep dependencies fresh",
"target": {
"repository": "fabro-sh/fabro",
"ref": "main",
"workflow": "dependency-update"
"kind": "git",
"repo": "fabro-sh/fabro",
"branch": "release",
"tag": "v2"
},
"workflow": "dependency-update",
"triggers": [
{
"id": "nightly",

View file

@ -0,0 +1,103 @@
CREATE TEMP TABLE automation_target_migration_candidates (
id TEXT PRIMARY KEY NOT NULL,
legacy_ref TEXT NOT NULL,
branch TEXT NOT NULL,
tag TEXT,
sha TEXT
);
CREATE TEMP TRIGGER reject_unsupported_automation_target
BEFORE INSERT ON automation_target_migration_candidates
WHEN
length(NEW.branch) NOT BETWEEN 1 AND 255
OR NEW.branch != trim(NEW.branch)
OR substr(NEW.branch, 1, 1) IN ('/', '-', '.')
OR substr(NEW.branch, -1, 1) IN ('/', '.')
OR NEW.branch = '@'
OR instr(NEW.branch, '..') > 0
OR instr(NEW.branch, '//') > 0
OR instr(NEW.branch, '@{') > 0
OR NEW.branch GLOB '*[^A-Za-z0-9/._-]*'
OR NEW.branch GLOB '*/.*'
OR NEW.branch GLOB '*.lock'
OR NEW.branch GLOB '*.lock/*'
OR NEW.branch = 'HEAD'
OR NEW.branch GLOB 'refs/*'
OR NEW.branch GLOB 'tags/*'
OR NEW.branch GLOB 'heads/*'
OR (length(NEW.branch) = 40 AND NEW.branch NOT GLOB '*[^0-9A-Fa-f]*')
OR (
NEW.tag IS NOT NULL
AND (
length(NEW.tag) NOT BETWEEN 1 AND 255
OR NEW.tag != trim(NEW.tag)
OR substr(NEW.tag, 1, 1) IN ('/', '-', '.')
OR substr(NEW.tag, -1, 1) IN ('/', '.')
OR NEW.tag = '@'
OR instr(NEW.tag, '..') > 0
OR instr(NEW.tag, '//') > 0
OR instr(NEW.tag, '@{') > 0
OR NEW.tag GLOB '*[^A-Za-z0-9/._-]*'
OR NEW.tag GLOB '*/.*'
OR NEW.tag GLOB '*.lock'
OR NEW.tag GLOB '*.lock/*'
OR NEW.tag = 'HEAD'
OR NEW.tag GLOB 'refs/*'
OR NEW.tag GLOB 'tags/*'
OR (length(NEW.tag) = 40 AND NEW.tag NOT GLOB '*[^0-9A-Fa-f]*')
)
)
BEGIN
SELECT RAISE(
ABORT,
'cannot migrate automations.target_ref: unsupported legacy selector; edit it to a branch, supported heads/tags selector, HEAD, or 40-hex SHA and restart'
);
END;
INSERT INTO automation_target_migration_candidates (id, legacy_ref, branch, tag, sha)
SELECT
id,
target_ref,
CASE
WHEN length(target_ref) = 40 AND target_ref NOT GLOB '*[^0-9A-Fa-f]*' THEN 'main'
WHEN target_ref GLOB 'refs/tags/?*' THEN 'main'
WHEN target_ref GLOB 'tags/?*' THEN 'main'
WHEN target_ref GLOB 'refs/heads/?*' THEN substr(target_ref, 12)
WHEN target_ref GLOB 'heads/?*' THEN substr(target_ref, 7)
WHEN target_ref = 'HEAD' THEN 'main'
ELSE target_ref
END,
CASE
WHEN target_ref GLOB 'refs/tags/?*' THEN substr(target_ref, 11)
WHEN target_ref GLOB 'tags/?*' THEN substr(target_ref, 6)
ELSE NULL
END,
CASE
WHEN length(target_ref) = 40 AND target_ref NOT GLOB '*[^0-9A-Fa-f]*' THEN lower(target_ref)
ELSE NULL
END
FROM automations;
DROP TRIGGER reject_unsupported_automation_target;
ALTER TABLE automations RENAME COLUMN target_ref TO target_branch;
ALTER TABLE automations ADD COLUMN target_tag TEXT
CHECK (target_tag IS NULL OR length(target_tag) BETWEEN 1 AND 255);
ALTER TABLE automations ADD COLUMN target_sha TEXT
CHECK (
target_sha IS NULL
OR (
length(target_sha) = 40
AND target_sha NOT GLOB '*[^0-9a-f]*'
)
);
UPDATE automations
SET
target_branch = candidates.branch,
target_tag = candidates.tag,
target_sha = candidates.sha
FROM automation_target_migration_candidates AS candidates
WHERE candidates.id = automations.id;
DROP TABLE automation_target_migration_candidates;

View file

@ -0,0 +1,30 @@
CREATE TABLE run_events (
run_id TEXT NOT NULL,
seq INTEGER NOT NULL,
event_name TEXT NOT NULL,
node_id TEXT,
stage_id TEXT,
session_id TEXT,
event_json TEXT NOT NULL,
PRIMARY KEY (run_id, seq),
FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE,
CHECK (seq BETWEEN 1 AND 999999),
CHECK (json_valid(event_json))
);
CREATE INDEX run_events_by_stage
ON run_events(run_id, stage_id, seq)
WHERE stage_id IS NOT NULL;
CREATE INDEX run_events_by_legacy_node
ON run_events(run_id, node_id, seq)
WHERE stage_id IS NULL AND node_id IS NOT NULL;
CREATE INDEX run_events_by_session
ON run_events(run_id, session_id, seq)
WHERE session_id IS NOT NULL
AND event_name GLOB 'run.session.*';
CREATE INDEX run_events_by_pull_request_creation_request
ON run_events(run_id, seq)
WHERE event_name = 'pull_request.creation_requested';

View file

@ -20,6 +20,14 @@ static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
/// the production blob schema without a filesystem path into this crate.
pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blobs.sql");
/// The run summary migration, exposed so fixtures in other crates can install
/// the production schema without a filesystem path into this crate.
pub const RUNS_MIGRATION_SQL: &str = include_str!("../migrations/2026071104_runs.sql");
/// The run-event migration, exposed so fixtures in other crates can install
/// the production schema without a filesystem path into this crate.
pub const RUN_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/2026082701_run_events.sql");
#[derive(Clone)]
pub struct Database {
pool: DbPool,

View file

@ -429,9 +429,11 @@ async fn insert_minimal_automation(
name,
api_enabled,
target_repository,
target_ref,
target_branch,
target_tag,
target_sha,
target_workflow
) VALUES (?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?)
",
)
.bind(id)
@ -446,6 +448,168 @@ async fn insert_minimal_automation(
Ok(())
}
#[tokio::test]
async fn automation_targets_migrate_offline_and_preserve_related_rows() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let db_path = dir.path().join("fabro.sqlite3");
let database = fabro_db::Database::connect(&db_path).await?;
database.migrate().await?;
rewind_automation_target_migration(&database).await?;
let values = [
("sha", "ABCDEF0123456789ABCDEF0123456789ABCDEF01"),
("tag-ref", "refs/tags/v1.2.3"),
("tag", "tags/v2"),
("head-ref", "refs/heads/release"),
("head", "heads/feature/test"),
("head-literal", "HEAD"),
("branch", "feature/bare"),
];
for (id, selector) in values {
insert_legacy_automation(database.pool(), id, selector).await?;
}
sqlx::query(
"INSERT INTO automation_triggers (automation_id, id, enabled, expression) \
VALUES ('tag-ref', 'nightly', 1, '0 3 * * *')",
)
.execute(database.pool())
.await?;
database.migrate().await?;
let rows = sqlx::query(
"SELECT id, revision, target_branch, target_tag, target_sha, target_workflow \
FROM automations ORDER BY id",
)
.fetch_all(database.pool())
.await?;
let projected = rows
.iter()
.map(|row| {
(
row.get::<String, _>("id"),
row.get::<String, _>("target_branch"),
row.get::<Option<String>, _>("target_tag"),
row.get::<Option<String>, _>("target_sha"),
)
})
.collect::<Vec<_>>();
assert_eq!(projected, vec![
("branch".to_string(), "feature/bare".to_string(), None, None),
("head".to_string(), "feature/test".to_string(), None, None),
("head-literal".to_string(), "main".to_string(), None, None),
("head-ref".to_string(), "release".to_string(), None, None),
(
"sha".to_string(),
"main".to_string(),
None,
Some("abcdef0123456789abcdef0123456789abcdef01".to_string()),
),
(
"tag".to_string(),
"main".to_string(),
Some("v2".to_string()),
None
),
(
"tag-ref".to_string(),
"main".to_string(),
Some("v1.2.3".to_string()),
None,
),
]);
assert!(
rows.iter()
.all(|row| row.get::<String, _>("revision") == "a".repeat(64))
);
assert!(
rows.iter()
.all(|row| row.get::<String, _>("target_workflow") == "release")
);
let trigger_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM automation_triggers WHERE automation_id = 'tag-ref'",
)
.fetch_one(database.pool())
.await?;
assert_eq!(trigger_count, 1);
assert!(fabro_db::pre_migration_snapshot_path(&db_path).exists());
database.migrate().await?;
assert_eq!(
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM automations")
.fetch_one(database.pool())
.await?,
7
);
Ok(())
}
#[tokio::test]
async fn unsupported_automation_targets_abort_before_schema_changes() -> anyhow::Result<()> {
for selector in ["refs/pull/123/head", "refs/heads/-bad", "tags/HEAD"] {
let dir = tempfile::tempdir()?;
let db_path = dir.path().join("fabro.sqlite3");
let database = fabro_db::Database::connect(&db_path).await?;
database.migrate().await?;
rewind_automation_target_migration(&database).await?;
insert_legacy_automation(database.pool(), "blocked", selector).await?;
let error = database.migrate().await.expect_err("migration must abort");
let rendered = format!("{error:#}");
assert!(rendered.contains("edit it to a branch"), "{rendered}");
let columns = sqlx::query("PRAGMA table_info(automations)")
.fetch_all(database.pool())
.await?;
let names = columns
.iter()
.map(|row| row.get::<String, _>("name"))
.collect::<Vec<_>>();
assert!(names.iter().any(|name| name == "target_ref"));
assert!(!names.iter().any(|name| name == "target_branch"));
let stored: String =
sqlx::query_scalar("SELECT target_ref FROM automations WHERE id = 'blocked'")
.fetch_one(database.pool())
.await?;
assert_eq!(stored, selector);
assert!(fabro_db::pre_migration_snapshot_path(&db_path).exists());
}
Ok(())
}
async fn rewind_automation_target_migration(database: &fabro_db::Database) -> anyhow::Result<()> {
sqlx::query("ALTER TABLE automations DROP COLUMN target_sha")
.execute(database.pool())
.await?;
sqlx::query("ALTER TABLE automations DROP COLUMN target_tag")
.execute(database.pool())
.await?;
sqlx::query("ALTER TABLE automations RENAME COLUMN target_branch TO target_ref")
.execute(database.pool())
.await?;
sqlx::query("DELETE FROM _sqlx_migrations WHERE version = 2026082601")
.execute(database.pool())
.await?;
Ok(())
}
async fn insert_legacy_automation(
pool: &fabro_db::DbPool,
id: &str,
selector: &str,
) -> Result<(), sqlx::Error> {
sqlx::query(
"INSERT INTO automations (\
id, revision, name, api_enabled, target_repository, target_ref, target_workflow\
) VALUES (?, ?, 'Automation', 1, 'fabro-sh/fabro', ?, 'release')",
)
.bind(id)
.bind("a".repeat(64))
.bind(selector)
.execute(pool)
.await?;
Ok(())
}
#[tokio::test]
async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
@ -475,22 +639,283 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul
Ok(())
}
#[tokio::test]
async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?;
database.migrate().await?;
let run_columns = sqlx::query("PRAGMA table_info(runs)")
.fetch_all(database.pool())
.await?;
assert_eq!(
run_columns.len(),
24,
"the existing runs row must stay unchanged"
);
let event_columns = sqlx::query("PRAGMA table_info(run_events)")
.fetch_all(database.pool())
.await?;
let event_column_contract = event_columns
.iter()
.map(|column| {
(
column.get::<String, _>("name"),
column.get::<String, _>("type"),
column.get::<i64, _>("notnull"),
column.get::<i64, _>("pk"),
)
})
.collect::<Vec<_>>();
assert_eq!(event_column_contract, vec![
("run_id".to_string(), "TEXT".to_string(), 1, 1),
("seq".to_string(), "INTEGER".to_string(), 1, 2),
("event_name".to_string(), "TEXT".to_string(), 1, 0),
("node_id".to_string(), "TEXT".to_string(), 0, 0),
("stage_id".to_string(), "TEXT".to_string(), 0, 0),
("session_id".to_string(), "TEXT".to_string(), 0, 0),
("event_json".to_string(), "TEXT".to_string(), 1, 0),
]);
let foreign_keys = sqlx::query("PRAGMA foreign_key_list(run_events)")
.fetch_all(database.pool())
.await?;
assert_eq!(foreign_keys.len(), 1);
assert_eq!(foreign_keys[0].get::<String, _>("table"), "runs");
assert_eq!(foreign_keys[0].get::<String, _>("from"), "run_id");
assert_eq!(foreign_keys[0].get::<String, _>("to"), "id");
assert_eq!(foreign_keys[0].get::<String, _>("on_delete"), "CASCADE");
let indexes = sqlx::query("PRAGMA index_list(run_events)")
.fetch_all(database.pool())
.await?;
let named_indexes = indexes
.iter()
.filter_map(|index| {
let name = index.get::<String, _>("name");
name.starts_with("run_events_by_").then_some((
name,
index.get::<i64, _>("unique"),
index.get::<i64, _>("partial"),
))
})
.collect::<Vec<_>>();
assert_eq!(named_indexes, vec![
(
"run_events_by_pull_request_creation_request".to_string(),
0,
1,
),
("run_events_by_session".to_string(), 0, 1),
("run_events_by_legacy_node".to_string(), 0, 1),
("run_events_by_stage".to_string(), 0, 1),
]);
assert!(indexes.iter().all(|index| {
index.get::<i64, _>("unique") == 0
|| index.get::<String, _>("name") == "sqlite_autoindex_run_events_1"
}));
insert_run_with_id(database.pool(), "parent", None).await?;
insert_run_with_id(database.pool(), "child", Some("parent")).await?;
insert_run_event(database.pool(), "parent", 1, "run.created").await?;
for invalid in [
insert_run_event(database.pool(), "parent", 1, "run.created").await,
insert_run_event(database.pool(), "missing", 1, "run.created").await,
insert_run_event(database.pool(), "parent", 0, "run.created").await,
insert_run_event(database.pool(), "parent", 1_000_000, "run.created").await,
] {
assert!(invalid.is_err());
}
let invalid_json = sqlx::query(
"INSERT INTO run_events (run_id, seq, event_name, event_json) VALUES (?, ?, ?, ?)",
)
.bind("parent")
.bind(2_i64)
.bind("run.started")
.bind("not-json")
.execute(database.pool())
.await;
assert!(invalid_json.is_err());
sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)")
.bind("a".repeat(64))
.bind(vec![1_u8])
.execute(database.pool())
.await?;
sqlx::query("DELETE FROM runs WHERE id = ?")
.bind("parent")
.execute(database.pool())
.await?;
let event_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM run_events WHERE run_id = 'parent'")
.fetch_one(database.pool())
.await?;
let child_parent: Option<String> =
sqlx::query_scalar("SELECT parent_id FROM runs WHERE id = 'child'")
.fetch_one(database.pool())
.await?;
let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs")
.fetch_one(database.pool())
.await?;
assert_eq!(event_count, 0);
assert_eq!(child_parent.as_deref(), Some("parent"));
assert_eq!(blob_count, 1);
Ok(())
}
#[tokio::test]
async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?;
database.migrate().await?;
for (sql, expected_index) in [
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?",
"sqlite_autoindex_run_events_1",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq = ?",
"sqlite_autoindex_run_events_1",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND stage_id = ? ORDER BY seq ASC LIMIT ?",
"run_events_by_stage",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND stage_id IS NULL AND node_id = ? ORDER BY seq ASC LIMIT ?",
"run_events_by_legacy_node",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND session_id = ? AND event_name GLOB 'run.session.*' ORDER BY seq ASC LIMIT ?",
"run_events_by_session",
),
(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE event_name = 'pull_request.creation_requested' ORDER BY run_id, seq",
"run_events_by_pull_request_creation_request",
),
] {
let details = sqlx::query(sql)
.bind("run")
.bind("value")
.bind(10_i64)
.fetch_all(database.pool())
.await?
.into_iter()
.map(|row| row.get::<String, _>("detail"))
.collect::<Vec<_>>()
.join("; ");
assert!(
details.contains(expected_index),
"expected {expected_index} in query plan: {details}"
);
}
// The first-visit stage listing unions both shapes so each arm keeps its
// own partial index instead of scanning the run's primary key range.
let details = sqlx::query(
"EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq >= ? AND stage_id = ? \
UNION ALL SELECT * FROM run_events WHERE run_id = ? AND seq >= ? AND stage_id IS NULL AND node_id = ? \
ORDER BY seq ASC LIMIT ?",
)
.bind("run")
.bind(1_i64)
.bind("stage")
.bind("run")
.bind(1_i64)
.bind("node")
.bind(10_i64)
.fetch_all(database.pool())
.await?
.into_iter()
.map(|row| row.get::<String, _>("detail"))
.collect::<Vec<_>>()
.join("; ");
for expected_index in ["run_events_by_stage", "run_events_by_legacy_node"] {
assert!(
details.contains(expected_index),
"expected {expected_index} in query plan: {details}"
);
}
Ok(())
}
async fn insert_run_event(
pool: &fabro_db::DbPool,
run_id: &str,
seq: i64,
event_name: &str,
) -> Result<(), sqlx::Error> {
sqlx::query(
r"
INSERT INTO run_events (run_id, seq, event_name, event_json)
VALUES (?, ?, ?, '{}')
",
)
.bind(run_id)
.bind(seq)
.bind(event_name)
.execute(pool)
.await?;
Ok(())
}
async fn insert_minimal_run(
pool: &fabro_db::DbPool,
status: &str,
input_tokens: i64,
summary_json: &str,
) -> Result<(), sqlx::Error> {
insert_run_row(
pool,
&format!("run-{status}-{input_tokens}"),
None,
status,
input_tokens,
summary_json,
)
.await
}
async fn insert_run_with_id(
pool: &fabro_db::DbPool,
id: &str,
parent_id: Option<&str>,
) -> Result<(), sqlx::Error> {
insert_run_row(
pool,
id,
parent_id,
"submitted",
0,
&format!(r#"{{"id":"{id}"}}"#),
)
.await
}
async fn insert_run_row(
pool: &fabro_db::DbPool,
id: &str,
parent_id: Option<&str>,
status: &str,
input_tokens: i64,
summary_json: &str,
) -> Result<(), sqlx::Error> {
sqlx::query(
r"
INSERT INTO runs (
id, source_last_seq, created_at_ms, last_event_at_ms, status, title,
id, source_last_seq, created_at_ms, last_event_at_ms, status, parent_id, title,
input_tokens, summary_json
) VALUES (?, 1, 0, 0, ?, 'title', ?, ?)
) VALUES (?, 1, 0, 0, ?, ?, 'title', ?, ?)
",
)
.bind(format!("run-{status}-{input_tokens}"))
.bind(id)
.bind(status)
.bind(parent_id)
.bind(input_tokens)
.bind(summary_json)
.execute(pool)

View file

@ -64,7 +64,6 @@ models/automation-list-meta.ts
models/automation-list-response.ts
models/automation-ref.ts
models/automation-schedule-trigger.ts
models/automation-target.ts
models/automation-trigger.ts
models/automation.ts
models/batch-delete-runs-request.ts

View file

@ -1,33 +0,0 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Repository and workflow selected by an automation.
*/
export interface AutomationTarget {
/**
* GitHub repository slug in `owner/repo` form.
*/
'repository': string;
/**
* Branch, tag, or SHA selector resolved when materializing a run.
*/
'ref': string;
/**
* Workflow slug or path resolved in the target repository.
*/
'workflow': string;
}

View file

@ -15,10 +15,10 @@
// May contain unused imports in some cases
// @ts-ignore
import type { AutomationTarget } from './automation-target';
import type { AutomationTrigger } from './automation-trigger';
// May contain unused imports in some cases
// @ts-ignore
import type { AutomationTrigger } from './automation-trigger';
import type { RunTarget } from './run-target';
/**
* Public automation definition.
@ -31,6 +31,10 @@ export interface Automation {
'revision': string;
'name': string;
'description': string | null;
'target': AutomationTarget;
'target': RunTarget;
/**
* Workflow slug or path resolved in the selected repository checkout.
*/
'workflow': string;
'triggers': Array<AutomationTrigger>;
}

View file

@ -15,10 +15,10 @@
// May contain unused imports in some cases
// @ts-ignore
import type { AutomationTarget } from './automation-target';
import type { AutomationTrigger } from './automation-trigger';
// May contain unused imports in some cases
// @ts-ignore
import type { AutomationTrigger } from './automation-trigger';
import type { RunTarget } from './run-target';
/**
* Request body for creating an automation.
@ -27,6 +27,10 @@ export interface CreateAutomationRequest {
'id': string;
'name': string;
'description'?: string | null;
'target': AutomationTarget;
'target': RunTarget;
/**
* Workflow slug or path resolved in the selected repository checkout.
*/
'workflow': string;
'triggers': Array<AutomationTrigger>;
}

View file

@ -35,7 +35,6 @@ export * from './automation-list-meta';
export * from './automation-list-response';
export * from './automation-ref';
export * from './automation-schedule-trigger';
export * from './automation-target';
export * from './automation-trigger';
export * from './batch-delete-runs-request';
export * from './batch-delete-runs-response';

View file

@ -15,10 +15,10 @@
// May contain unused imports in some cases
// @ts-ignore
import type { AutomationTarget } from './automation-target';
import type { AutomationTrigger } from './automation-trigger';
// May contain unused imports in some cases
// @ts-ignore
import type { AutomationTrigger } from './automation-trigger';
import type { RunTarget } from './run-target';
/**
* Request body for replacing an automation.
@ -26,6 +26,10 @@ import type { AutomationTrigger } from './automation-trigger';
export interface ReplaceAutomationRequest {
'name': string;
'description'?: string | null;
'target': AutomationTarget;
'target': RunTarget;
/**
* Workflow slug or path resolved in the selected repository checkout.
*/
'workflow': string;
'triggers': Array<AutomationTrigger>;
}