Chapter 7: List Processing and Recursion

Core Concepts

An imperative loop is a mutable index advancing through a mutable array under a termination condition checked at runtime. It has three moving parts — the counter, the array pointer, and the termination predicate — all of which can be wrong independently. It allocates a stack frame for every nested call it makes within the body. And its correctness depends on the programmer maintaining a mental model of mutable state across every iteration.

Prolog has no loop syntax. It has recursion, unification, and the list structure [H|T]. These three primitives, composed correctly, produce the equivalent of map, filter, fold, any, all, and every higher-order list operation — without mutation, without a counter, and with a formal model of termination built into the data structure itself. The empty list [] is the base case. The non-empty list [H|T] is the recursive case. There is no third option.

Five properties define the Prolog list as a WAM engineering primitive.

1. A list is a nested pair on the Heap, not a contiguous array. The list [100, 101, 102] is not three integers stored adjacently. It is a chain of functor cells on the WAM Heap: each cell is a ./2 compound with a head value and a pointer to the next cell. The final cell's tail pointer points to the atom [] (nil). The cost of accessing the head of a list is one heap dereference. The cost of accessing the nth element is n dereferences. There is no O(1) random access. Infrastructure code that requires random access by index should use nth0/3 sparingly, or — better — model the data as a predicate with a key argument rather than as a positional list.

2. Recursion is Prolog's only iteration primitive. Every operation over a list in Prolog is a recursive predicate. The base case unifies with []. The recursive case unifies with [H|T], processes H, and recurses on T. The WAM allocates a new environment frame for each recursive call unless Tail-Call Optimisation (TCO) applies. Without TCO, processing a list of n elements costs O(n) Local Stack frames — identical to a naive recursive function in any language. TCO reduces this to O(1) when the recursive call is the last goal in the clause.

3. Tail-Call Optimisation is not optional for production workloads. A list of 10,000 VM IDs processed without TCO exhausts 10,000 Local Stack frames. With SWI-Prolog's default frame size of approximately 40 bytes, that is 400KB per batch operation — before any work is done. With TCO, the same predicate runs in a single reused frame regardless of list length. The accumulator pattern is the standard mechanism for enabling TCO: it converts a recursive predicate that computes a result on the way back up the call stack into one that carries the result forward through the recursion, making the recursive call the last goal.

4. Higher-order predicates are first-class batch operators. maplist/2, maplist/3, include/3, exclude/3, foldl/4 — these predicates from library(apply) abstract the recursion pattern entirely. maplist(Goal, List) applies Goal to every element of List. include(Pred, List, Filtered) retains only elements satisfying Pred. foldl(Goal, List, V0, V) threads an accumulator through List. Each of these is internally tail-recursive. Using them is not a style preference — it is the correct choice for readability, correctness, and stack safety simultaneously.

