feat(runner): report a run.flow that resolves in no loaded skill tier (#189)

The trigger's run.flow reached the container only as prompt prose, and pi
never matches prose against loaded skill names, so a flow that materialised
in no tier (repo, injected, overlay or staged package) ran to a clean
exit 0 without the procedure it was written for and reported success for
work it could not have done.

The worker now forwards the flow structurally as PI_FLOW (omitted when the
job carries none, never an empty string; env and not event.json, because an
execution knob is not a fact about the delivery), and the runner compares it
against the skill set the loader actually materialised, immediately after
resource load and before any session or spend, emitting one flow_not_loaded
line on a miss. Report, not refusal, deliberately: run.flow is by long
doctrine a prompt hint, and a refusal shipped in an image upgrade would fail
yesterday's jobs for a value the reviewed file has carried all along. The
check sits at the pre-spend moment, so flipping report to refusal later is a
one-line change at the same site.

getSkills() moves out of the packages-only guard so the check runs for every
job. isFlowLoaded is exact name equality against the LOADED names (pi names
a skill frontmatter name || parent dir at the pin, now pinned by a loader
test), and a disableModelInvocation skill counts as loaded: it is absent
from the catalogue but invocable, so it is not the silent no-op this line
exists to catch.

Specs: INT-CONTAINER-JOB-INPUTS AMENDED (PI_FLOW), REQ-PER-TRIGGER-SKILLS
AMENDED (the runner acceptance clause), NEW
DES-FLOW-RESOLUTION-TWO-ADVISORY-LAYERS (report-not-refuse and the rejected
alternatives: boot-time failure, event.json carriage, a new top-level
outcome). UNCHANGED, checked: INT-RUNNER-EXIT-CODE-PROTOCOL,
INT-TRIGGERS-FILE-CONTRACT, INT-WEBHOOK-PAYLOAD-SUBSET,
REQ-GLOBAL-PI-OVERLAY, REQ-DEPLOYMENT-BOOTSTRAP, DES-AI-TRIGGER-FLOW-GATE,
DES-TRIGGERS-UNIFIED-FILE.

Signed-off-by: Rob Boerman <robboerman@live.nl>

Author: edgehero

image/runner/run-job.mjs

@@ -15,7 +15,7 @@ import {
 	decideExit,
 	EXIT_INFRA,
 } from "./src/outcome.mjs";
-import { countPackageResources, findShadowedSkills, owningRoot } from "./src/packages.mjs";
+import { countPackageResources, findShadowedSkills, isFlowLoaded, owningRoot } from "./src/packages.mjs";
 import { openSessionManager } from "./src/session.mjs";
 import { attachTokenBudget } from "./src/token-budget.mjs";
 import { attachTurnBudget } from "./src/turn-budget.mjs";
@@ -107,9 +107,28 @@ async function main() {
 		log,
 	});
 
