M6 CI (contract between workers)

Decided with the user (don't re-open):

• Workflow file .forge/ci.toml. No YAML anywhere.
• Jobs run in rootless podman.
• The runner is a separate forge-runner binary that pulls jobs over HTTPS. The forge never connects to a runner.
• v1 has secrets, artifacts and live logs. No cache.

Global rules:

• No JS and no inline styles.
• No unwrap in non-test code.
• Tests use tempfile::TempDir.
• Never put cargo target dirs or big files in /tmp.
• New crates go in at their latest version with cargo add, recorded in the root [workspace.dependencies].
• User values come after --end-of-options when passed to git.

0. Crates

• New crates/forge-ci (library; deps: serde, toml, serde_json, globset, thiserror). It is shared by forge-web and forge-runner, and has no turso, topcoat or tokio dependency.
• New crates/forge-runner (binary). Depends on forge-ci, reqwest (rustls, stream), tokio, clap, serde, toml, tempfile, sha2 and tracing. Not forge-core or forge-web.
• Add both to the root members.

1. forge-ci (worker W1)

// config.rs : parse .forge/ci.toml
pub struct CiConfig { pub on: Triggers, pub jobs: BTreeMap<String, JobSpec> }   // job names: ^[a-z0-9][a-z0-9_-]{0,39}$, max 20 jobs
pub struct Triggers { pub push: Vec<String> /* branch globs, e.g. "main", "release/*" */, pub tags: Vec<String> /* tag globs */,
                      pub pull_request: bool, pub manual: bool /* default true */ }
pub struct JobSpec { pub image: String, pub run: String, pub needs: Vec<String>, pub secrets: Vec<String> /* ^[A-Z_][A-Z0-9_]{0,63}$ */,
                     pub env: BTreeMap<String, String>, pub artifacts: Vec<String> /* relative globs, no "..", no leading "/" */,
                     pub runs_on: Vec<String> /* runner labels, all must match; default [] */, pub timeout_minutes: u32 /* default 60, 1..=360 */,
                     pub network: bool /* default true */ }
pub fn parse_config(text: &str) -> Result<CiConfig, ConfigError>  // deny_unknown_fields; validates names, needs refer to existing jobs, no cycles, image non-empty without whitespace, run non-empty, ≤ 64 KiB file
pub enum Event<'a> { Push { branch: &'a str }, Tag { tag: &'a str }, PullRequest, Manual }
impl CiConfig { pub fn triggered_by(&self, ev: &Event) -> bool; pub fn job_order(&self) -> Vec<&str> /* topological, stable by name */ }

// proto.rs : runner <-> forge JSON (serde, snake_case)
pub struct RegisterRequest { pub token: String, pub name: String, pub labels: Vec<String> }
pub struct RegisterResponse { pub runner_id: String, pub runner_secret: String }
pub struct ClaimRequest { pub labels: Vec<String> }                       // POST, long-poll ≤ 30s; 204 = nothing
pub struct JobAssignment { pub job_id: String, pub job_token: String, pub repo: String /* owner/name */, pub clone_url: String,
                           pub commit: String, pub git_ref: String, pub job_name: String, pub image: String, pub run: String,
                           pub env: BTreeMap<String,String>, pub secrets: BTreeMap<String,String>, pub artifacts: Vec<String>,
                           pub timeout_minutes: u32, pub network: bool }
pub struct HeartbeatResponse { pub cancel: bool }
pub struct FinishRequest { pub outcome: Outcome, pub exit_code: Option<i32>, pub message: Option<String> }
pub enum Outcome { Success, Failure, Cancelled, Error }

// mask.rs
pub struct Masker  // new(secrets: impl IntoIterator<Item=String>); ignores values shorter than 4 chars; mask(&str) -> String replaces every
                   // occurrence (and the base64 standard encoding of each value) with "***"; must handle a value split across two
                   // chunks: `MaskStream` with push(&[u8]) -> Vec<u8> that holds back max_len-1 bytes, and finish() -> Vec<u8>

Unit tests cover every validation error, glob triggers, topological order, cycle detection, and masking across chunk boundaries.

2. Schema and store (migration 0024_ci, worker W2)

CREATE TABLE ci_runners (id TEXT PRIMARY KEY, name TEXT NOT NULL, labels TEXT NOT NULL /* json */, secret_hash TEXT NOT NULL UNIQUE,
  created_at TEXT NOT NULL, last_seen_at TEXT, disabled INTEGER NOT NULL DEFAULT 0);
CREATE TABLE ci_registration_tokens (token_hash TEXT PRIMARY KEY, created_by TEXT NOT NULL REFERENCES users(id),
  created_at TEXT NOT NULL, expires_at TEXT NOT NULL, used_at TEXT);
CREATE TABLE ci_runs (id TEXT PRIMARY KEY, repo_id TEXT NOT NULL REFERENCES repos(id), number INTEGER NOT NULL,
  trigger TEXT NOT NULL /* push|tag|pull_request|manual */, git_ref TEXT NOT NULL, commit_oid TEXT NOT NULL, pull_number INTEGER,
  actor_id TEXT NOT NULL, state TEXT NOT NULL /* waiting_approval|queued|running|success|failure|cancelled|error */,
  config_error TEXT, created_at TEXT NOT NULL, finished_at TEXT, UNIQUE (repo_id, number));
CREATE TABLE ci_jobs (id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES ci_runs(id) ON DELETE CASCADE, name TEXT NOT NULL,
  spec TEXT NOT NULL /* JobSpec json */, state TEXT NOT NULL /* blocked|queued|running|success|failure|cancelled|error|skipped */,
  runner_id TEXT, job_token_hash TEXT, exit_code INTEGER, message TEXT, queued_at TEXT, started_at TEXT, finished_at TEXT,
  heartbeat_at TEXT, cancel_requested INTEGER NOT NULL DEFAULT 0, log_bytes INTEGER NOT NULL DEFAULT 0, UNIQUE (run_id, name));
CREATE INDEX ci_jobs_queue ON ci_jobs(state, queued_at);
CREATE TABLE ci_artifacts (id TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES ci_jobs(id) ON DELETE CASCADE, name TEXT NOT NULL,
  size INTEGER NOT NULL, sha256 TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE (job_id, name));
CREATE TABLE ci_secrets (repo_id TEXT NOT NULL REFERENCES repos(id), name TEXT NOT NULL, sealed BLOB NOT NULL,
  created_at TEXT NOT NULL, updated_at TEXT NOT NULL, PRIMARY KEY (repo_id, name));

crates/forge-core/src/ci.rs holds the models. crates/forge-core/src/store/ci.rs holds the functions below; they follow the read-pool rules in ~/.cache/forge-orch/read-pool-rules.md, with pure SELECTs on read_conn().

pub enum RunState { WaitingApproval, Queued, Running, Success, Failure, Cancelled, Error }   // as_str/FromStr
pub enum JobState { Blocked, Queued, Running, Success, Failure, Cancelled, Error, Skipped }  // as_str/FromStr, is_terminal()
pub struct CiRunner { id, name, labels: Vec<String>, created_at, last_seen_at: Option<String>, disabled: bool }
pub struct CiRun { id, repo_id, number: i64, trigger: String, git_ref, commit_oid, pull_number: Option<i64>, actor_id, state: RunState,
                   config_error: Option<String>, created_at, finished_at: Option<String> }
pub struct CiJob { id, run_id, name, spec_json: String, state: JobState, runner_id: Option<String>, exit_code: Option<i64>,
                   message: Option<String>, queued_at, started_at, finished_at, heartbeat_at: Option<String>, cancel_requested: bool, log_bytes: i64 }
pub struct CiArtifact { id, job_id, name, size: i64, sha256, created_at }
pub struct NewJob<'a> { pub name: &'a str, pub spec_json: &'a str, pub needs: &'a [String] }

