Function runPipeline

packages/pi-plugin/src/context-handler.ts:3429–4490

(args: RunPipelineArgs)

}

async function runPipeline(args: RunPipelineArgs): Promise<RunPipelineResult> {
	let executedWorkThisPass = false;
	let historyWasConsumedThisPass = false;
	let materializationSatisfiedThisPass = false;
	let pendingOpsAppliedThisPass = false;
	let pendingOpsDidMutate = false;
	let heuristicOrReasoningDidMutate = false;
	let didMutateFromFlushedStatuses = false;
	let droppedCount = 0;
	const droppedTokens = 0;
	let emergency = false;
	let autoReclaimDidMutateThisPass = false;
	let suppressDeferredHistoryDrain = false;
	let casLost = false;
	const deferredHistoryWasPendingAtPassStart =
		deferredHistoryRefreshSessions.has(args.sessionId);

	// 0. Inject temporal `<!-- +Xm -->` markers into user messages
	// BEFORE tagging so the §N§ tag prefix wraps around our marker on
	// re-tagging. Idempotent: existing markers are detected by regex
	// and skipped. Same invariants as OpenCode's `injectTemporalMarkers`
	// at transform.ts:648 — runs on every pass, deterministic from
	// timestamps, retroactive when the flag flips.
	if (args.temporalAwareness) {
		const tTemporal = performance.now();
		try {
			const injected = injectPiTemporalMarkers(args.messages);
			if (injected > 0) {
				sessionLog(
					args.sessionId,
					`temporal-awareness: injected ${injected} gap markers`,
				);
			}
		} catch (err) {
			sessionLog(
				args.sessionId,
				`temporal-awareness failed (continuing): ${err instanceof Error ? err.message : String(err)}`,
			);
		}
		logTransformTiming(args.sessionId, "injectTemporalMarkers", tTemporal);
	}

	// Pass entryIds so the transcript tags each message under its real SessionEntry
	// id (position-independent) instead of the index-based pi-msg-* id that drifts
	// when the visible array shifts (compaction trim / custom_message inserts),
	// orphaning tags/source_contents/caveman/drop-state. Positional entryIds is
	// exactly aligned here: tagging runs at transcript-build time, before any splice.
	const transcript = createPiTranscript(
		args.messages,
		args.sessionId,
		args.entryIds,
	);
	// Reasoning clearing/replay mutate `part.thinking` in place. They MUST target
	// the transcript's `working` array (the channel commit() flushes), not the
	// original `args.messages`: tagging/drops/caveman reassign working[idx] to
	// fresh objects, so a reasoning mutation written to args.messages[idx] (a
	// now-divergent object) would be discarded when commit() does
	// source[idx] = working[idx], leaving the cleared-reasoning watermark ahead