5. Untrusted list inputs are a DoS vector. A list arriving from findall/3, from a KB with a corrupted or adversarially-crafted assertion, or from an external source (Chapter 13's HTTP interface) has no guaranteed bound on its length. A recursive predicate that processes such a list without checking its length first is vulnerable to CPU exhaustion (very long list) and stack overflow (circular list / rational tree). The mandatory pattern: must_be(list, L), length(L, N), N =< MaxBound at the entry point of every recursive batch predicate. This is not a courtesy — it is the list-processing equivalent of the depth limit from Chapter 4's reachability tracer.

Chapter Roadmap

~~Section~~	~~Title~~	~~Focus~~
~~7.1~~	~~The Physics of~~ `[H\|T]`	~~WAM Heap list representation,~~ `./2` ~~functor, nil termination~~
~~7.2~~	~~Recursion and TCO~~	~~Environment frame cost, last-call optimisation, deterministic recursion~~
~~7.3~~	~~The Accumulator Pattern~~	~~Naive vs. accumulator recursion,~~ `total_cluster_ram/2`, `foldl/4`
~~7.4~~	~~The Build: Batch VM Command Generator~~	`include/3`, `maplist/3`, `atomic_list_concat/2`~~, constraint filter~~
~~7.5~~	~~Security Context: Recursion Limits~~	~~List length attack, rational tree DoS,~~ `must_be/2` + `length/2` ~~guard~~
~~Outcome~~	~~Relational Batching~~	~~Verification checklist, conceptual transition~~

7.1 The Physics of `[H|T]`

7.1.1 The `./2` Functor: What the Heap Actually Contains

Every Prolog list is syntactic sugar over the ./2 functor (dot-two, or in modern SWI-Prolog '[|]'/2). The list [100, 101, 102] is identical to:

'[|]'(100, '[|]'(101, '[|]'(102, [])))
% Or in traditional notation:
.(100, .(101, .(102, [])))

Each ./2 cell is a compound term on the WAM Heap occupying two consecutive words: the first word is a tagged pointer to the head value, the second is a tagged pointer to the tail (the next ./2 cell or []).

WAM Heap layout for [100, 101, 102]:

Address │ Tag  │ Value
────────┼──────┼───────────────────────────────
H₀      │ FUNC │ ./2  (list cell functor tag)
H₁      │ INT  │ 100  (head: integer 100)
H₂      │ REF  │ → H₃ (tail: pointer to next cell)
H₃      │ FUNC │ ./2
H₄      │ INT  │ 101
H₅      │ REF  │ → H₆
H₆      │ FUNC │ ./2
H₇      │ INT  │ 102
H₈      │ REF  │ → []  (nil atom — list terminator)

Three list cells, nine heap words. No contiguous integer array exists anywhere. The integer 101 is at heap address H₄ — reachable only by following the chain H₀ → H₂ → H₃ → H₅ → H₆ → H₄.

The practical consequences:

Head access is O(1): [H|_] unifies H with the value at the first word of the first cell — one dereference.
Tail access is O(1): [_|T] unifies T with the pointer at the second word — one dereference giving the rest of the list.
Nth element access is O(n): nth0(2, List, Elem) dereferences two list cells.
Length computation is O(n): length/2 walks every cell to the nil terminator.
Append is O(n) in the first argument: append([a,b,c], Rest, Combined) copies all three cells of the first list.

For the batch operation predicates in this chapter, the heap layout has one important implication: findall/3 producing a list of VM IDs allocates a new list cell on the Heap for each result. A findall/3 over a 10,000-VM KB allocates 10,000 list cells — 90,000 heap words — plus the integer cells for each ID. The list is garbage-collected when the enclosing clause exits if no references survive. Understanding the allocation cost is the prerequisite for understanding why the batch predicates in Section 7.4 are designed the way they are.

7.1.2 Unification with `[H|T]`

List processing predicates work by unifying their input with [H|T], which simultaneously:

Verifies the list is non-empty (would fail on [])
Binds H to the head element
Binds T to the tail (the remaining list, which may be [])

% Manual unification anatomy:
?- [H|T] = [100, 101, 102].
H = 100,
T = [101, 102].

?- [H|T] = [100].
H = 100,
T = [].   % Tail of a single-element list is nil

?- [H|T] = [].
false.   % [] does not unify with [H|T] — base case vs. recursive case

% Nested pattern matching:
?- [H1, H2 | T] = [100, 101, 102, 103].
H1 = 100, H2 = 101, T = [102, 103].

The [H|T] pattern is the only mechanism Prolog needs to implement all list operations. There is no array indexing, no pointer arithmetic, no bounds check to write. The structure of the data enforces the structure of the computation.

7.1.3 Diagram: WAM Heap List Visualisation

%%{init: {"themeVariables": {"fontSize": "18px"}}}%%
flowchart LR
    subgraph HEAP["WAM Heap — List [100, 101, 102]"]
        CELL1["Cell H₀–H₂\n./2\nHead: INT 100\nTail: → H₃"]
        CELL2["Cell H₃–H₅\n./2\nHead: INT 101\nTail: → H₆"]
        CELL3["Cell H₆–H₈\n./2\nHead: INT 102\nTail: → NIL"]
        NIL["[] Atom\n(Nil — list terminator)\nAddress: constant"]
    end

    CELL1 --->|"Tail pointer\n(H₂ → H₃)"| CELL2
    CELL2 --->|"Tail pointer\n(H₅ → H₆)"| CELL3
    CELL3 --->|"Tail pointer\n(H₈ → [])"| NIL

    VARLIST["Variable: VMList\npoints to H₀"]
    VARLIST --->|"dereference"| CELL1

    style HEAP fill:#1A2B4A,color:#FFFFFF
    style CELL1 fill:#1A4070,color:#FFFFFF
    style CELL2 fill:#1A4070,color:#FFFFFF
    style CELL3 fill:#1A4070,color:#FFFFFF
    style NIL fill:#4A235A,color:#FFFFFF
    style VARLIST fill:#1A6B3A,color:#FFFFFF

Reading the diagram: The variable VMList holds a tagged heap pointer to H₀, the first list cell. Each cell contains a head value (integer) and a tail pointer to the next cell. The chain terminates when a tail pointer reaches the nil atom []. To access VM ID 102, the engine must dereference three cells — there is no shortcut.

7.2 Recursion and Tail-Call Optimisation

7.2.1 Why There Are No Loops

A Prolog clause is a logical statement of the form "head is true if body is true." There is no imperative sequencing, no loop syntax, no mutable iteration variable. The only way to express "do something for every element of a list" is to define a predicate whose recursive structure mirrors the list's recursive structure:

% Base case: nothing to do for the empty list
process_vms([]).

% Recursive case: process the head, then recurse on the tail
process_vms([VMID | Rest]) :-
    do_something_with(VMID),
    process_vms(Rest).

This is not a limitation. The recursive structure eliminates entire categories of bugs: off-by-one errors (the base case unifies with [], not length(List) == 0), boundary conditions (the pattern [H|T] fails cleanly on []), and mutable state corruption (there is no counter to mutate). The trade-off — in naive recursion — is stack depth.

7.2.2 The Stack Cost of Naive Recursion

Every recursive call in Prolog that is not the last goal in its clause requires the WAM to allocate a new environment frame on the Local Stack before the call. The frame stores:

The continuation pointer (where to return after the call succeeds)
The current clause's permanent variables (those that survive across multiple goals)
The pointer to the parent environment frame

For a naive list-length predicate:

% Naive length — NOT tail-recursive
list_length([], 0).
list_length([_|T], N) :-
    list_length(T, N1),   % Recursive call is NOT the last goal
    N is N1 + 1.          % This goal executes AFTER the recursive call returns
                           % so an environment frame must be kept for every level

For a list of n elements, this allocates n environment frames before any N is N1 + 1 computation begins. The frames are released on the way back up the call chain. Peak Local Stack depth: O(n) frames.

Local Stack at maximum depth for list_length([100, 101, 102], N):

Frame 3 (deepest):  list_length([102], N1_2)  — waiting for N1_1 + 1
Frame 2:            list_length([101, 102], N1_1)  — waiting for N1_0 + 1
Frame 1:            list_length([100, 101, 102], N)  — waiting for N0 + 1
                    (base case: N2 = 0 found at Frame 4, now unwinding)

For a list of 50,000 elements — well within the range of a Proxmox cluster's complete VM inventory — naive recursion allocates 50,000 frames before processing a single element.

7.2.3 Tail-Call Optimisation: Last-Call Reuse

Tail-Call Optimisation (TCO) applies when the recursive call is the last goal in a clause — there is nothing to do after it returns. In this case, the WAM can reuse the current environment frame for the recursive call instead of allocating a new one. The "return address" for the current frame is simply replaced with the continuation for the recursive call. Local Stack depth remains constant throughout the recursion: O(1) regardless of list length.

The WAM implements this as the last-call optimisation (LCO): before executing the last goal in a clause, if the clause has no remaining permanent variables needed after that goal, the environment frame is deallocated first, and the last call executes in the parent's frame. Combined with deterministic predicates (no open choice points), this is full tail recursion.

% Tail-recursive length using an accumulator
list_length_acc([], Acc, Acc).   % Base case: return accumulated count
list_length_acc([_|T], Acc, Length) :-
    Acc1 is Acc + 1,
    list_length_acc(T, Acc1, Length).  % LAST GOAL — TCO applies
                                        % Acc1 is bound before the call
                                        % No computation needed after return

% Convenience wrapper
list_length_tco(List, Length) :-
    list_length_acc(List, 0, Length).

In list_length_acc/3, Acc1 is Acc + 1 binds Acc1 before the recursive call. The recursive call list_length_acc(T, Acc1, Length) is the last goal — nothing needs to happen after it returns. The WAM deallocates the current frame before making the call. Stack depth remains constant for any list length.

Local Stack during list_length_tco([100, 101, 102], N):

Call 1: list_length_acc([100,101,102], 0, N)
  → Acc1 = 1
  → [frame deallocated before recursive call]
Call 1 (reused frame): list_length_acc([101,102], 1, N)
  → Acc1 = 2
  → [frame deallocated]
Call 1 (reused frame): list_length_acc([102], 2, N)
  → Acc1 = 3
  → [frame deallocated]
Call 1 (reused frame): list_length_acc([], 3, N)
  → base case: N = 3
  → [return to original caller]

Peak stack depth: 1 frame throughout.

7.2.4 Determinism and TCO: The Interaction

TCO applies only when the recursive clause has no open choice points at the point of the last call. An open choice point prevents frame deallocation because the choice point frame holds a reference to the current environment frame — the engine may need to backtrack into it. This is why the determinism design from Chapter 6 is not just a logical concern but a memory-safety concern: a non-deterministic tail-recursive predicate does not get TCO and may still exhaust the Local Stack.

% TCO-eligible: deterministic (single matching clause per call)
process_list_det([], _Goal).
process_list_det([H|T], Goal) :-
    call(Goal, H),              % May create choice points for Goal
    !,                          % Cut eliminates them before last call
    process_list_det(T, Goal).  % Last goal, no open choice points → TCO

% NOT TCO-eligible: choice point from call(Goal, H) left open
process_list_ndet([], _Goal).
process_list_ndet([H|T], Goal) :-
    call(Goal, H),              % Open choice point remains
    process_list_ndet(T, Goal). % Last goal, but choice point blocks TCO

7.3 The Accumulator Pattern

7.3.1 The Accumulator as a Forward-Carrying Result

The accumulator pattern converts a "result-on-the-way-back" recursive predicate into a "result-carried-forward" one. Instead of computing the answer during the unwinding phase (as each frame returns to its parent), the predicate carries a partial result as an extra argument through the recursion. The base case unifies the accumulator with the output. No computation happens during unwinding — there is no unwinding, because TCO eliminated the frames.

The pattern has three elements:

An entry predicate with the user-facing signature, which calls the accumulator predicate with an initial accumulator value
A base case that unifies the accumulator with the output argument
A recursive case that updates the accumulator and recurses

% Generic accumulator skeleton:
predicate(Input, Output) :-
    predicate_acc(Input, InitialAccumulator, Output).

predicate_acc([], Acc, Acc).                   % Base: accumulator IS the result
predicate_acc([H|T], Acc0, Result) :-
    update_accumulator(H, Acc0, Acc1),         % Compute new accumulator
    predicate_acc(T, Acc1, Result).            % Recurse with updated accumulator

7.3.2 `total_cluster_ram/2`: RAM Summation with Accumulator

%% total_cluster_ram(+HostList, -TotalRAM)
%% Sums the RAM_GB field across all physical_host/3 facts for the given hosts.
%% Uses accumulator pattern for O(1) stack recursion (TCO-eligible).

total_cluster_ram(HostList, TotalRAM) :-
    must_be(list, HostList),
    total_cluster_ram_acc(HostList, 0, TotalRAM).

total_cluster_ram_acc([], Acc, Acc).
total_cluster_ram_acc([Host|Rest], Acc0, Total) :-
    ( physical_host(Host, _, RAM) ->
        Acc1 is Acc0 + RAM
    ;
        Acc1 = Acc0   % Host not in KB — skip, don't fail
    ),
    total_cluster_ram_acc(Rest, Acc1, Total).   % Last goal → TCO

%% total_all_cluster_ram(-Total)
%% Convenience: sum RAM for ALL physical_host/3 facts in the KB.
%% Uses findall/3 to collect the list, then sums.

total_all_cluster_ram(Total) :-
    findall(Host, physical_host(Host, _, _), Hosts),
    total_cluster_ram(Hosts, Total).

?- total_all_cluster_ram(Total).
Total = 640.
% pve-node-01: 256 + pve-node-02: 128 + pve-node-03: 256 = 640 GB

?- total_cluster_ram(['pve-node-01', 'pve-node-02'], Total).
Total = 384.

% Comparison: with and without accumulator on a 3-element list
% (for 10,000 elements, the difference is 10,000 frames vs. 1)
?- length(BigList, 5000),
   maplist(=(100), BigList),
   statistics(local_stack, [Before, _]),
   total_cluster_ram(BigList, _),
   statistics(local_stack, [After, _]),
   Delta is After - Before,
   format("Stack delta for 5000-element sum: ~w bytes~n", [Delta]).
Stack delta for 5000-element sum: 0 bytes.   % TCO: constant stack

7.3.3 `foldl/4`: The General Accumulator Abstraction

foldl/4 from library(apply) abstracts the accumulator pattern completely. It takes a goal, a list, an initial accumulator value, and an output accumulator value:

% foldl(+Goal, +List, +V0, -V)
% Goal is called as: call(Goal, Elem, AccIn, AccOut)
% for each element Elem in List, threading AccIn → AccOut.
% foldl/4 is internally tail-recursive.

:- use_module(library(apply)).

% Reimplement total_cluster_ram using foldl:
ram_fold_step(Host, Acc0, Acc1) :-
    ( physical_host(Host, _, RAM) ->
        Acc1 is Acc0 + RAM
    ;
        Acc1 = Acc0
    ).

total_cluster_ram_fold(HostList, Total) :-
    must_be(list, HostList),
    foldl(ram_fold_step, HostList, 0, Total).

% foldl for building a structured inventory summary
inventory_fold_step(Host, Acc0, [host(Host, RAM, VMCount)|Acc0]) :-
    physical_host(Host, _, RAM),
    aggregate_all(count, vm(_, _, Host, _), VMCount).

cluster_inventory_summary(Summary) :-
    findall(Host, physical_host(Host, _, _), Hosts),
    foldl(inventory_fold_step, Hosts, [], Summary).

?- cluster_inventory_summary(S).
S = [
    host('pve-node-03', 256, 1),
    host('pve-node-02', 128, 2),
    host('pve-node-01', 256, 2)
].

7.3.4 `maplist`, `include`, `exclude`: The Batch Operators

% maplist/2 — apply a predicate to every element (side-effects only)
% maplist/3 — map a list to another list via a transformation predicate
% maplist/4 — zip two lists element-wise via a predicate

% include/3 — keep elements satisfying a predicate
% exclude/3 — remove elements satisfying a predicate

% All are internally tail-recursive. All expect the predicate to be deterministic
% for each element (or wrap it in once/1 if not).

:- use_module(library(apply)).

% Extract all running VM IDs from the KB:
running_vm_ids(IDs) :-
    findall(ID, vm(ID, _, _, running), IDs).

% Filter a list of IDs to only those with RAM requirement <= threshold:
within_ram_budget(MaxRAM, VMID) :-
    vm_spec_by_id(VMID, Required),
    Required =< MaxRAM.

% Build a predicate that binds the MaxRAM argument partially:
% (This is the standard partial-application pattern for include/3)
filter_by_ram(MaxRAM, AllIDs, BudgetIDs) :-
    include(within_ram_budget(MaxRAM), AllIDs, BudgetIDs).

% vm_spec_by_id/2: maps a VM ID to its spec
vm_spec_by_id(VMID, Required) :-
    vm(VMID, Name, _, _),
    vm_spec(Name, Required).

?- running_vm_ids(IDs).
IDs = [100, 101, 102, 104].

?- filter_by_ram(8, [100, 101, 102, 104], Budget).
Budget = [100, 102, 104].
% VM 101 (postgres-prod-01) requires 16GB — excluded from 8GB budget.

7.4 The Build: Batch VM Command Generator

7.4.1 Supporting KB Facts

%% Add to /opt/logic-node/kb/inventory/proxmox_inventory.pl

%% vm_vlan(+VMID, +VLANID)
%% Maps VM IDs to their primary VLAN assignment.
vm_vlan(100, 20).    % nginx-prod-01 on production VLAN
vm_vlan(101, 20).    % postgres-prod-01 on production VLAN
vm_vlan(102, 20).    % nginx-prod-02 on production VLAN
vm_vlan(103, 20).    % worker-01 on production VLAN
vm_vlan(104, 10).    % monitoring-01 on management VLAN

%% batch_excluded_vm(+VMID)
%% "Do Not Touch" exclusion list for batch operations.
%% A VM listed here is NEVER included in the output of vm_constraint_filter/3,
%% regardless of how many positive constraints it satisfies.
%%
%% Add VMs here that:
%%   — Are the Logic Node VM itself (self-operation risk)
%%   — Are active quorum nodes for a HA cluster (removing them breaks quorum)
%%   — Are management-plane VMs (monitoring, IPAM, DNS) whose batch
%%     disruption would impair the ability to recover from the batch itself
%%   — Are explicitly flagged by the operations team as "no automated touch"
%%
%% This list is append-only during a maintenance window.
%% Removal requires explicit senior-operator sign-off.
%%
%% Note: VMID 104 (monitoring-01) is excluded because it is the host
%% running the infrastructure monitoring stack. Stopping it during a
%% maintenance batch eliminates the operator's visibility into whether
%% the batch is causing problems.

batch_excluded_vm(104).    % monitoring-01 — infrastructure observability
% batch_excluded_vm(105).  % (example: logic-node-vm — the Logic Node itself)

7.4.2 `batch_snapshot/3`: The Full Pipeline

logicadmin@logic-node-01:~$ nano /opt/logic-node/kb/oracle/batch_oracle.pl

%% =============================================================================
%% FILE:    /opt/logic-node/kb/oracle/batch_oracle.pl
%% PURPOSE: Batch VM operation command generators using list processing.
%%
%% SECURITY CONTRACT:
%%   All entry points enforce: must_be(list, L), length(L, N), N =< 500.
%%   No list of unbounded length enters a recursive predicate.
%%   All generated commands pass through shell_quote/2.
%%   No execution primitives. Text output only.
%% =============================================================================

:- module(batch_oracle, [
    batch_snapshot/3,
    batch_stop/3,
    batch_start/3,
    vm_constraint_filter/3,
    generate_script/2,
    generate_script_staggered/3,
    intersperse_sleep/3,
    vm_in_exclusion_list/2
]).

:- use_module(library(apply)).
:- use_module('/opt/logic-node/kb/inventory/proxmox_inventory').
:- use_module('/opt/logic-node/kb/oracle/shell_safety').

%% ---------------------------------------------------------------------------
%% CONSTANTS
%% ---------------------------------------------------------------------------

max_batch_size(500).   %% Hard ceiling on batch input list length

%% ---------------------------------------------------------------------------
%% ENTRY-POINT GUARD
%% ---------------------------------------------------------------------------

%% batch_guard(+List, +CallerContext)
%% Mandatory entry point for all batch predicates.
%% Enforces: proper list, non-empty, within length ceiling.

batch_guard(List, Context) :-
    must_be(proper_list, List),   % proper_list gives more diagnostic error than list
    ( List = [] ->
        throw(error(empty_batch(Context), context(batch_guard/2, 'Empty list')))
    ; true ),
    max_batch_size(Max),
    length(List, N),
    ( N > Max ->
        throw(error(
            batch_too_large(N, Max, Context),
            context(batch_guard/2, 'Batch size exceeds maximum')
        ))
    ; true ).

%% ---------------------------------------------------------------------------
%% STEP 1 — Filter: live_only/2
%% Uses include/3 with a ground predicate to select running VMs.
%% ---------------------------------------------------------------------------

%% vm_is_running(+VMID)
%% Succeeds if VM with this ID is currently in running state.
%% Deterministic: vm/4 is ground on VMID (first-argument indexed).

vm_is_running(VMID) :-
    vm(VMID, _, _, running).

%% live_only(+AllIDs, -RunningIDs)
%% Filters a list of VM IDs to only those currently running.

live_only(AllIDs, RunningIDs) :-
    include(vm_is_running, AllIDs, RunningIDs).

%% ---------------------------------------------------------------------------
%% STEP 2 — Map: vm_snapshot_command/3
%% Generates a single snapshot command for one VM ID.
%% Wraps the Chapter 5 oracle pattern for use with maplist/3.
%% ---------------------------------------------------------------------------

%% vm_snapshot_command(+SnapName, +VMID, -Command)
%% Generates: qm snapshot <VMID> <SnapName>
%% Preconditions: VMID must exist in vm/4; SnapName must be shell-safe atom.
%%
%% Argument order: SnapName first, then VMID, then Command.
%% This ordering enables partial application with maplist/3:
%%   maplist(vm_snapshot_command(SnapName), IDList, CmdList)

vm_snapshot_command(SnapName, VMID, Command) :-
    must_be(atom,             SnapName),
    must_be(positive_integer, VMID),
    vm(VMID, _Name, _Host, _Status),    % Precondition: VM must exist in KB
    shell_quote_integer(VMID,      QVMID),
    shell_quote(SnapName, QSnapName),
    with_output_to(string(Command),
        format("qm snapshot ~w ~w", [QVMID, QSnapName])).

%% ---------------------------------------------------------------------------
%% STEP 3 — Serialise: generate_script/2 and generate_script_staggered/3
%% ---------------------------------------------------------------------------

%% generate_script(+Commands, -Script)
%% Joins command strings with newlines and adds a bash shebang header.
%% Commands run sequentially with no inter-command delay.
%% Use generate_script_staggered/3 for batch operations on shared storage.

generate_script(Commands, Script) :-
    must_be(proper_list, Commands),
    Commands \= [],
    atomic_list_concat(Commands, '\n', Body),
    atomic_list_concat(['#!/bin/bash', 'set -euo pipefail', '', Body, ''], '\n', Script).

%% generate_script_staggered(+Commands, +SleepSeconds, -Script)
%% Inserts `sleep N` between each command to prevent IO-Wait spikes
%% on the Proxmox storage backend during batch snapshot/stop/start runs.
%%
%% Rationale: submitting 100 simultaneous `qm snapshot` commands triggers
%% 100 concurrent copy-on-write operations against the ZFS pool. Lock
%% contention elevates host IO-Wait above the 40% threshold defined in
%% Section 6.4.1.1 — which then disqualifies that host from receiving
%% future migrations. The batch operation's own IO load creates the next
%% migration's blocking condition: a performance cascade seeded by the
%% Logic Node's own output.
%%
%% A 1-second stagger distributes 100 snapshots across ~100 seconds,
%% keeping per-operation IO pressure below the cascade threshold.
%% Recommended: 1s for NVMe/SSD pools, 3s for spinning disk or NFS.
%%
%% SleepSeconds: positive integer.

generate_script_staggered(Commands, SleepSeconds, Script) :-
    must_be(proper_list, Commands),
    Commands \= [],
    must_be(positive_integer, SleepSeconds),
    with_output_to(string(SleepLine),
        format("sleep ~w", [SleepSeconds])),
    with_output_to(string(SleepComment),
        format("# Stagger interval: ~ws between operations (IO-Wait cascade prevention)",
               [SleepSeconds])),
    intersperse_sleep(Commands, SleepLine, StaggeredCmds),
    atomic_list_concat(StaggeredCmds, '\n', Body),
    atomic_list_concat(
        ['#!/bin/bash', 'set -euo pipefail', SleepComment, '', Body, ''],
        '\n', Script).

%% intersperse_sleep(+Commands, +SleepLine, -Interspersed)
%% Inserts SleepLine between every adjacent pair of commands.
%% Base cases: empty list and single command receive no sleep line.

intersperse_sleep([], _, []).
intersperse_sleep([Cmd], _, [Cmd]).
intersperse_sleep([Cmd|Rest], SleepLine, [Cmd, SleepLine | Interspersed]) :-
    Rest \= [],
    intersperse_sleep(Rest, SleepLine, Interspersed).

%% ---------------------------------------------------------------------------
%% FULL PIPELINE: batch_snapshot/3
%% ---------------------------------------------------------------------------

%% batch_snapshot(+VMIDList, +SnapName, -Script)
%%
%% Pipeline:
%%   1. Guard: validate and bound the input list
%%   2. Filter: keep only running VMs (include/3)
%%   3. Map: generate snapshot command per VM (maplist/3)
%%   4. Serialise: join to staggered bash script (1s default stagger)
%%
%% Uses generate_script_staggered/3 with a 1-second interval.
%% For unstaggered output (testing only), call generate_script/2 directly.

batch_snapshot(VMIDList, SnapName, Script) :-
    batch_guard(VMIDList, batch_snapshot/3),
    must_be(atom, SnapName),

    % Step 1: Filter to running VMs only
    live_only(VMIDList, RunningIDs),
    RunningIDs \= [],   % Fail cleanly if no VMs are running

    % Step 2: Map to snapshot commands
    maplist(vm_snapshot_command(SnapName), RunningIDs, Commands),

    % Step 3: Serialise to staggered bash script
    % Default 1-second stagger — prevents IO-Wait cascade on shared ZFS pools.
    generate_script_staggered(Commands, 1, Script).

% REPL demonstration:

?- batch_snapshot([100, 101, 102, 103, 104], 'pre-maintenance-20260315', Script).
Script = "#!/bin/bash\nset -euo pipefail\n\nqm snapshot 100 'pre-maintenance-20260315'\nqm snapshot 101 'pre-maintenance-20260315'\nqm snapshot 102 'pre-maintenance-20260315'\nqm snapshot 104 'pre-maintenance-20260315'\n".
% VM 103 (worker-01) is stopped — filtered out by live_only/2.
% VM 104 (monitoring-01) is running — included.

% Format for readability:
?- batch_snapshot([100, 101, 102, 103, 104], 'pre-maintenance-20260315', Script),
   writeln(Script).
#!/bin/bash
set -euo pipefail

qm snapshot 100 'pre-maintenance-20260315'
qm snapshot 101 'pre-maintenance-20260315'
qm snapshot 102 'pre-maintenance-20260315'
qm snapshot 104 'pre-maintenance-20260315'

true.

% All VMs stopped — filter eliminates all, batch_snapshot fails cleanly:
?- batch_snapshot([103], 'test-snap', Script).
false.
% VM 103 is stopped. live_only/2 returns []. RunningIDs \= [] fails.

7.4.3 `batch_stop/3` and `batch_start/3`

%% vm_stop_command(+VMID, -Command)
%% Generates: qm stop <VMID>
%% Precondition: VM must be running (stopping a stopped VM is a no-op but
%% the oracle enforces semantic correctness — generate only meaningful commands).

vm_stop_command(VMID, Command) :-
    must_be(positive_integer, VMID),
    vm(VMID, _, _, running),
    shell_quote_integer(VMID, QVMID),
    with_output_to(string(Command),
        format("qm stop ~w", [QVMID])).

%% vm_start_command(+VMID, -Command)
vm_start_command(VMID, Command) :-
    must_be(positive_integer, VMID),
    vm(VMID, _, _, stopped),
    shell_quote_integer(VMID, QVMID),
    with_output_to(string(Command),
        format("qm start ~w", [QVMID])).

%% batch_stop(+VMIDList, +Reason, -Script)
%% Generates stop commands for all RUNNING VMs in the input list.
%% Reason is an atom embedded as a bash comment for audit trail.

batch_stop(VMIDList, Reason, Script) :-
    batch_guard(VMIDList, batch_stop/3),
    must_be(atom, Reason),
    include(vm_is_running, VMIDList, RunningIDs),
    RunningIDs \= [],
    maplist(vm_stop_command, RunningIDs, StopCmds),
    with_output_to(string(ReasonComment),
        format("# Reason: ~w", [Reason])),
    generate_script([ReasonComment | StopCmds], Script).

%% batch_start(+VMIDList, +Reason, -Script)
%% Generates start commands for all STOPPED VMs in the input list.

vm_is_stopped(VMID) :-
    vm(VMID, _, _, stopped).

batch_start(VMIDList, Reason, Script) :-
    batch_guard(VMIDList, batch_start/3),
    must_be(atom, Reason),
    include(vm_is_stopped, VMIDList, StoppedIDs),
    StoppedIDs \= [],
    maplist(vm_start_command, StoppedIDs, StartCmds),
    with_output_to(string(ReasonComment),
        format("# Reason: ~w", [Reason])),
    generate_script([ReasonComment | StartCmds], Script).

?- batch_stop([100, 101, 102, 103, 104], 'scheduled-maintenance-window', Script),
   writeln(Script).
#!/bin/bash
set -euo pipefail

# Reason: scheduled-maintenance-window
qm stop 100
qm stop 101
qm stop 102
qm stop 104

true.
% VM 103 (worker-01) is already stopped — excluded from stop script.

7.4.4 Constraint Filter: Multi-Condition VM Selection

The constraint filter builds a list of VM IDs satisfying multiple conditions simultaneously — VLAN membership AND RAM requirement. This is the relational batching pattern: instead of writing a special-purpose loop for each filter combination, declare the constraints and let findall/3 + include/3 compose them.

%% vm_constraint_filter(+Constraints, -MatchingIDs, +Options)
%%
%% Constraints is a list of constraint terms:
%%   vlan(ID)         — VM must be on VLAN ID
%%   ram_min(GB)      — VM's spec must require >= GB RAM
%%   ram_max(GB)      — VM's spec must require =< GB RAM
%%   status(S)        — VM must have status S
%%   host(H)          — VM must be on host H
%%   exclude(IDList)  — explicit caller-supplied exclusion list (IDs to remove)
%%
%% Options: options(MaxResults, MaxBatch)
%%
%% CRITICAL EXCLUSION: regardless of the supplied Constraints, all VMs
%% listed in batch_excluded_vm/1 are stripped from the result AFTER
%% positive filtering. The exclusion is not a constraint the caller can
%% override — it is a hard architectural boundary encoded in the KB.
%% A VM in batch_excluded_vm/1 cannot appear in any batch operation output.

vm_constraint_filter(Constraints, MatchingIDs, options(MaxResults, _MaxBatch)) :-
    must_be(list, Constraints),
    must_be(positive_integer, MaxResults),

    % Collect all VM IDs from the KB
    findall(ID, vm(ID, _, _, _), AllIDs),

    % Apply each positive/negative constraint as a filter via foldl
    foldl(apply_vm_constraint, Constraints, AllIDs, MatchingIDs1),

    % MANDATORY: strip all batch_excluded_vm IDs from the result.
    % This step runs AFTER all caller-supplied constraints.
    % It cannot be disabled, composed away, or overridden by any constraint.
    findall(ExID, batch_excluded_vm(ExID), ExcludedIDs),
    exclude(vm_in_exclusion_list(ExcludedIDs), MatchingIDs1, MatchingIDs2),

    % Enforce result count ceiling
    length(MatchingIDs2, NFound),
    ( NFound > MaxResults ->
        length(MatchingIDs, MaxResults),
        append(MatchingIDs, _, MatchingIDs2)
    ;
        MatchingIDs = MatchingIDs2
    ).

%% vm_in_exclusion_list(+ExcludedIDs, +VMID)
%% Succeeds if VMID is in the ExcludedIDs list.
%% Used with exclude/3: exclude(vm_in_exclusion_list(ExcludedIDs), Candidates, Safe).

vm_in_exclusion_list(ExcludedIDs, VMID) :-
    memberchk(VMID, ExcludedIDs).

%% apply_vm_constraint(+Constraint, +IDsIn, -IDsOut)
%% Folds a single constraint over the current list of candidate IDs.

apply_vm_constraint(vlan(VID), IDsIn, IDsOut) :-
    must_be(integer, VID),
    include(vm_on_vlan(VID), IDsIn, IDsOut).

apply_vm_constraint(ram_min(GB), IDsIn, IDsOut) :-
    must_be(positive_integer, GB),
    include(vm_ram_min(GB), IDsIn, IDsOut).

apply_vm_constraint(ram_max(GB), IDsIn, IDsOut) :-
    must_be(positive_integer, GB),
    include(vm_ram_max(GB), IDsIn, IDsOut).

apply_vm_constraint(status(S), IDsIn, IDsOut) :-
    must_be(atom, S),
    include(vm_has_status(S), IDsIn, IDsOut).

apply_vm_constraint(host(H), IDsIn, IDsOut) :-
    must_be(atom, H),
    include(vm_on_host(H), IDsIn, IDsOut).

%% exclude(+IDList): caller-supplied exclusion list.
%% This is ADDITIONAL to the mandatory batch_excluded_vm/1 stripping —
%% the caller can supply extra exclusions without touching the KB.
%% Neither this nor any constraint can un-exclude a batch_excluded_vm entry.

apply_vm_constraint(exclude(IDList), IDsIn, IDsOut) :-
    must_be(proper_list, IDList),
    exclude(vm_in_exclusion_list(IDList), IDsIn, IDsOut).

%% Primitive constraint predicates (for use with include/3):

vm_on_vlan(VID, VMID) :-
    vm_vlan(VMID, VID).

vm_ram_min(MinGB, VMID) :-
    vm(VMID, Name, _, _),
    vm_spec(Name, Required),
    Required >= MinGB.

vm_ram_max(MaxGB, VMID) :-
    vm(VMID, Name, _, _),
    vm_spec(Name, Required),
    Required =< MaxGB.

vm_has_status(Status, VMID) :-
    vm(VMID, _, _, Status).

vm_on_host(Host, VMID) :-
    vm(VMID, _, Host, _).

% REPL: find all running VMs on VLAN 20

?- vm_constraint_filter(
       [vlan(20), status(running)],
       IDs,
       options(100, 500)
   ).
IDs = [100, 101, 102].
% VM 103 (worker-01): stopped — excluded by status(running)
% VM 104 (monitoring-01): on VLAN 10 — excluded by vlan(20)
% Even if 104 were on VLAN 20, batch_excluded_vm(104) would strip it.

% Critical exclusion demonstrated directly:
% Add a test fact placing monitoring-01 on VLAN 20, run the filter,
% verify it is still excluded by batch_excluded_vm/1:
?- assertz(vm_vlan(104, 20)),   % Temporarily: monitoring-01 now also on VLAN 20
   vm_constraint_filter([vlan(20), status(running)], IDs, options(100, 500)),
   retract(vm_vlan(104, 20)).
IDs = [100, 101, 102].
% VM 104 matched vlan(20) and has status running —
% but batch_excluded_vm(104) stripped it AFTER positive filtering.
% The exclusion cannot be overridden by any constraint combination.

% Caller-supplied exclusion (exclude/1 constraint):
?- vm_constraint_filter(
       [vlan(20), status(running), exclude([101])],
       IDs,
       options(100, 500)
   ).
IDs = [100, 102].
% VM 101 (postgres-prod-01) additionally excluded by caller request.
% This is in addition to the KB-level batch_excluded_vm exclusions.

% Combine with batch_snapshot — exclusion applies before script generation:
?- vm_constraint_filter([vlan(20), status(running)], IDs, options(100, 500)),
   batch_snapshot(IDs, 'pre-vlan20-maintenance', Script),
   writeln(Script).
#!/bin/bash
set -euo pipefail
# Stagger interval: 1s between operations (IO-Wait cascade prevention)

qm snapshot 100 'pre-vlan20-maintenance'
sleep 1
qm snapshot 101 'pre-vlan20-maintenance'
sleep 1
qm snapshot 102 'pre-vlan20-maintenance'

true.
% monitoring-01 (104) absent: excluded by batch_excluded_vm/1.
% 1-second stagger between each snapshot command.

7.5 Security Context: Recursion Limits and Input Sanitisation

7.5.1 The List Length Attack

The list length attack is a denial-of-service vector against any recursive predicate that accepts external input. The input does not need to be malformed in the sense of having wrong types or values — it just needs to be longer than the stack can accommodate, or (worse) circular.

Vector 1 — Exhaustion via length: A batch operation endpoint (Chapter 13's HTTP interface) that accepts a JSON array of VM IDs and passes it to batch_snapshot/3 without a length check will process lists of arbitrary length. Processing a list of 100,000 IDs through a non-TCO predicate allocates 100,000 frames — approximately 4MB of Local Stack on a 64-bit system — before any command is generated. Under concurrent load (100 simultaneous requests), this is 400MB of Local Stack allocation, potentially exhausting the Logic Node's configured stack limit and crashing the inference service.

With TCO and an accumulator, a 100,000-element list uses one frame — but it still takes O(n) time. An adversary submitting a batch of 10,000,000 VM IDs (syntactically valid — just very long) produces a CPU exhaustion attack even when stack safety is preserved.

Vector 2 — Circular list / Rational Tree: SWI-Prolog permits rational trees — cyclic terms — when set_prolog_flag(occurs_check, false) is active (the default). A circular list L where L = [1 | L] (the list's own tail is itself) is a valid Prolog term under this flag. The unification [H|T] = L succeeds, binding H = 1 and T = L (the same circular list). Any predicate that recurses on the tail without checking for cycles will loop forever.

% Creating a circular list (possible under default occurs_check=false):
?- L = [1, 2 | L].   % Under occurs_check=false, this SUCCEEDS
L = [1, 2, 1, 2, ...].   % Infinite rational tree

% Naive recursive predicate on a circular list:
?- list_length_naive([1, 2 | L_circular], N).
% ... never terminates — N increments forever
% ERROR: Stack limit exceeded (eventually)

The circular list attack does not require an adversary to construct one deliberately. A corrupted KB update that incorrectly uses assertz/1 with a self-referential list term, or a findall/3 operating over rules that generate infinite solutions, can produce circular structures silently.

7.5.2 The Mandatory Entry-Point Guard

Every batch predicate in the Logic Node follows the three-step entry guard:

%% batch_entry_guard(+List, +MaxLength, +CallerName)
%% Step 1: must_be(proper_list, L) — rejects non-lists, partial lists,
%%         and improper lists with a descriptive typed error message.
%%         Preferred over must_be(list, L): the 'proper_list' type is
%%         more descriptive in error output, distinguishing partial
%%         lists ([H|_Unbound]) from full non-list terms.
%% Step 2: is_list(L)     — cycle-safe check: fails on circular lists
%%         before length/2 can be called (length/2 on circular → infinite loop)
%% Step 3: length(L, N)   — computes length once is_list/1 has confirmed safety
%% Step 4: N =< MaxLength — enforces the hard ceiling

batch_entry_guard(List, MaxLength, Caller) :-
    % Step 1: proper_list type check — more diagnostic than must_be(list, L)
    % Error on partial list: type_error(proper_list, [1,2|_])
    % Error on atom: type_error(proper_list, foo)
    % Error on unbound: instantiation_error
    must_be(proper_list, List),

    % Step 2: circularity check — is_list/1 uses occurs-check-safe traversal
    % in SWI-Prolog. It fails on circular/improper lists.
    ( is_list(List) ->
        true
    ;
        throw(error(
            circular_or_improper_list(Caller),
            context(batch_entry_guard/3, 'Circular or improper list detected')
        ))
    ),

    % Step 3 + 4: length check with ceiling
    length(List, N),
    ( N > MaxLength ->
        throw(error(
            batch_too_large(N, MaxLength, Caller),
            context(batch_entry_guard/3, 'Input list exceeds maximum batch size')
        ))
    ;
        true
    ).

Note the distinction between must_be(list, L), must_be(proper_list, L), and is_list/1:

Predicate	Accepts	Rejects	Circular list behaviour	Error diagnostic
`must_be(list, L)`	Proper lists	Non-lists, unbound	Implementation-dependent (may loop)	`type_error(list, Term)`
`must_be(proper_list, L)`	Proper lists `[...\|[]]`	Partial lists, atoms, unbound	Fails on circular (SWI-Prolog)	`type_error(proper_list, Term)` — distinguishes partial from non-list
`is_list/1`	Proper lists	Partial lists `[...\|_]`, atoms	Fails on circular lists (cycle-safe)	No error — boolean fail
`length(L, N)`	Proper lists	Unbound tails	Loops on circular lists	No error — infinite loop

The diagnostic advantage of must_be(proper_list, L) over must_be(list, L) is material for Domain 2 review: when a malformed batch arrives at the HTTP interface (Chapter 13), the error term type_error(proper_list, [100, 101|_G334]) immediately identifies that a partial list with an unbound tail was the issue — the operator sees the partial list structure in the error, not an opaque type_error(list, ...). The guard order is unchanged: must_be(proper_list, ...) precedes is_list/1 which precedes length/2.

7.5.3 The `occurs_check` Flag: Closing the Circular Term Vector

Under the default Prolog flag occurs_check=false, the unification X = f(X) succeeds, creating a circular term. This enables rational tree operations (some constraint programming applications require them) but opens the circular list vector for batch predicates.

For the Logic Node, the correct setting is:

%% In /opt/logic-node/main.pl — at session initialisation
%% Set occurs_check=true to prevent circular term creation during unification.
%% This is O(1) overhead for ground unification, O(n) for unification involving
%% large terms — acceptable for the oracle workloads in this architecture.
:- set_prolog_flag(occurs_check, true).

With occurs_check=true, L = [1 | L] throws a type error rather than creating a rational tree. The circular list vector is closed at the unification layer rather than requiring defensive checks in every predicate.

The tradeoff: occurs_check=true makes unification O(n) in term size rather than O(1). For the KB's ground atoms and small compound terms, this is negligible. For large term construction (building a 10,000-element list via backtracking), it adds measurable overhead. The Logic Node's batch operations build lists via findall/3, not via unification over large structures, so the overhead is acceptable.

7.5.4 Partial List Detection

An improper list — one whose tail is not [] but is another non-list term — is distinct from a circular list but equally dangerous:

% Improper list: tail is an atom, not []
?- X = [1, 2 | foo].   % X is an improper list — tail is atom 'foo'
X = [1, 2|foo].

% is_list/1 rejects improper lists:
?- is_list([1, 2|foo]).
false.

% length/2 on an improper list: throws instantiation_error or fails
?- length([1, 2|foo], N).
false.   % Depends on SWI-Prolog version; always incorrect to use on improper lists

Improper lists arrive most commonly from:

append/3 called with an unbound third argument not yet unified
Partial list construction patterns where the tail variable was never closed
External input (JSON parsing producing [elem|Rest] where Rest was never unified)

The is_list/1 check in batch_entry_guard/3 rejects all of these.

7.5.5 Verification: Security Properties of Batch Oracle

?- consult('/opt/logic-node/kb/oracle/batch_oracle.pl').
true.

% 1. Length attack — list too large
?- length(HugeList, 501),
   maplist(=(100), HugeList),
   catch(
       batch_snapshot(HugeList, test, _),
       error(batch_too_large(501, 500, _), _),
       format("PASS: length attack caught at 501 elements~n")
   ).
PASS: length attack caught at 501 elements

% 2. Empty list — fail-fast
?- catch(
       batch_snapshot([], test, _),
       error(empty_batch(_), _),
       format("PASS: empty batch rejected~n")
   ).
PASS: empty batch rejected

% 3. Non-list input — proper_list gives more diagnostic error
?- catch(
       batch_snapshot(not_a_list, test, _),
       error(type_error(proper_list, not_a_list), _),
       format("PASS: non-list rejected by must_be(proper_list)~n")
   ).
PASS: non-list rejected by must_be(proper_list)

% 4. Valid batch — staggered script produced
?- batch_snapshot([100, 101], 'sec-verify', Script),
   sub_string(Script, _, _, _, "sleep 1").
true.   % ✓ Stagger sleep present in generated script

% 5. Accumulator: constant stack for large list
?- length(TestList, 300),
   maplist(=(100), TestList),
   statistics(local_stack, [B, _]),
   total_cluster_ram(TestList, _),
   statistics(local_stack, [A, _]),
   Delta is A - B,
   format("Stack delta (300 elements, accumulator): ~w bytes~n", [Delta]).
Stack delta (300 elements, accumulator): 0 bytes.

Outcome: The Relational Batching Model

7.6.1 The Conceptual Transition

Imperative batch scripting generates commands by iterating over a data source — a CMDB query, a CSV, an API response — and building strings in a loop. The correctness of the output depends on the correctness of the data source (which is external), the correctness of the string template (which has no type safety), and the correctness of the iteration (which has mutable state).

Relational batching inverts this. The data source is the static KB, verified at load time. The iteration is a combination of findall/3 (which exhausts the KB) and include/3 (which filters it) and maplist/3 (which transforms it) — each of which is internally correct by construction. The string output is produced by shell_quote/2-wrapped format/2 calls. The entire pipeline from KB fact to shell command has no mutable state, no loop counter, and no opportunity for a mismatched template to silently produce incorrect output.

Imperative batch scripting	Relational batching
External data source (CMDB, API)	Static KB (ground facts, verified at load)
String template with variable substitution	`format/2` with `shell_quote/2`-wrapped arguments
Loop with mutable counter	`maplist/3`, `include/3`, `foldl/4` — no mutation
Bounds check is optional (usually absent)	`batch_entry_guard/3` — mandatory, throws on violation
Circular data structure: silent infinite loop	`is_list/1` + `occurs_check=true` — detected and rejected
Error in element N continues to element N+1	Single proof failure aborts the batch (all-or-nothing)
Script correctness: runtime (after execution)	Script correctness: proof time (before any text is generated)
Exclusion logic: ad-hoc conditional in the loop	`batch_excluded_vm/1` KB facts — unconditional, non-overridable
Burst IO: all commands submitted simultaneously	`generate_script_staggered/3` — `sleep N` between operations

7.6.2 Verification Checklist

% Load all oracle modules
?- consult('/opt/logic-node/kb/oracle/batch_oracle.pl').
true.

% 1. WAM heap: list cell anatomy
?- X = [100, 101, 102],
   X = [H|T],
   H = 100, T = [101, 102].
H = 100, T = [101, 102].   % ✓ [H|T] unification works as expected

% 2. TCO: accumulator pattern is constant-stack
?- length(L, 1000), maplist(=('pve-node-01'), L),
   statistics(local_stack, [B, _]),
   total_cluster_ram(L, _),
   statistics(local_stack, [A, _]),
   A =:= B.
true.   % ✓ Stack unchanged after 1000-element accumulator recursion

% 3. include/3: filter to running only
?- live_only([100, 101, 102, 103, 104], Running).
Running = [100, 101, 102, 104].   % ✓ VM 103 (stopped) excluded

% 4. maplist/3: command generation across list
?- maplist(vm_snapshot_command('test'), [100, 101], Cmds).
Cmds = ["qm snapshot 100 'test'", "qm snapshot 101 'test'"].

% 5. batch_snapshot: full pipeline
?- batch_snapshot([100, 101, 102, 103, 104], 'verify-snap', Script),
   atomic_list_concat(Lines, '\n', Script),
   length(Lines, N), N > 4.
true.   % ✓ Script has more than 4 lines (shebang + set -e + blank + commands)

% 6. Constraint filter: multi-condition
?- vm_constraint_filter([vlan(20)], IDs, options(100, 500)),
   length(IDs, N), N =:= 4.
true.   % ✓ All 4 production VMs are on VLAN 20

% 7. Critical exclusion: batch_excluded_vm stripping is unconditional
?- assertz(vm_vlan(104, 20)),
   vm_constraint_filter([vlan(20)], IDs, options(100, 500)),
   retract(vm_vlan(104, 20)),
   \+ memberchk(104, IDs).
true.   % ✓ VM 104 stripped by batch_excluded_vm/1 even when matching all constraints

% 8. Caller exclude/1 constraint removes additional IDs
?- vm_constraint_filter([vlan(20), status(running), exclude([101])], IDs, options(100, 500)),
   \+ memberchk(101, IDs).
true.   % ✓ Caller-supplied exclusion works alongside KB exclusion

% 9. Security: batch too large
?- length(Big, 501), maplist(=(100), Big),
   \+ batch_snapshot(Big, t, _).
true.   % ✓ 501-element batch is rejected

% 8. Security: non-list input rejected
?- \+ batch_snapshot(not_a_list, t, _).
true.   % ✓ type_error thrown and caught as failure in \+

Exercises

Exercise 7.1 — WAM Heap Trace Write a predicate heap_list_cost/2 that, given a list, reports the number of list cells it contains (not the elements — the ./2 cells). Verify your implementation against [100, 101, 102] (expected: 3 cells), [] (expected: 0 cells), and a nested list [[1,2],[3,4]] (expected: 6 cells — 2 outer + 4 inner). Explain the heap cost of append/3 in terms of cell count.

Exercise 7.2 — Accumulator Conversion The following naive predicate computes the maximum RAM across all hosts in a list:

max_ram_naive([H], Max) :- physical_host(H, _, Max).
max_ram_naive([H|T], Max) :-
    max_ram_naive(T, MaxT),
    physical_host(H, _, RAMH),
    Max is max(RAMH, MaxT).

Convert it to an accumulator predicate max_ram_acc/3 that is TCO-eligible. Verify that statistics(local_stack, ...) shows zero delta for both a 3-element and a 1,000-element list.

Exercise 7.3 — Constraint Filter Extension Add two new constraint types to vm_constraint_filter/3: ha_group(GroupName) (VM must be in the specified HA group from Chapter 6's vm_ha_group/2 facts) and not_on_host(HostName) (VM must NOT be on the specified host). Implement the corresponding apply_vm_constraint clauses and verify with REPL queries.

Exercise 7.4 — Circular List Defence Write a predicate safe_list_assert(+Term) that prevents asserting a list-containing fact to the KB if the list argument is circular or improper. Use is_list/1 and must_be/2. Test it with: a proper list (should succeed), an improper list [1,2|foo] (should throw), and — with occurs_check=false temporarily — a circular list [1|L] with L=[1|L] (should throw). Restore occurs_check=true after the test.

Exercise 7.5 — Batch Migration Script Using the predicates from Chapter 6's preflight_oracle.pl and this chapter's batch_oracle.pl, write batch_migrate_script/3:

batch_migrate_script(+VMIDList, +TargetHost, -Script)

The script should: (1) run can_migrate/3 on each VM ID, (2) include only permit/2 decisions, (3) generate a qm migrate command for each permitted VM, and (4) embed the deny/1 reasons as bash comments for rejected VMs. Enforce the batch size guard. Verify with a mixed list where some VMs are blocked by HA policy and some are permitted.