// runners
create_registration_token(created_by, ttl_hours: u32) -> Result<String /*plaintext, shown once*/>   // stores sha256
register_runner(token_plain, name, labels: &[String]) -> Result<(CiRunner, String /*runner secret plaintext*/)>  // token single-use, not expired; Invalid otherwise
runner_by_secret(secret_plain) -> Result<Option<CiRunner>>     // constant-time compare via hash lookup; also updates last_seen_at
list_runners() / set_runner_disabled(id, bool) / delete_runner(id)
// runs and jobs
create_run(repo_id, trigger, git_ref, commit_oid, pull_number: Option<i64>, actor_id, needs_approval: bool, jobs: &[NewJob], config_error: Option<&str>) -> Result<CiRun>
   // number = max+1 per repo, in one transaction. Jobs with no needs start Queued (or Blocked when needs_approval), others Blocked.
   // With config_error: no jobs, state Error.
approve_run(run_id) -> Result<()>                          // WaitingApproval -> Queued; needless jobs -> Queued
claim_job(runner: &CiRunner) -> Result<Option<(CiJob, CiRun, String /*job token plaintext*/)>>
   // oldest Queued job whose spec.runs_on ⊆ runner.labels, in a transaction; sets Running, runner_id, started_at, heartbeat_at, token hash; run -> Running
