diff --git a/apps/server/src/plugins/Workflows.testkit.ts b/apps/server/src/plugins/Workflows.testkit.ts
index 09d275452d6f..7aed770c4295 100644
--- a/apps/server/src/plugins/Workflows.testkit.ts
+++ b/apps/server/src/plugins/Workflows.testkit.ts
@@ -52,6 +52,8 @@ export const fixture = Effect.gen(function* () {
let head = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let dirty = false;
let providerAvailable = true;
+ let forgeAvailable = true;
+ let redactions = 0;
let lostAcknowledgement = false;
let checkExecution = Effect.succeed({
exitCode: 0,
@@ -68,7 +70,11 @@ export const fixture = Effect.gen(function* () {
const host = Host.of({
environmentId,
cancelPending: () => Effect.void,
- redact: (input) => Effect.succeed(input.text),
+ redact: (input) =>
+ Effect.sync(() => {
+ redactions++;
+ return input.text;
+ }),
projects: () =>
Effect.succeed([{ id: projectId, title: "Workflow", workspaceRoot: directory }]),
providers: () =>
@@ -94,7 +100,19 @@ export const fixture = Effect.gen(function* () {
yield* Queue.offer(checkStarts, input.command);
return yield* checkExecution;
}),
- verifyPullRequestHead: () => Effect.succeed({ head, branch: "feature" }),
+ verifyPullRequestHead: () =>
+ Effect.suspend(() =>
+ forgeAvailable
+ ? Effect.succeed({ head, branch: "feature" })
+ : Effect.fail(
+ new PluginError({
+ pluginId: "host",
+ code: "unavailable",
+ operation: "verify-pull-request-head",
+ message: "The forge is unreachable.",
+ }),
+ ),
+ ),
receipt: (id) => Effect.succeed(receipts.get(id) ?? null),
retryPreparation: () =>
Effect.fail(
@@ -240,6 +258,18 @@ export const fixture = Effect.gen(function* () {
if (!Effect.isEffect(result)) return yield* Effect.die("Expected request");
return yield* result;
});
+ /** A subscription API on the currently running plugin (after any restart). */
+ const stream = (method: string, input: unknown) =>
+ Stream.unwrap(
+ Effect.suspend(() => runtime.registry.api(`plugins.workflows.${method}`)).pipe(
+ Effect.flatMap((api) => {
+ const result = api.invoke(input);
+ return Stream.isStream(result)
+ ? Effect.succeed(result)
+ : Effect.die("Expected a subscription");
+ }),
+ ),
+ );
const query = (runId: string) =>
invoke("get", { environmentId, projectId, runId }).pipe(Effect.flatMap(decodeRun));
const reconcile = invoke("reconcile", { environmentId, projectId }).pipe(Effect.asVoid);
@@ -276,6 +306,7 @@ export const fixture = Effect.gen(function* () {
query,
reconcile,
invoke,
+ stream,
report,
launches,
commands,
@@ -291,6 +322,10 @@ export const fixture = Effect.gen(function* () {
holdCheck: () => {
checkExecution = Effect.never;
},
+ /** Result the next check commands return. */
+ setCheck: (result: { exitCode: number; timedOut: boolean; stdout: string }) => {
+ checkExecution = Effect.succeed({ ...result, stderr: "" });
+ },
threads,
restart: Effect.gen(function* () {
yield* runtime.close;
@@ -327,6 +362,14 @@ export const fixture = Effect.gen(function* () {
setHead: (value: string) => {
head = value;
},
+ /** Display reads redact through the host; counting them shows when a view re-read. */
+ redactions: () => redactions,
+ /** Whether the forge can verify the pull request head. */
+ setForgeAvailable: (value: boolean) => {
+ forgeAvailable = value;
+ },
+ /** The host's attention stream of the currently running plugin (after any restart). */
+ hostAttention: Stream.unwrap(Effect.sync(() => runtime.registry.attention(environmentId))),
setDirty: (value: boolean) => {
dirty = value;
},
diff --git a/apps/server/src/plugins/WorkflowsAttentionIsolation.integration.test.ts b/apps/server/src/plugins/WorkflowsAttentionIsolation.integration.test.ts
new file mode 100644
index 000000000000..2562e58dc67a
--- /dev/null
+++ b/apps/server/src/plugins/WorkflowsAttentionIsolation.integration.test.ts
@@ -0,0 +1,197 @@
+import { expect, it } from "@effect/vitest";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Schema from "effect/Schema";
+import * as Stream from "effect/Stream";
+import { Host } from "@t3tools/plugin-host-contract/server";
+import { CommandId, EventId, MessageId, NodeId, RuntimeRequestId } from "@t3tools/contracts";
+import { AttentionPage, Run } from "@t3tools/plugin-workflows/contracts";
+import { sequence } from "./Workflows.testkit.ts";
+import { makeCoreWorkflowFixture } from "./WorkflowsCore.testkit.ts";
+
+const decodeRun = Schema.decodeUnknownEffect(Run);
+const decodePage = Schema.decodeUnknownEffect(AttentionPage);
+
+it.live(
+ "keeps workflow attention when its thread is read or snoozed and clears each request on its own answer",
+ () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const test = yield* makeCoreWorkflowFixture;
+ // The replayed provider reports its execution interrupted once the test says so; the
+ // thread, its requests, read state and snooze are real core state.
+ let interrupted = false;
+ const runtime = yield* test.boot(
+ Host.of({
+ ...test.core,
+ lifecycle: () => Stream.never,
+ inspect: (target) =>
+ test.core.inspect(target).pipe(
+ Effect.map((state) =>
+ interrupted
+ ? {
+ ...state,
+ runs: state.runs.map((run) => ({ ...run, status: "interrupted" })),
+ resultRunId: state.runs.at(-1)?.id ?? null,
+ outstandingWork: [],
+ checkpoints: [],
+ }
+ : state,
+ ),
+ ),
+ }),
+ );
+ const attention = runtime
+ .invoke("attention-page", test.scope)
+ .pipe(Effect.flatMap(decodePage));
+ const started = yield* runtime
+ .invoke("start", {
+ ...test.scope,
+ definition: sequence,
+ clientRequestId: "start",
+ input: {},
+ workspace: { type: "current" },
+ })
+ .pipe(Effect.flatMap(decodeRun));
+ yield* runtime.invoke("reconcile", test.scope);
+ const threadId = started.attempts[0]!.threadId!;
+ yield* test.threads.dispatch({
+ type: "message.dispatch",
+ threadId,
+ commandId: CommandId.make("deferred"),
+ messageId: MessageId.make("message"),
+ text: "Provider native execution",
+ attachments: [],
+ dispatchMode: { type: "defer_start" },
+ createdBy: "agent",
+ creationSource: "mcp",
+ });
+ const records = yield* test.threads.getProjectThreadRecords({ ...test.scope, threadId }, [
+ "runs",
+ ]);
+ const began = DateTime.nowUnsafe();
+ const native = { ...records.runs[0]!, status: "waiting" as const, startedAt: began };
+ // Two questions wait in the thread's native queue, answered in order.
+ const request = (id: string, offset: number) => ({
+ id: RuntimeRequestId.make(id),
+ nodeId: NodeId.make(`${id}-node`),
+ providerTurnId: null,
+ nativeRequestRef: null,
+ kind: "user_input" as const,
+ status: "pending" as const,
+ responseCapability: { type: "message" as const },
+ createdAt: DateTime.makeUnsafe(DateTime.toEpochMillis(began) + offset),
+ resolvedAt: null,
+ });
+ const approval = request("approval", 0);
+ const question = request("question", 1);
+ yield* test.sink.write({
+ events: [
+ {
+ id: EventId.make("native-wait-run"),
+ type: "run.updated",
+ threadId,
+ runId: native.id,
+ occurredAt: began,
+ payload: native,
+ },
+ ...[approval, question].map((item) => ({
+ id: EventId.make(`${item.id}-pending`),
+ type: "runtime-request.updated" as const,
+ threadId,
+ occurredAt: began,
+ payload: item,
+ })),
+ ],
+ });
+ yield* runtime.invoke("reconcile", test.scope);
+ const waiting = yield* attention;
+ expect(waiting.total).toBe(1);
+ const items = waiting.runs[0]!.items;
+ expect(
+ items.map((item) => [item.kind, item.threadId, item.request?.id, item.request?.position]),
+ ).toEqual([
+ ["needs-input", threadId, "approval", 1],
+ ["needs-input", threadId, "question", 2],
+ ]);
+
+ // Reading the thread is presentation; nothing workflow-owned moves. Native
+ // attention keeps its priority: a thread with a pending request cannot be snoozed.
+ const visit = (id: string) =>
+ test.threads.dispatch({
+ type: "thread.visit",
+ commandId: CommandId.make(id),
+ threadId,
+ visitedAt: DateTime.formatIso(DateTime.nowUnsafe()),
+ });
+ const snooze = (id: string) =>
+ test.threads.dispatch({
+ type: "thread.snooze",
+ commandId: CommandId.make(id),
+ threadId,
+ snoozedUntil: DateTime.formatIso(DateTime.add(DateTime.nowUnsafe(), { hours: 4 })),
+ });
+ yield* visit("visit");
+ expect((yield* Effect.result(snooze("snooze-refused")))._tag).toBe("Failure");
+ yield* runtime.invoke("reconcile", test.scope);
+ expect(yield* attention).toEqual(waiting);
+ const unchanged = yield* runtime
+ .invoke("get", { ...test.scope, runId: started.id })
+ .pipe(Effect.flatMap(decodeRun));
+ expect(unchanged.attempts[0]).toMatchObject({ phase: "waiting-input", report: null });
+
+ // Each item clears only on its own native resolution.
+ const answered = DateTime.nowUnsafe();
+ yield* test.sink.write({
+ events: [
+ {
+ id: EventId.make("approval-resolved"),
+ type: "runtime-request.updated",
+ threadId,
+ occurredAt: answered,
+ payload: { ...approval, status: "resolved", resolvedAt: answered },
+ },
+ ],
+ });
+ yield* runtime.invoke("reconcile", test.scope);
+ const remaining = (yield* attention).runs[0]!.items;
+ expect(remaining.map((item) => [item.id, item.request?.position])).toEqual([
+ [items[1]!.id, 1],
+ ]);
+ // The last answer arrives, then native execution is interrupted: the run stops for a person.
+ yield* test.sink.write({
+ events: [
+ {
+ id: EventId.make("question-resolved"),
+ type: "runtime-request.updated",
+ threadId,
+ occurredAt: answered,
+ payload: { ...question, status: "resolved", resolvedAt: answered },
+ },
+ ],
+ });
+ interrupted = true;
+ yield* runtime.invoke("reconcile", test.scope);
+ const stopped = yield* attention;
+ expect(stopped.runs[0]!.items).toEqual([
+ expect.objectContaining({
+ kind: "interrupted",
+ attemptId: started.attempts[0]!.id,
+ threadId,
+ request: null,
+ }),
+ ]);
+ // Without a pending request the thread can be snoozed; the workflow decision stays.
+ yield* snooze("snooze");
+ yield* visit("visit-again");
+ yield* runtime.invoke("reconcile", test.scope);
+ expect(yield* attention).toEqual(stopped);
+ expect(
+ (yield* runtime
+ .invoke("get", { ...test.scope, runId: started.id })
+ .pipe(Effect.flatMap(decodeRun))).state,
+ ).toBe("unresolved");
+ }),
+ ).pipe(Effect.provide(NodeServices.layer)),
+);
diff --git a/apps/server/src/plugins/WorkflowsDecisions.test.ts b/apps/server/src/plugins/WorkflowsDecisions.test.ts
new file mode 100644
index 000000000000..9add1e0dc951
--- /dev/null
+++ b/apps/server/src/plugins/WorkflowsDecisions.test.ts
@@ -0,0 +1,434 @@
+import { expect, it } from "@effect/vitest";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import * as Effect from "effect/Effect";
+import * as Schema from "effect/Schema";
+import { CatalogEntry, Definition, Run } from "@t3tools/plugin-workflows/contracts";
+import { completed, fixture, sequence } from "./Workflows.testkit.ts";
+
+const decodeRun = Schema.decodeUnknownEffect(Run);
+const decodeEntry = Schema.decodeUnknownEffect(CatalogEntry);
+const decodeDefinition = Schema.decodeUnknownEffect(Definition);
+const implement = sequence.nodes[0]!;
+
+/** Implement, run checks, then route on the recorded check result with one bounded repeat. */
+const checked = (maxVisits?: number) =>
+ decodeDefinition({
+ version: 1,
+ id: "checked",
+ revision: 1,
+ title: "Checked implementation",
+ entry: "implement",
+ atLimit: "review",
+ ...(maxVisits === undefined ? {} : { maxVisits }),
+ nodes: [
+ { ...implement, next: { to: "checks" } },
+ {
+ id: "checks",
+ kind: "check",
+ title: "Checks",
+ command: "project-check",
+ args: [],
+ next: { to: "decide" },
+ },
+ {
+ id: "decide",
+ kind: "decision",
+ title: "Check result",
+ source: "checks",
+ rules: [
+ {
+ when: {
+ op: "all",
+ terms: [
+ { op: "eq", path: "outcome", value: "completed" },
+ {
+ op: "any",
+ terms: [
+ { op: "eq", path: "exitCode", value: 0 },
+ { op: "eq", path: "timedOut", value: true },
+ ],
+ },
+ ],
+ },
+ route: { to: "review" },
+ },
+ ],
+ otherwise: { to: "implement", repeat: { max: 1, atLimit: "review" } },
+ },
+ {
+ id: "review",
+ kind: "human",
+ title: "Human review",
+ approve: { to: "done" },
+ changes: { to: "stopped" },
+ },
+ { id: "done", kind: "end", title: "Done", outcome: "completed" },
+ { id: "stopped", kind: "end", title: "Changes requested", outcome: "failed" },
+ ],
+ });
+
+it.effect(
+ "routes on recorded check results through an initial visit, one repeat and At limit, then a human decision",
+ () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const test = yield* fixture;
+ const scope = { environmentId: test.environmentId, projectId: test.projectId };
+ const definition = yield* checked();
+ expect(yield* test.invoke("validate", { ...scope, definition })).toMatchObject({
+ runnable: true,
+ problems: [],
+ });
+ test.setCheck({ exitCode: 1, timedOut: false, stdout: "1 test failed" });
+ const started = yield* test.start(definition);
+ const implementOnce = Effect.fnUntraced(function* (visit: number) {
+ yield* test.reconcile;
+ const run = yield* test.wait(
+ started.id,
+ (run) =>
+ run.attempts.filter((attempt) => attempt.nodeId === "implement").length === visit &&
+ run.attempts.at(-1)?.phase === "running",
+ );
+ const threadId = run.attempts.at(-1)!.threadId!;
+ yield* test.report(threadId, completed);
+ test.settle(threadId);
+ yield* test.reconcile;
+ });
+
+ // Initial visit: the failed check result drives the decision to its Otherwise repeat.
+ yield* implementOnce(1);
+ const repeated = yield* test.wait(started.id, (run) =>
+ run.trace.some((item) => item.nodeId === "decide"),
+ );
+ const check = repeated.attempts.find((attempt) => attempt.nodeId === "checks")!;
+ expect(check).toMatchObject({
+ phase: "failed",
+ check: { outcome: "failed", exitCode: 1, stdout: "1 test failed" },
+ });
+ expect(repeated.trace.find((item) => item.nodeId === "checks")).toMatchObject({
+ route: "next",
+ chosen: "decide",
+ sourceIds: [check.id],
+ });
+ expect(repeated.trace.find((item) => item.nodeId === "decide")).toMatchObject({
+ route: "otherwise",
+ chosen: "implement",
+ considered: [{ matched: false }],
+ reason: "Admitted a bounded repeat.",
+ repeatCount: 1,
+ repeat: { max: 1, atLimit: "review", exhausted: false },
+ });
+
+ // A restart keeps the counter; the repeated visit's failure exhausts it.
+ yield* test.restart;
+ yield* implementOnce(2);
+ const limited = yield* test.wait(started.id, (run) => run.state === "awaiting-review");
+ expect(limited.repeats).toEqual({ "decide:implement": 1 });
+ expect(limited.attempts.filter((attempt) => attempt.nodeId === "implement")).toHaveLength(
+ 2,
+ );
+ expect(limited.trace.at(-1)).toMatchObject({
+ nodeId: "decide",
+ route: "otherwise",
+ chosen: "review",
+ reason: "The repeat limit was reached.",
+ repeatCount: 1,
+ repeat: { max: 1, atLimit: "review", exhausted: true },
+ });
+ expect(limited).toMatchObject({
+ automationStopped: true,
+ gate: { nodeId: "review" },
+ allowedActions: ["cancel", "approve", "request-changes"],
+ });
+ yield* test.restart;
+ const reloaded = yield* test.query(started.id);
+ expect(reloaded.repeats).toEqual(limited.repeats);
+ expect(reloaded.trace).toEqual(limited.trace);
+
+ // The human decision follows its authored destination and is recorded with its route.
+ const decided = yield* test
+ .invoke("gate", {
+ ...scope,
+ runId: started.id,
+ clientRequestId: "changes",
+ expectedRevision: reloaded.revision,
+ decision: "request-changes",
+ })
+ .pipe(Effect.flatMap(decodeRun));
+ expect(decided).toMatchObject({ state: "failed", currentNode: "stopped" });
+ expect(decided.trace.at(-1)).toMatchObject({
+ nodeId: "review",
+ route: "changes",
+ chosen: "stopped",
+ reason: "Human decision: request-changes.",
+ });
+ }),
+ ).pipe(Effect.provide(NodeServices.layer)),
+);
+
+it.effect("takes the first matching nested rule on a passing check and approves to its end", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const test = yield* fixture;
+ const scope = { environmentId: test.environmentId, projectId: test.projectId };
+ test.setCheck({ exitCode: 0, timedOut: false, stdout: "ok" });
+ const started = yield* test.start(yield* checked());
+ yield* test.reconcile;
+ const running = yield* test.wait(started.id, (run) => run.attempts[0]?.phase === "running");
+ yield* test.report(running.attempts[0]!.threadId!, completed);
+ test.settle(running.attempts[0]!.threadId!);
+ yield* test.reconcile;
+ const review = yield* test.wait(started.id, (run) => run.state === "awaiting-review");
+ expect(review.trace.at(-1)).toMatchObject({
+ nodeId: "decide",
+ route: "rules.0",
+ chosen: "review",
+ considered: [{ matched: true }],
+ reason: "First matching rule.",
+ repeatCount: null,
+ });
+ expect(review.trace.at(-1)?.repeat).toBeUndefined();
+ expect(review).toMatchObject({ automationStopped: false, repeats: {} });
+ const approved = yield* test
+ .invoke("gate", {
+ ...scope,
+ runId: started.id,
+ clientRequestId: "approve",
+ expectedRevision: review.revision,
+ decision: "approve",
+ })
+ .pipe(Effect.flatMap(decodeRun));
+ expect(approved).toMatchObject({ state: "completed", currentNode: "done" });
+ expect(approved.trace.at(-1)).toMatchObject({ route: "approve", chosen: "done" });
+ }),
+ ).pipe(Effect.provide(NodeServices.layer)),
+);
+
+it.effect("sends the run to At limit when the whole-run visit bound is exhausted", () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const test = yield* fixture;
+ test.setCheck({ exitCode: 0, timedOut: false, stdout: "ok" });
+ // Implement and the check use both visits; the decision would be the third.
+ const started = yield* test.start(yield* checked(2));
+ yield* test.reconcile;
+ const running = yield* test.wait(started.id, (run) => run.attempts[0]?.phase === "running");
+ yield* test.report(running.attempts[0]!.threadId!, completed);
+ test.settle(running.attempts[0]!.threadId!);
+ yield* test.reconcile;
+ const limited = yield* test.wait(started.id, (run) => run.state === "awaiting-review");
+ expect(limited).toMatchObject({ automationStopped: true, currentNode: "review" });
+ expect(limited.trace.at(-1)).toMatchObject({
+ nodeId: "checks",
+ route: "next",
+ chosen: "review",
+ reason: "The whole-run visit limit was reached.",
+ });
+ expect(limited.trace.some((item) => item.nodeId === "decide")).toBe(false);
+ }),
+ ).pipe(Effect.provide(NodeServices.layer)),
+);
+
+it.effect(
+ "locates invalid operands, removed fields and repeat targets at their exact controls",
+ () =>
+ Effect.scoped(
+ Effect.gen(function* () {
+ const test = yield* fixture;
+ const scope = { environmentId: test.environmentId, projectId: test.projectId };
+ const valid = yield* checked();
+ const definition = yield* decodeDefinition({
+ ...valid,
+ nodes: valid.nodes.flatMap((node): ReadonlyArray Native thread view {countLabel}
+ {more} + {count === null ? "+" : ""} more +
+ )} +{description}
} + {children} + {footer} ++ Loading attention… +
+ ) : null + ) : latest.total === 0 ? ( ++ Newest {runs.length} of {latest.total} + {projectId === null ? " · open a run's project to see the rest" : ""} +
+ ) : latest !== null && (runs.length < latest.total || loading) ? ( +