-	if (cfg.packages.length > 0) {
-		const { skills, diagnostics } = resourceLoader.getSkills();
+	// Read ONCE, unconditionally: the flow check below needs the loaded set whether or not packages
+	// are staged, and the packages diagnostics block reuses the same bindings.
+	const { skills, diagnostics } = resourceLoader.getSkills();
+
+	// REPORT a flow that resolved in NO tier (issue #189). The trigger's run.flow reaches the model
+	// as prompt prose, and pi never matches prose against loaded skill names, so without this line a
+	// flow that materialised nowhere -- repo, injected, overlay or staged package -- runs to a clean
+	// exit 0 without the procedure it was written for and reports success for work it could not have
+	// done. That is the exact outcome assertPackagePathsExist refuses for an unmounted package root,
+	// and the deliberate difference is that this one only REPORTS: run.flow is by long doctrine a
+	// prompt hint (prepare.mjs), deployments legitimately run flows as loose hints over repos with no
+	// .pi/skills, and the runner cannot tell that steady state from breakage. Refusing would break
+	// them on an image upgrade for a value their reviewed file has carried all along. The line sits
+	// at the pre-spend moment anyway, so flipping report to refusal is a one-line change here plus a
+	// spec row (DES-FLOW-RESOLUTION-TWO-ADVISORY-LAYERS records the choice). Doctor's host-side tier
+	// lines are the other advisory layer; this one is exact because it reads what actually loaded.
+	// Flow name, never task content: run.flow is operator config out of the reviewed triggers file.
+	if (!isFlowLoaded(cfg.flow, skills)) {
+		log("flow_not_loaded", { flow: cfg.flow, skills: skills.length });
+	}
 
+	if (cfg.packages.length > 0) {
 		// REPORT a staged package skill that TRIED to shadow a repo or operator-overlay skill.
 		//
 		// pi builds skillPaths as mergePaths(cliEnabledSkills, additionalSkillPaths) -- package paths

image/runner/src/config.mjs

@@ -33,6 +33,10 @@ export function parseRunnerEnv(env) {
 		// path under the per-job /session mount. `null` when the trigger did not arm run.resume, which is
 		// the default and is byte-identical to every job before the feature existed.
 		sessionFile: parseSessionFile(env, "PI_SESSION_FILE"),
+		// INT-CONTAINER-JOB-INPUTS (issue #189): the trigger's run.flow, structurally, so run-job can
+		// compare it against the loaded skill names. `null` when the job carries no flow (a bare
+		// run.task cron job), which skips the check entirely.
+		flow: parseFlowName(env, "PI_FLOW"),
 		retry: {
 			maxRetries: parsePositiveInt(env, "PI_RETRY_MAX", 2),
 			baseDelayMs: parsePositiveInt(env, "PI_RETRY_BASE_MS", 2000),
@@ -132,6 +136,25 @@ function parsePackagePaths(env, name) {
 	return paths;
 }
 
+/**
+ * Parse the flow name the worker forwarded (issue #189). Unset or empty is `null` -- a job whose
+ * trigger names no flow has nothing to verify, and that is the normal state for a bare run.task
+ * cron job, not a misconfiguration.
+ *
+ * Deliberately NO charset validation here, unlike every sibling parser above. The value is used for
+ * exactly two things -- name-equality against pi's loaded skill set and one log field -- and is
+ * never interpolated into a path or a shell word, so a strange name cannot escape anything. Refusing
+ * a shape the worker's own validator accepted (parseTriggers pins run.flow to a non-empty string,
+ * nothing narrower) would mean an image upgrade starts failing jobs that ran yesterday, for a value
+ * the operator's reviewed file has carried all along. The comparison simply misses and the miss is
+ * reported, which is this variable's whole purpose.
+ */
+function parseFlowName(env, name) {
+	const raw = env[name];
+	if (raw === undefined || raw === "") return null;
+	return raw;
+}
+
 /**
  * Parse the persisted-session path (INT-SESSION-STORE-CONTRACT).
  *

image/runner/src/packages.mjs

@@ -185,6 +185,26 @@ export function countPackageResources({ packageRoots = [], extensionPaths = [],
 	}));
 }
 
+/**
+ * Does the trigger's flow name any skill pi actually loaded? (Issue #189.)
+ *
+ * Exact name equality against the LOADED set, which makes this the one check in the system that is
+ * not an approximation: doctor probes tier directories host-side and a dir name can lie (pi names a
+ * skill `frontmatter.name || parentDirName` at the 0.80.7 pin, skills.js:221), but here the names
+ * come off the skills the loader materialised for THIS job, after every tier and override.
+ *
+ * A `disableModelInvocation` skill still counts as loaded -- it is absent from the system-prompt
+ * catalogue but present in the session, invocable as /skill:name, so the flow it names is not the
+ * silent no-op this check exists to catch.
+ *
+ * `flow` null/empty returns true: no flow, nothing to verify, no line. Malformed skill entries
+ * (no `name`) are skipped rather than crashing a job over a diagnostic input.
+ */
+export function isFlowLoaded(flow, skills) {
+	if (flow === null || flow === undefined || flow === "") return true;
+	return (skills ?? []).some((skill) => skill?.name === flow);
+}
+
 /**
  * Containment by path SEGMENT, not by string prefix -- `/opt/pi-global/packages/tool` must not
  * claim a path under `/opt/pi-global/packages/tools`. Trailing slashes on a root are tolerated

image/runner/test/compose.test.mjs

@@ -98,3 +98,24 @@ test("run-job.mjs staples `usage` onto the success-path exit line -- worker pars
 	// The catch-path exit line legitimately omits `usage` for the same reason it omits turns: no meter
 	// exists when a preflight throw kills the run before any session started.
 });
+
+test("run-job.mjs verifies the flow against the LOADED skill set, unconditionally and pre-spend", () => {
+	// Same source-guard tactic as the turns/usage pins above (main() self-runs on import, log is
+	// unexported). Three orderings pinned, each of which failed silently before issue #189:
+	// (1) getSkills() sits OUTSIDE the packages guard -- inside it, the flow check would only run for
+	//     jobs with staged packages, which is exactly the blindness being closed;
+	// (2) the flow_not_loaded line exists and carries the flow -- a flow that resolves in no tier
+	//     must leave a named, greppable trace, so "a silent exit 0 for this case" is a failing test;
+	// (3) the check sits before openSessionManager -- the pre-spend moment, so flipping report to
+	//     refusal (DES-FLOW-RESOLUTION-TWO-ADVISORY-LAYERS) stays a one-line change at this site.
+	const src = readFileSync(new URL("../run-job.mjs", import.meta.url), "utf8");
+	assert.ok(
+		src.indexOf("resourceLoader.getSkills()") < src.indexOf("if (cfg.packages.length"),
+		"getSkills() must be read before (outside) the packages guard",
+	);
+	assert.match(src, /log\("flow_not_loaded",\s*\{[^}]*flow:/, "the miss must leave a named line carrying the flow");
+	assert.ok(
+		src.indexOf('log("flow_not_loaded"') < src.indexOf("openSessionManager("),
+		"the flow check must sit at the pre-spend moment",
+	);
+});

image/runner/test/config.test.mjs

@@ -209,3 +209,17 @@ test("PI_SESSION_FILE refuses a relative path, a .. segment, and anything under
 	// A path merely NAMED like the workspace is fine -- the check is on the path boundary, not a prefix.
 	assert.equal(parseRunnerEnv({ ...base, PI_SESSION_FILE: "/workspace-sessions/s.jsonl" }).sessionFile, "/workspace-sessions/s.jsonl");
 });
+
+test("PI_FLOW is optional: unset or empty is null, so a flowless job is byte-identical to today", () => {
+	assert.equal(parseRunnerEnv(base).flow, null);
+	assert.equal(parseRunnerEnv({ ...base, PI_FLOW: "" }).flow, null);
+});
+
+test("PI_FLOW parses to the exact string, with no charset opinion", () => {
+	assert.equal(parseRunnerEnv({ ...base, PI_FLOW: "review" }).flow, "review");
+	// Deliberately accepted: the value is only compared against loaded skill names and logged, never
+	// interpolated into a path or a shell word, and refusing a shape the worker's own validator
+	// accepted (parseTriggers pins non-empty string, nothing narrower) would start failing
+	// yesterday's jobs on an image upgrade. The comparison simply misses, and the miss is the report.
+	assert.equal(parseRunnerEnv({ ...base, PI_FLOW: "Not A Skill Name" }).flow, "Not A Skill Name");
+});

image/runner/test/loader.test.mjs

@@ -743,6 +743,30 @@ test("a staged skill whose name collides with nothing is left completely alone",
 	assert.equal(skill.filePath, join(pkg, "skills", "pkg-skill", "SKILL.md"));
 });
 
+test("pi names a skill frontmatter `name` || its parent dir -- the premise the flow check compares against", { skip }, async () => {
+	// isFlowLoaded (issue #189) does exact equality against these loaded names, and doctor's
+	// host-side tier probes approximate them by DIR name. This pins the naming rule at the pin
+	// (loadSkill: frontmatter.name || basename(dirname)), so a pi bump that changes it fails here
+	// rather than silently turning the runner's check, or doctor's ✓, into a lie.
+	const f = fixture();
+	// No `name:` in frontmatter -> the parent DIR is the name.
+	mkdirSync(join(f.jobPi, "skills", "dir-named"), { recursive: true });
+	writeFileSync(join(f.jobPi, "skills", "dir-named", "SKILL.md"), "---\ndescription: named by its dir\n---\n\nSteps.\n");
+	// Frontmatter `name:` wins over the dir -- the rename case a dir-name probe cannot see.
+	mkdirSync(join(f.jobPi, "skills", "some-dir"), { recursive: true });
+	writeFileSync(join(f.jobPi, "skills", "some-dir", "SKILL.md"), "---\nname: renamed\ndescription: named by frontmatter\n---\n\nSteps.\n");
+	const loader = await loaderModule.buildLoadedResourceLoader({
+		cwd: f.workspace,
+		jobPiDir: f.jobPi,
+		guardrailsPath: f.guardrailsPath,
+		outboxProtocolPath: f.outboxProtocolPath,
+	});
+	const names = loader.getSkills().skills.map((s) => s.name);
+	assert.ok(names.includes("dir-named"), `no frontmatter name -> the dir names the skill; got ${JSON.stringify(names)}`);
+	assert.ok(names.includes("renamed"), "a frontmatter name must win over the dir name");
+	assert.ok(!names.includes("some-dir"), "the renamed skill's dir must NOT also be a loaded name");
+});
+
 // --- enforceProtectedSkillPrecedence, decided on injected input (no skills tree, no collisions) ---
 
 /** A Skill-shaped record; only name and filePath are load-bearing for the precedence decision. */

image/runner/test/packages.test.mjs

@@ -6,6 +6,7 @@ import {
 	countPackageResources,
 	extensionEntryName,
 	findShadowedSkills,
+	isFlowLoaded,
 	isUnderAnyRoot,
 	owningRoot,
 	partitionAdminExtensions,
@@ -313,3 +314,35 @@ test("nothing admin-like means nothing dropped, and no input shape throws", () =
 	// Tools may arrive as plain names too -- the decision must not depend on pi's container type.
 	assert.equal(adminExtensionReason({ path: WORKSPACE_EXT, tools: ["dispatch_run"] }), "admin-tools");
 });
+
+// --- isFlowLoaded (issue #189): the flow-resolves-somewhere check, against the LOADED set ---
+
+test("a flow that names a loaded skill is loaded, and a near-miss is not", () => {
+	const skills = [{ name: "review", filePath: "/job/pi/skills/review/SKILL.md" }, { name: "deploy" }];
+	assert.equal(isFlowLoaded("review", skills), true);
+	// Exact equality, no normalisation: pi's own catalogue is exact-name, so a check that fuzzed
+	// ("revieww", case folds) would report loaded for a skill pi will never surface.
+	assert.equal(isFlowLoaded("revieww", skills), false);
+	assert.equal(isFlowLoaded("Review", skills), false);
+});
+
+test("no flow means nothing to verify: null, undefined and empty all pass without a report", () => {
+	// A bare run.task cron job carries no flow. That is the normal state, not a miss -- reporting it
+	// would teach operators to ignore the line that matters.
+	for (const flow of [null, undefined, ""]) {
+		assert.equal(isFlowLoaded(flow, []), true, `flow ${JSON.stringify(flow)} must not report`);
+	}
+});
+
+test("a disableModelInvocation skill still counts as loaded", () => {
+	// It is absent from the system-prompt catalogue but present in the session (invocable as
+	// /skill:name), so the flow it names is not the silent no-op this check exists to catch.
+	assert.equal(isFlowLoaded("quiet", [{ name: "quiet", disableModelInvocation: true }]), true);
+});
+
+test("malformed skill entries and an absent list are skipped, never a crash", () => {
+	// This runs inside every job; a diagnostic input must not take the job down with it.
+	assert.equal(isFlowLoaded("review", [null, {}, { filePath: "/x" }]), false);
+	assert.equal(isFlowLoaded("review", undefined), false);
+	assert.equal(isFlowLoaded("review", []), false);
+});