job_by_token(job_id, token_plain) -> Result<Option<CiJob>>  // only while Running
heartbeat(job_id) -> Result<bool /*cancel_requested*/>
add_log_bytes(job_id, n: i64) -> Result<()>
finish_job(job_id, state: JobState, exit_code: Option<i64>, message: Option<&str>) -> Result<CiRun>
   // terminal; then unblock dependants (all needs Success -> Queued; any need not Success -> Skipped, cascading);
   // run state recomputed: any running/queued/blocked -> Running; else Failure if any failure/error, Cancelled if any cancelled, else Success
request_cancel(run_id) -> Result<()>   // Queued/Blocked jobs -> Cancelled at once; Running jobs get cancel_requested = 1
rerun(run_id, actor_id) -> Result<CiRun>  // new run with the same commit, ref, trigger and job specs
reap_lost_jobs(older_than: &str /*stamp*/) -> Result<Vec<CiJob>>  // Running jobs with heartbeat_at < older_than -> Error "runner lost"
run_by_number(repo_id, n) / run_by_id(id) / list_runs(repo_id, page, per_page) -> (Vec<CiRun>, total) / jobs_for_run(run_id) / job_by_id(id)
add_artifact(job_id, name, size, sha256) / list_artifacts(job_id) / artifact(job_id, name)
purge_old(logs_before: &str, artifacts_before: &str) -> Result<(Vec<String> /*job ids whose logs to delete*/, Vec<String> /*artifact ids*/)>
// secrets (sealed with the instance key like mirror secrets)
set_ci_secret(repo_id, name, value: &[u8]) / delete_ci_secret(repo_id, name) -> Result<bool> / list_ci_secret_names(repo_id) -> Result<Vec<(String, String /*updated_at*/)>>
ci_secret_values(repo_id, names: &[String]) -> Result<BTreeMap<String, String>>   // only for claim; missing names omitted

• delete_repo deletes the ci_runs (cascade) and ci_secrets rows.
• User deletion reassigns actor_id and created_by to the ghost user.

3. forge-web: scheduling (worker W3), crates/forge-web/src/ci/{mod.rs, schedule.rs, reaper.rs}

• pub async fn on_ref_update(state, repo, actor_id, refname, new_oid) is called from post_update::after_ref_update for each updated refs/heads/* or refs/tags/* (not deletions; not pull mirrors pulling? yes, pull mirrors also trigger, with the system actor).
  • It reads .forge/ci.toml at new_oid (forge-git blob read, capped at 64 KiB), parses it, and checks triggered_by.
  • It creates the run with jobs in job_order, then sets a commit status ci/<job> = pending for each job, with target_url = the job page.
  • A parse error creates a run in state Error with config_error and a status ci/config = error.
  • No file means nothing happens.
• pub async fn on_pull_event(state, repo, pull, actor_id) runs on PR open, synchronize and reopen (hook into the same places that emit the pull_request webhook).
  • It reads the config at the PR head commit, with event PullRequest.
  • If the head repo is a fork, or the author doesn't have Write on the base repo: needs_approval = true, the state is WaitingApproval, and the statuses are pending with the description "Waiting for approval".
• pub async fn manual_run(state, repo, branch, actor_id) -> Result<CiRun> runs only when manual is true.
• Statuses: whenever a job changes state, crate::ci::sync_status(state, run, job) sets ci/<job>:
  • queued/blocked/running → pending;
  • success → success;
  • failure → failure;
  • error/cancelled → error;
  • skipped → error, "skipped".
• Reaper task, started at boot:
  • every 30s it runs reap_lost_jobs(now - 120s) and syncs the statuses;
  • daily it runs purge_old with [ci] log_retention_days (default 90) and artifact_retention_days (default 30), and deletes the files.
• AppState.ci_waker: Arc<Notify> is notified whenever a job becomes Queued, so long-polling claims wake.
• Config [ci]:
  • enabled = true
  • max_log_mb = 16
  • max_artifact_mb = 512
  • max_artifacts_per_job = 50
  • log_retention_days = 90
  • artifact_retention_days = 30
  • registration_token_hours = 24

4. forge-web: runner API (worker W4), crates/forge-web/src/ci/api.rs

All responses are JSON (forge_ci::proto). Runner auth is Authorization: Bearer <runner_secret>. Job auth is Authorization: Bearer <job_token> on /jobs/{id}/…, checked with job_by_token.

• POST /api/v1/ci/runners/register takes RegisterRequest and returns RegisterResponse.
• POST /api/v1/ci/runners/claim takes ClaimRequest. It loops claim_job; if nothing is found, it awaits ci_waker or 25s, then tries once more. It returns 200 with a JobAssignment, or 204.
  • The labels in the request are ignored; the stored labels are used.
  • Disabled runners get 403.
  • Secrets come from ci_secret_values(repo, spec.secrets), except for runs whose trigger is pull_request from a fork (always empty there).
  • clone_url is the repo's HTTPS clone URL.
• POST /api/v1/ci/jobs/{id}/logs: the body is a raw text chunk (≤ 1 MiB), with header X-Forge-Log-Offset: <byte offset>.
  • It appends to <data_dir>/ci/logs/<job id>.log only if the offset equals the current size (idempotent retry: an equal end returns 200 without writing; a mismatch returns 409 with X-Forge-Log-Size).
  • The server masks again with Masker built from the job's secret values before writing.
  • The cap is max_log_mb. Past it, the server writes "[log truncated]" once and drops the rest.
  • It calls add_log_bytes.
• POST /api/v1/ci/jobs/{id}/heartbeat returns HeartbeatResponse (cancel_requested).
• PUT /api/v1/ci/jobs/{id}/artifacts/{name} streams the raw body to the BlobStore (crate::blobs, key ci/<artifact id>), with a sha256 and the max_artifact_mb cap.
  • The name must be [A-Za-z0-9._+-]{1,128} and must not start with ".".
  • At most max_artifacts_per_job.
• POST /api/v1/ci/jobs/{id}/finish takes FinishRequest and calls finish_job. It syncs statuses for this job and for every job that became skipped or queued, and notifies ci_waker.
• Git read with a job token: smart-HTTP authenticate in git_http.rs also accepts Basic auth with username ci-job and password <job_id>:<job_token>. That gives read-only access to exactly that job's repo while the job is Running. Every write is refused.
• Compression is excluded for /api/v1/ci/ paths (add the path to compression.rs).

5. forge-web UI: runs (worker W5), crates/forge-web/src/pages/ci_runs.rs, ci_job.rs

• "CI" repo tab.
  • /{o}/{r}/ci lists runs, newest first: number, state icon, trigger, ref, short commit, actor, and age.
  • A "Run workflow" form (Write, manual allowed) offers a branch select.
• /{o}/{r}/ci/runs/{n} shows the jobs table with state, duration and runner name, plus config_error in a <pre>.
  • Buttons, each a POST with CSRF: Cancel (Write), Re-run (Write), and "Approve and run" (Write, only in WaitingApproval).
• /{o}/{r}/ci/jobs/{job_id} shows the job log: the last 512 KiB of the log file in <pre>, with a link to the full log.
  • No JS: while the job is not terminal, the page has <meta http-equiv="refresh" content="3">.
  • Artifacts are listed with download links.
• /{o}/{r}/ci/jobs/{job_id}/log.txt serves text/plain; charset=utf-8, X-Content-Type-Options: nosniff, and streams live. It sends the current file, then keeps following it (poll every 1s) until the job is terminal and the file is fully sent. This is the live log; a browser shows it progressively.
• /{o}/{r}/ci/jobs/{job_id}/artifacts/{name} downloads as an attachment, with nosniff.
• Read access follows repo Read. Logs of private repos are never on the user-content domain.
• The commit list and PR page already show statuses through the status API, so check that the ci/* contexts render there.

6. forge-web UI: secrets and runners (worker W6)

• /{o}/{r}/settings/secrets (Admin):
  • lists names and when each was updated, never the values;
  • has an add/update form (name plus value textarea; the value is write-only);
  • deletes with CSRF;
  • records audit events ci_secret_set and ci_secret_deleted (name only).
• /admin/runners (instance admin):
  • lists runners with labels, last seen and disabled state;
  • has disable/enable and delete buttons;
  • has "Create registration token", which shows the token once along with the exact command forge-runner register --url <base_url> --token <token> --name <name> --labels linux;
  • records audit events ci_runner_registered, ci_runner_deleted and ci_registration_token_created.

7. forge-runner (worker W7)

• CLI:
  • forge-runner register --url URL --token T --name N [--labels a,b] [--config PATH] writes the config (TOML, 0600, default ~/.config/forge-runner/runner.toml): url, runner_id, runner_secret, labels, work_dir, concurrency = 1, and the [limits] memory = "4g", cpus = "2" and pids = 4096 defaults.
  • forge-runner run [--config PATH] loops claim → execute → finish, with concurrency jobs in parallel.
• Executor trait:

#[async_trait-free: use Pin<Box<dyn Future>> or an enum] 
trait Executor { async fn run(&self, job: &JobAssignment, workspace: &Path, log: &mut LogSink, cancel: CancellationToken) -> Outcome+exit code }

• PodmanExecutor runs podman run --rm --name forge-job-<job_id> --pull=newer --network=<slirp4netns|none> --memory M --cpus C --pids-limit P --security-opt no-new-privileges --cap-drop ALL -v <ws>:/workspace:Z -w /workspace -e NAME... <image> sh -eu -c <run>.
  • Env and secret values are passed through the child's environment with -e NAME (name only), never in argv.
  • On cancel or timeout it runs podman kill forge-job-<id>, then podman rm -f.
  • The podman argv construction is unit-tested.
• HostExecutor exists only for tests, behind #[cfg(any(test, feature = "test-host-executor"))]. It runs sh -eu -c in the workspace with the env, and has no isolation.
• Workspace: a fresh work_dir/<job_id>. Run git init, then git fetch --depth=1 <clone_url> <commit> with GIT_CONFIG_* env carrying http.extraHeader: Authorization: Basic base64(ci-job:<job_id>:<token>), never in argv, with GIT_TERMINAL_PROMPT=0. Then git checkout --detach <commit>. The workspace is deleted afterwards.
• Logs:
  • stdout and stderr are merged line-wise and prefixed with nothing.
  • They pass through MaskStream and are batched (≤ 64 KiB or 1s), then POSTed with the offset. On 409 the runner resyncs to X-Forge-Log-Size.
  • The first lines are runner info: runner name, image, and commit.
• Heartbeat every 10s; cancel: true triggers the cancellation token. The timeout is timeout_minutes.
• Artifacts: after a successful run, expand the globs inside the workspace (the canonicalized path must stay inside it, symlinks must not escape, and only regular files count). Upload each file under its relative path flattened with / → _, respecting the limits.
• Finish: report Success on exit 0, Failure on a non-zero exit, Cancelled, and Error for runner or podman failures, with a message.
• Tests: a mock forge (a small hyper or tokio TCP server in tests) plus HostExecutor cover:
  • claim → logs → artifacts → finish;
  • masking;
  • cancel;
  • timeout;
  • 409 resync.
  • Real podman only in an #[ignore] test.

8. End to end and docs (worker W8, after W1–W7)

• crates/forge-web/tests/ci_e2e.rs:
  1. Boot the forge and register a runner through the API.
  2. Push a repo with .forge/ci.toml (two jobs, with needs, a secret and an artifact).
  3. Drive the job with forge-runner's library API and HostExecutor (forge-runner exposes lib.rs).
  4. Assert the commit statuses, the masked secret in the log, the artifact download, the skip on failure, and that a fork PR waits for approval.
• docs/ci.md:
  • the format reference with examples;
  • runner install (rootless podman setup: subuid/subgid, loginctl enable-linger, a systemd user unit);
  • the security model (fork PRs, secrets, masking limits);
  • the runner protocol.