#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "module.h"
#include "network.h"
#include "parser.h"
#include "prolog.h"
#include "query.h"

#ifdef _WIN32
#include <windows.h>
#define msleep Sleep
#else
static void msleep(int ms)
{
	struct timespec tv = {0};
	tv.tv_sec = (ms) / 1000;
	tv.tv_nsec = ((ms) % 1000) * 1000 * 1000;
	nanosleep(&tv, &tv);
}
#endif

#define Trace(p1,p2,p3,p4) if (q->trace /*&& !consulting*/) trace_call(p1,p2,p3,p4)

#define DEBUG_MATCH if (0)

#ifdef INDEX_PROFILE

// Deliberately process-global and opt-in: this is diagnostic accounting for
// one workload, not query state. It reports which dynamic predicate lookups
// lose selectivity after an indexing change.

#define INDEX_PROFILE_ROWS 1024

typedef struct {
	const predicate *pr;
	char name[64];
	unsigned arity;
	uint64_t calls, linear, idx0, idx1, idx2, candidates;
} index_profile_row;

static index_profile_row g_index_profile[INDEX_PROFILE_ROWS];
static bool g_index_profile_registered;

static index_profile_row *index_profile_get(const predicate *pr)
{
	unsigned i = ((size_t)pr >> 4) % INDEX_PROFILE_ROWS;

	for (unsigned probes = 0; probes < INDEX_PROFILE_ROWS; probes++) {
		index_profile_row *r = &g_index_profile[i];

		if (!r->pr) {
			r->pr = pr;
			r->arity = pr->key.arity;
			snprintf(r->name, sizeof(r->name), "%s", C_STR(pr->m, &pr->key));
			return r;
		}

		if (r->pr == pr)
			return r;

		i = (i + 1) % INDEX_PROFILE_ROWS;
	}

	return NULL;
}

static void index_profile_report(void)
{
	for (unsigned rank = 0; rank < 20; rank++) {
		index_profile_row *best = NULL;

		for (unsigned i = 0; i < INDEX_PROFILE_ROWS; i++) {
			index_profile_row *r = &g_index_profile[i];
			if (r->pr && (!best || (r->candidates > best->candidates)))
				best = r;
		}

		if (!best || !best->candidates)
			break;

		fprintf(stderr, "INDEX_PROFILE %s/%u calls=%llu linear=%llu idx0=%llu idx1=%llu idx2=%llu candidates=%llu avg=%.1f\n",
			best->name, best->arity,
			(unsigned long long)best->calls, (unsigned long long)best->linear,
			(unsigned long long)best->idx0, (unsigned long long)best->idx1,
			(unsigned long long)best->idx2, (unsigned long long)best->candidates,
			best->calls ? (double)best->candidates / best->calls : 0.0);

		best->candidates = 0;
	}
}

#define INDEX_PROFILE_START(pr) index_profile_row *ip = index_profile_get(pr); if (ip) ip->calls++
#define INDEX_PROFILE_MODE(ip, n) if (ip) (ip)->n++
#define INDEX_PROFILE_CANDIDATES(ip, n) if (ip) ((ip)->candidates += (n))

#else

#define INDEX_PROFILE_START(pr)
#define INDEX_PROFILE_MODE(ip, n)
#define INDEX_PROFILE_CANDIDATES(ip, n)

#endif

static const unsigned INITIAL_NBR_QUEUE_CELLS = 100;
static const unsigned INITIAL_NBR_HEAP_CELLS = 100;
static const unsigned INITIAL_NBR_SLOTS = 1000;
static const unsigned INITIAL_NBR_TRAILS = 1000;
static const unsigned INITIAL_NBR_CHOICES = 100;
static const unsigned INITIAL_NBR_FRAMES = 100;
static const unsigned INITIAL_NBR_CELLS = 100;

int g_tpl_interrupt = 0;

typedef enum { CALL, EXIT, REDO, NEXT, FAIL } box_t;

#define YIELD_INTERVAL 100000	// Goal interval between yield checks
#define REDUCE_PRESSURE 1
#define PRESSURE_FACTOR 4
#define TRACE_MEM 0
#define OOM_RESERVE_SIZE (1024U * 1024U)

static void rearm_oom_reserve(query *q)
{
	if (!q->oom_reserve)
		q->oom_reserve = TPL_malloc(OOM_RESERVE_SIZE);
}

void release_oom_reserve(query *q)
{
	TPL_free(q->oom_reserve);
	q->oom_reserve = NULL;
}

void dump_term(query *q, const char *s, const cell *c)
{
	unsigned num_cells = c->num_cells;
	printf("*** %s\n", s);

	for (unsigned i = 0; i < num_cells; i++, c++) {
		printf("    ");
		printf("[%u] tag=%u ", i, c->tag);

		if (is_atom(c))
			printf("%s ", C_STR(q, c));
		else if (is_var(c))
			printf("_%u ", c->var_num);
		else if (is_compound(c))
			printf("%s/%u ", C_STR(q, c), c->arity);

		printf("\n");
	}
}

static void trace_call(query *q, cell *c, pl_ctx c_ctx, box_t box)
{
	if (!c || is_empty(c))
		return;

	if (is_builtin(c) && c->bif_ptr && !c->bif_ptr->fn)
		return;

#ifndef DEBUG
	if (c->val_off == g_sys_succeed_on_retry_s)
		return;

	if (c->val_off == g_sys_fail_on_retry_s)
		return;

	if (c->val_off == g_sys_jump_s)
		return;

	if (c->val_off == g_sys_drop_barrier_s)
		return;

	if (c->val_off == g_sys_block_catcher_s)
		return;

	if (c->val_off == g_conjunction_s)
		return;

	if (c->val_off == g_disjunction_s)
		return;
#endif

	if (box == CALL)
		box = q->retry?REDO:CALL;

	const char *src = C_STR(q, c);
	frame *f = GET_CURR_FRAME();
	q->step++;
	SB(pr);

	SB_sprintf(pr, "[%u:%s:%"PRIu64":f%u:fp%u:cp%u:sp%u:tp%u:hp%u/%u:nr%d] ",
		q->my_chan,
		q->st.m->name,
		q->step,
		q->st.cur_ctx, q->st.fp, q->st.cp, q->st.sp,
		q->st.tp,
		q->st.hp, q->st.hp_num,
		f->no_recov
		);

	SB_sprintf(pr, "%s ",
		box == CALL ? "CALL" :
		box == EXIT ? "EXIT" :
		box == REDO ? "REDO" :
		box == NEXT ? "NEXT" :
		box == FAIL ? "FAIL":
		"????");

	q->quoted = true;
	q->double_quotes = true;
	char *dst = print_term_to_strbuf(q, c, c_ctx, -1);
	SB_strcat(pr, dst);
	TPL_free(dst);
	q->quoted = false;
	q->double_quotes = false;
	SB_sprintf(pr, "%s", "\n");
	src = SB_cstr(pr);
	size_t srclen = srclen = SB_strlen(pr);
	int n = q->pl->current_error;
	stream *str = &q->pl->streams[n];
	tpl_write(src, srclen, str);
	SB_free(pr);
	if (++q->vgen == 0) q->vgen = 1;

	if (q->creep) {
		msleep(250);
	}
}

void check_pressure(query *q)
{
#if REDUCE_PRESSURE
	if (q->tmp_heap && (q->tmph_size > 4000)) {
		TPL_free(q->tmp_heap);
		q->tmp_heap = NULL;
		q->tmph_size = 1000;
	}

#if TRACE_MEM
	printf("*** q->st.sp=%u, q->slots_size=%u\n", (unsigned)q->st.sp, (unsigned)q->slots_size);
#endif
	if (q->st.sp < (q->slots_size / 2)) {
		unsigned new_size = q->st.sp < INITIAL_NBR_SLOTS ? INITIAL_NBR_SLOTS : q->st.sp + 1;
		q->slots_size = alloc_grow(q, (void**)&q->slots, sizeof(slot), new_size, new_size*5/4);
	}
#endif
}

static bool check_choice(query *q)
{
	choice_page *a = q->choice_current;

	if (a && (q->choice_next < (a->entries + a->page_size)))
		return true;

	if (a && a->next) {
		q->choice_current = a = a->next;
		q->choice_next = a->entries;
		return true;
	}

	a = TPL_calloc(1, sizeof(choice_page));
	if (!a) {
		q->oom = q->error = true;
		return false;
	}

	a->page_size = q->choice_current ? q->choice_current->page_size * 2 : INITIAL_NBR_CHOICES;
	a->entries = TPL_calloc(a->page_size, sizeof(choice));

	if (!a->entries) {
		TPL_free(a);
		q->oom = q->error = true;
		return false;
	}

	a->base = q->st.cp;
	a->prev = q->choice_current;

	if (a->prev)
		a->prev->next = a;
	else
		q->choice_pages = a;

	q->choice_current = a;
	q->choice_next = a->entries;
	return true;
}

bool check_frame(query *q, unsigned max_vars)
{
	CHECKED(check_slot(q, max_vars));
	pl_idx page_idx = q->st.fp >> FRAME_PAGE_SHIFT;

	if (page_idx >= q->frame_pages_size) {
		pl_idx pages = alloc_grow(q, (void**)&q->frame_pages, sizeof(frame *),
			page_idx + 1, (page_idx + 1) * 2);

		if (!pages) {
			q->oom = q->error = true;
			return false;
		}

		memset(q->frame_pages + q->frame_pages_size, 0,
			(pages - q->frame_pages_size) * sizeof(frame *));
		q->frame_pages_size = pages;
	}

	if (!q->frame_pages[page_idx]) {
		frame *frames = TPL_calloc(FRAME_PAGE_SIZE, sizeof(frame));
		if (!frames) {
			q->oom = q->error = true;
			return false;
		}

		for (unsigned i = 0; i < FRAME_PAGE_SIZE; i++)
			frames[i].idx = (page_idx << FRAME_PAGE_SHIFT) + i;

		q->frame_pages[page_idx] = frames;
	}

	frame *f = GET_NEW_FRAME();
	f->max_vars = max_vars;
	f->base = q->st.sp;
	return true;
}

bool check_slot(query *q, unsigned cnt)
{
	cnt += 2;	// Allow some extra

	pl_idx num = q->st.sp + cnt;

	if (num < q->slots_size)
		return true;

	pl_idx new_slotssize = alloc_grow(q, (void**)&q->slots, sizeof(slot), num+1, num * 2);

	if (!new_slotssize) {
		q->oom = q->error = true;
		return false;
	}

	q->slots_size = new_slotssize;
	return true;
}

bool check_trail(query *q)
{
	trail_page *a = q->trail_current;

	if (a && (q->trail_next < (a->entries + a->page_size)))
		return true;

	if (a && a->next) {
		q->trail_current = a = a->next;
		q->trail_next = a->entries;
		return true;
	}

	a = TPL_calloc(1, sizeof(trail_page));
	if (!a) {
		q->oom = q->error = true;
		return false;
	}

	a->page_size = q->trail_current ? q->trail_current->page_size * 2 : INITIAL_NBR_TRAILS;
	a->entries = TPL_calloc(a->page_size, sizeof(trail));

	if (!a->entries) {
		TPL_free(a);
		q->oom = q->error = true;
		return false;
	}

	a->base = q->st.tp;
	a->prev = q->trail_current;

	if (a->prev)
		a->prev->next = a;
	else
		q->trail_pages = a;

	q->trail_current = a;
	q->trail_next = a->entries;
	return true;
}

trail *get_trail(query *q, pl_idx idx)
{
	trail_page *a = q->trail_current;

	while (a && (idx < a->base))
		a = a->prev;

	while (a && (idx >= (a->base + a->page_size)))
		a = a->next;

	assert(a);
	return a->entries + (idx - a->base);
}

bool undo_on_backtrack(query *q, void *v, enum undo_item type)
{
	undo_item *u = TPL_calloc(1, sizeof(undo_item));
	if (!u) return false;
	u->m = q->st.m;
	u->c = v;

	if (type == UNDO_BBOARD)
		u->is_bboard = true;
	else if (type == UNDO_RULE)
		u ->is_rule = true;
	else
		u->is_cells = true;

	list *undo;

	if (q->st.cp) {
		choice *ch = GET_CURR_CHOICE();
		undo = &ch->undo;
	} else
		undo = &q->undo;

	list_push_back(undo, u);
	return true;
}

void make_call_engine(query *q, cell *tmp, cell *c)
{
	make_end(tmp);
	const frame *f = GET_CURR_FRAME();
	tmp->ret_instr = c + c->num_cells;	// save next as the return instruction
	tmp->chgen = f->chgen;				// ... choice-generation
	tmp->mid = q->st.m->id;				// ... current-module
}

void make_call(query *q, cell *tmp)
{
	make_end(tmp);
	const frame *f = GET_CURR_FRAME();
	cell *c = q->st.instr;
	tmp->ret_instr = c + c->num_cells;	// save next as the return instruction
	tmp->chgen = f->chgen;				// ... choice-generation
	tmp->mid = q->st.m->id;				// ... current-module
}

void make_call_redo(query *q, cell *tmp)
{
	make_end(tmp);
	const frame *f = GET_CURR_FRAME();
	tmp->ret_instr = q->st.instr;		// save the return instruction
	tmp->chgen = f->chgen;				// ... choice-generation
	tmp->mid = q->st.m->id;				// ... current-module
}

cell *prepare_call(query *q, bool noskip, cell *p1, pl_ctx p1_ctx, unsigned extras)
{
	unsigned num_cells = p1->num_cells + extras;
	cell *tmp = alloc_heap(q, num_cells);
	if (!tmp) return NULL;
	q->noskip = noskip;
	dup_cells_by_ref(tmp, p1, p1_ctx, p1->num_cells);
	return tmp;
}

const char *dump_id(const void *k, const void *v, const void *p)
{
	uint64_t id = (uint64_t)(size_t)k;
	static char tmpbuf[1024];
	snprintf(tmpbuf, sizeof(tmpbuf), "%"PRIu64"", id);
	return tmpbuf;
}

static size_t scan_is_chars_list_internal(query *q, cell *l, pl_ctx l_ctx, bool allow_codes, bool *has_var, bool *is_partial, cell **cptr)
{
	*is_partial = *has_var = false;
	size_t is_chars_list = 0;
	cell *save_l = l;
	pl_ctx save_l_ctx = l_ctx;
	bool any1 = false, any2 = false;
	PROLOG_LIST_HANDLER(l);

	while (is_list(l) && (q->st.m->flags.double_quote_chars || allow_codes)) {
		cell *h = PROLOG_LIST_HEAD(l);
		pl_ctx h_ctx = l_ctx;
		slot *e = NULL;
		uint32_t save_vgen = 0;
		int both = 0;
		DEREF_VAR(any1, both, save_vgen, e, e->vgen, h, h_ctx, q->vgen);
		q->suspect = h;

		if (is_var(h)) {
			*has_var = true;
			return 0;
		}

		if (!is_integer(h) && !is_iso_atom(h))
			return 0;

		if (is_integer(h) && !allow_codes)
			return 0;

		if (is_integer(h)) {
			int ch = get_smallint(h);
			char tmp[MAX_BYTES_PER_CODEPOINT+1];
			put_char_utf8(tmp, ch);
			size_t len = len_char_utf8(tmp);
			is_chars_list += len;
		} else {
			const char *src = C_STR(q, h);
			size_t len = len_char_utf8(src);

			if (len != C_STRLEN(q, h))
				return 0;

			is_chars_list += len;
		}

		if (e) e->vgen = save_vgen;
		l = PROLOG_LIST_TAIL(l);
		cell *lsave = l;

		both = 0;
		DEREF_VAR(any2, both, save_vgen, e, e->vgen, l, l_ctx, q->vgen);

		if (both) {
			*is_partial = true;
			save_l = lsave;
			break;
		}
	}

	if (any2 && !*is_partial) {
		cell *l2 = save_l;
		pl_ctx l2_ctx = save_l_ctx;
		PROLOG_LIST_HANDLER(l2);

		while (is_list(l2) && (q->st.m->flags.double_quote_chars || allow_codes)) {
			PROLOG_LIST_HEAD(l2);
			l2 = PROLOG_LIST_TAIL(l2);
			RESTORE_VAR(l2, l2_ctx, l2, l2_ctx, q->vgen);
		}
	}

	if (is_var(l)) {
		*has_var = *is_partial = true;
		if (cptr) *cptr = l;
	} else if ((is_interned(l) || is_string(l) || is_number(l)) && !is_nil(l)) {
		*is_partial = true;
		if (cptr) *cptr = save_l;
	} else if (!is_interned(l) || !is_nil(l))
		is_chars_list = 0;

	return is_chars_list;
}

size_t scan_is_chars_list2(query *q, cell *l, pl_ctx l_ctx, bool allow_codes, bool *has_var, bool *is_partial, cell **cptr)
{
	if (++q->vgen == 0) q->vgen = 1;
	return scan_is_chars_list_internal(q, l, l_ctx, allow_codes, has_var, is_partial, cptr);
}

size_t scan_is_chars_list(query *q, cell *l, pl_ctx l_ctx, bool allow_codes)
{
	bool has_var, is_partial;
	return scan_is_chars_list2(q, l, l_ctx, allow_codes, &has_var, &is_partial, NULL);
}

bool make_slice(query *q, cell *d, const cell *orig, size_t off, size_t n)
{
	if (!n) {
		make_atom(d, g_empty_s);
		return true;
	}

	if (is_slice(orig)) {
		*d = *orig;
		d->val_str += off;
		d->str_len = n;
		return true;
	}

	const char *s = C_STR(q, orig);

	if (is_string(orig))
		return make_stringn(d, s+off, n);

	return make_cstringn(d, s+off, n);
}

#define MAX_LOCAL_VARS (1L<<30)

int create_vars(query *q, unsigned cnt)
{
	frame *f = GET_CURR_FRAME();

	if (!cnt)
		return f->actual_slots;

	// Fail soft: callers use CHECKED() to throw resource_error(memory).
	// Setting oom/error here would make start() abort the query even when
	// catch/3 handles the throw (issue #1094).
	if ((f->actual_slots + cnt) > MAX_LOCAL_VARS)
		return -1;

	if (!check_slot(q, cnt))
		return -1;

	unsigned var_num = f->actual_slots;

	if (!f->op && ((f->base + f->initial_slots) == q->st.sp)) {
		f->initial_slots += cnt;
	} else if (!f->op) {
		f->op = q->st.sp;
	} else if ((f->op + (f->actual_slots - f->initial_slots)) == q->st.sp) {
	} else {
		pl_idx save_overflow = f->op;
		f->op = q->st.sp;
		pl_idx cnt2 = f->actual_slots - f->initial_slots;

		if (!check_slot(q, cnt2))
			return -1;

		memmove(q->slots+f->op, q->slots+save_overflow, sizeof(slot)*cnt2);
		q->st.sp += cnt2;
	}

	slot *e = get_slot(q, f, f->actual_slots);
	memset(e, 0, sizeof(slot)*cnt);
	q->st.sp += cnt;
	f->actual_slots += cnt;
	return var_num;
}

static void enter_predicate(query *q, predicate *pr)
{
	frame *f = GET_FRAME(q->st.cur_ctx);
	f->dbgen = q->pl->dbgen;
	q->st.pr = pr;

	if (pr->is_dynamic)
		pr->refcnt++;
}

// Leaving a predicate and dropping the goal's OWN alternatives
// choicepoint is always these two, in this order. It was six
// hand-written call sites; made one call so there is nothing left to
// get wrong when the ordering constraint changes.
//
// NOT for the cut/prune paths that drop somebody ELSE'S choicepoint:
// they pass ch->st.pr, and q->st.iter there belongs to a different and
// still-live goal.

void leave_predicate_and_drop(query *q, predicate *pr, bool is_final)
{
	leave_predicate(q, pr, is_final);
	drop_choice(q);
}

void leave_predicate(query *q, predicate *pr, bool is_final)
{
	if (!pr)
		return;

	// Drop our handle but do not free: run_state is snapshotted into
	// every choice after find_key(), and those copies alias the same
	// prefetch. drop_choice() frees when the owning slot goes; next_key()
	// frees on exhaustion (clearing any alias first).
	q->st.iter = NULL;

	if (!pr->is_dynamic || !pr->refcnt)
		return;

	if (--pr->refcnt != 0)
		return;

	if (!list_count(&pr->dirty))
		return;

	if (pr->is_abolished)
		return;

	// Predicate is no longer being used

	//printf("*** leave %u, %s/%u, in_retractall=%d, is_final=%d, retry=%d\n",
	//	(unsigned)list_count(&pr->dirty), C_STR(q, &pr->key), pr->key.arity, q->in_retractall, is_final, q->retry);

	module_lock(pr->m);
	rule *r;
	const frame *f = GET_CURR_FRAME();

	while ((r = list_pop_front(&pr->dirty)) != NULL) {
		predicate_delink(pr, r);

		// Through index_remove_clause() rather than by hand: two
		// copies of the same withdrawal drift apart the moment a
		// third index or a side list is added, and this one is the
		// copy that gets forgotten.

		if (pr->cnt)
			index_remove_clause(pr, r);

		if (q->in_retract && !r->cl.num_vars && q->pl->opt) {
			undo_on_backtrack(q, r, UNDO_RULE);
		} else {
			r->cl.is_deleted = true;
			list_push_back(&q->dirty, r);
		}
	}

	if (pr->idx1 && !pr->cnt) {
		sl_destroy(pr->idx0);
		sl_destroy(pr->idx2);
		sl_destroy(pr->idx1);
		pr->idx0 = pr->idx1 = pr->idx2 = NULL;
		pr->is_var_in_head = false;
		pr->is_var_in_first_arg = false;
		pr->is_var_in_idx2_arg = false;
		pr->idx2_arg = 0;
	} else if (pr->is_var_in_head || pr->is_var_in_first_arg || pr->is_var_in_idx2_arg) {
		// Clauses just left the chain. If the last var-headed one was
		// among them the flags are now stale, and being stale here is
		// one-way: they are only ever set by assert_commit(). Safe to
		// walk - refcnt is 0, so no query is iterating this predicate.

		recheck_var_in_indexed_args(pr);
	}

	module_unlock(pr->m);
}

static void query_purge_dirty_list(query *q)
{
	unsigned cnt = 0;
	rule *r;

	// Withdraw index entries before releasing anything. This was the one
	// free path that dropped no entries at all - leave_predicate() only
	// calls sl_rem() while pr->cnt is non-zero, and clauses reach here by
	// other routes besides. An entry left pointing into a freed clause
	// turns the next descent into a use-after-free.
	//
	// Two passes, because sl_rem() compares against keys borrowed from
	// the other clauses on this same list.

	for (r = list_front(&q->dirty); r; r = list_next(r))
		index_remove_clause(r->owner, r);

	while ((r = list_pop_front(&q->dirty)) != NULL) {
		clear_clause(&r->cl);
		TPL_free(r);
		cnt++;
	}

	if (cnt && 0)
		printf("*** query_purge_dirty_list %u\n", cnt);
}

static void trim_trail(query *q, bool reused)
{
	if (q->undo_hi_tp)
		return;

	pl_idx tp;

	if (q->st.cp)  {
		const choice *ch = GET_CURR_CHOICE();
		tp = ch->st.tp;
	} else
		tp = 0;

	while (q->st.tp > tp) {
		const trail *tr = get_trail(q, q->st.tp - 1);

		if (tr->val_ctx != q->st.cur_ctx)
			break;

		// After reuse_frame() these entries MUST go: it already
		// unshared the old slot contents and moved the new frame's
		// cells in by plain copy, so the reference now in the slot
		// belongs to that transfer. Undoing against it would unshare
		// somebody else's reference (double free).
		//
		// After push_frame() the frame is still live and its bindings
		// still own what they hold. Dropping the entry there means
		// nothing ever unshares a MANAGED cell unless the frame later
		// gets recovered by trim_frame() - which only fires when the
		// frame is topmost, has no_recov clear and no choices resume
		// into it. When it does not fire, the blob leaks.

		// Retaining an entry is only safe if the frame it names can
		// never be recycled underneath it: trim_frame() lowers
		// q->st.sp, a later frame reuses those slot indices, and a
		// stale entry would then unshare a binding it does not own
		// (double free). That is what trim_trail() is really for.
		//
		// A frame with no_recov set is exactly the case that cannot be
		// recycled - set_var() marks it when a binding escapes to
		// another frame - so its entries can be kept, and they must
		// be: nothing else will ever unshare a MANAGED cell sitting in
		// a frame that trim_frame() will never touch.

		if (!reused) {
			const frame *f = GET_FRAME(tr->val_ctx);

			if (f->no_recov) {
				const slot *e = get_slot(q, f, tr->var_num);

				if (is_managed(&e->c))
					break;
			}
		}

		pop_trail(q);
	}
}

static void trim_frame(query *q, const frame *f)
{
	for (unsigned i = 0; i < f->actual_slots; i++) {
		slot *e = get_slot(q, f, i);
		cell *c = &e->c;
		unshare_cell(c);
		memset(e, 0, sizeof(slot));
	}

	q->st.sp -= f->actual_slots;
	q->st.fp = q->st.cur_ctx;
}

bool add_trail(query *q, pl_ctx c_ctx, unsigned c_var_nbr, cell *attrs)
{
	// Must not bind without a trail entry: a silent failure here leaves
	// the variable set while undo_me() cannot clear it. That resurfaces
	// after catch/3 clears oom as uninstantiation_error(N) on
	// $fail_on_retry/1 (and similar), because the CP index stays bound.
	if (!check_trail(q))
		return false;

	trail *tr = q->trail_next++;
	q->st.tp++;
	tr->val_ctx = c_ctx;
	tr->var_num = c_var_nbr;
	tr->attrs = attrs;
	return true;
}

void undo_me(query *q)
{
	q->total_retries++;
	const choice *ch = GET_CURR_CHOICE();

	while (q->st.tp > ch->st.tp) {
		const trail *tr = pop_trail(q);
		const frame *f = GET_FRAME(tr->val_ctx);
		slot *e = get_slot(q, f, tr->var_num);
		cell *c = &e->c;
		unshare_cell(c);
		memset(e, 0, sizeof(slot));
		c->val_attrs = tr->attrs;
	}
}

static void try_me(query *q, unsigned num_vars)
{
	frame *f = GET_NEW_FRAME();
	f->initial_slots = f->actual_slots = num_vars;
	q->total_matches++;

	for (unsigned i = 0; i < num_vars; i++) {
		slot *e = get_slot(q, f, i);
		memset(e, 0, sizeof(slot));
	}
}

// Skip the branch join points compile_term() emits on the way to the
// clause end: a bare `true` landing, or a forward `$jump` to one. Both
// are no-ops for machine state (bif_sys_jump_1 only moves q->st.instr),
// so a goal followed by nothing but these is followed by nothing.
// is_end() on the result means the clause is over.

static const cell *skip_landings(const cell *c)
{
	while (!is_end(c)) {
		if (!is_interned(c))
			break;

		if ((c->val_off == g_true_s) && !c->arity)
			c += c->num_cells;						// landing
		else if ((c->val_off == g_sys_jump_s) && (c->arity == 1)
			&& is_smallint(c+1) && (get_smallint(c+1) > 0))
			c += get_smallint(c+1);					// jump to a landing
		else
			break;
	}

	return c;
}

static void push_frame(query *q)
{
	const frame *f_cur = GET_CURR_FRAME();
	frame *f_new = GET_NEW_FRAME();
	const cell *next_cell = skip_landings(q->st.instr + q->st.instr->num_cells);

	// Avoid long chains of useless returns...

	if (q->pl->opt && is_end(next_cell) && !next_cell->ret_instr) {
		f_new->prev = f_cur->prev;
		f_new->instr = f_cur->instr;
	} else {
		f_new->prev = q->st.cur_ctx;
		f_new->instr = q->st.instr;
	}

	f_new->op = 0;
	f_new->no_recov = q->no_recov;
	f_new->chgen = ++q->chgen;
	f_new->hp = q->st.hp;
	f_new->hp_num = q->st.hp_num;
	q->st.sp += f_new->actual_slots;
	q->st.cur_ctx = q->st.fp;
	q->st.fp++;
}

// Note: TCO's clause might not be the caller clause... hence passing
// num_vars. Currently restricted to the same predicate though (still?).

static void reuse_frame(query *q, unsigned num_vars)
{
	cell *c_next = q->st.instr + q->st.instr->num_cells;

	// This is if the last call was actually call/n

	if (c_next->val_off == g_sys_drop_barrier_s)
		drop_choice(q);

	// Copy slots from the new frame to the current frame...

	const frame *f_new = GET_NEW_FRAME();
	frame *f_cur = GET_CURR_FRAME();
	f_cur->initial_slots = f_cur->actual_slots = num_vars;
	f_cur->no_recov = false;

	for (pl_idx i = 0; i < num_vars; i++) {
		const slot *from = get_slot(q, f_new, i);
		slot *to = get_slot(q, f_cur, i);
		unshare_cell(&to->c);
		*to = *from;
	}

	q->st.sp = f_cur->base + f_cur->actual_slots;
	q->st.dbe->tcos++;
	q->total_tcos++;
	q->st.hp = f_cur->hp;
	q->st.hp_num = f_cur->hp_num;
	trim_heap(q);
}

// Does any choicepoint still need this frame?
//
// Generations do not order frames, so ch->gen > f->chgen was wrong in
// both directions: a choicepoint pushed by an earlier goal of the same
// clause carries gen == f->chgen and was invisible, while an ancestor's
// choicepoint can carry gen == f->chgen while having nothing to do with
// this frame.
//
// The frames answer it directly. ch->st.fp is the frame count when the
// choicepoint was pushed, so ch->st.fp >= q->st.fp means this frame was
// already live and a retry restores into it; anything pushed earlier
// belongs to an ancestor, and retrying that throws this frame away
// whole.
//
// skip counts the choicepoints commit_frame() drops itself: the
// in-progress clause choice, plus the call/N barrier when is_last_call()
// found one - reuse_frame() performs that bookkeeping directly, which is
// what keeps p(N) :- M is N-1, call(p, M) tail recursive.

static bool commit_any_choices(const query *q, unsigned skip)
{
	if (q->st.cp <= skip)
		return false;

	const choice *ch = GET_CHOICE(q->st.cp - 1 - skip);
	return ch->st.fp >= q->st.fp;
}

// Is the goal about to be called really the last thing this frame has
// to do? LCO (see reuse_frame) throws away the rest of the current
// instruction stream along with the frame, so it is only sound when
// there is nothing left in that stream.
//
// FLAG_INTERNED_RECURSIVE_CALL is not enough on its own: it is set at
// clause-compile time on any cell that *ends* at the clause end, which
// includes a goal that is merely the argument of a trailing control
// construct - \+ G, once(G), ignore(G), (C -> G) and friends. Those
// constructs run that very cell as a goal, but with a continuation of
// their own planted after it (a cut, a fail, a then-branch, a barrier
// drop), either inlined by compile_term() or built on the heap by
// prepare_call(). Reusing the frame there silently discards that
// continuation - e.g. \+ G would lose its `!, $drop_barrier, fail'
// and so succeed for a provable G.

static bool is_last_call(const query *q, bool *has_barrier)
{
	const cell *c = q->st.instr + q->st.instr->num_cells;
	bool barrier = false;

	// call/N plants nothing after the goal but a $drop_barrier, which is
	// bookkeeping that reuse_frame() performs directly.

	if (is_interned(c) && (c->val_off == g_sys_drop_barrier_s)) {
		c += c->num_cells;
		barrier = true;
	}

	if (has_barrier)
		*has_barrier = barrier;

	// Past that, only the branch join points that compile_term() emits
	// on the way to the clause end may be skipped: they do nothing.

	c = skip_landings(c);

	if (!is_end(c))
		return false;

	// The clause's own end cell means nothing is left to do. The end
	// cell of a heap continuation instead carries a return address and
	// the frame state to restore along with it, which is work - bar the
	// call/N case above, where reuse_frame() has always taken over.

	return barrier || !c->ret_instr;
}

// head_has_vars is q->has_vars as left by the head unification THIS call
// is committing to, and is passed rather than read here.
//
// q->has_vars is scratch owned by unify(): it is cleared on entry and
// set only for a goal-side variable at depth > 1, so it describes the
// most recent unification - not the goal, and not necessarily this one.
// Reading it inside commit_frame() was correct only because the single
// call site happens to sit immediately after the unify(). Anything
// unifying in between - another clause attempt, a clone - would silently
// change the answer, and a wrong answer here is not a lost
// optimisation: is_det makes commit_frame() drop the goal's
// alternatives choicepoint.

static void commit_frame(query *q, bool head_has_vars)
{
	q->st.dbe->matched++;
	q->total_matched++;

	clause *cl = &q->st.dbe->cl;
	frame *f = GET_CURR_FRAME();
	f->m = q->st.m;

	rule *save_dbe = q->st.dbe;
	// A unique clause head does not make the goal deterministic if some
	// OTHER clause carries a variable in an indexed argument: such a
	// clause unifies with anything, so it is still a live alternative.
	// Logtalk's logtalk_library_path/2 is exactly this shape - a packs
	// RULE with a var first argument sitting among 500 ground facts -
	// and claiming determinism there dropped the alternatives
	// choicepoint before the walk could reach the matching fact.

	bool is_det = !head_has_vars && cl->is_unique
		&& !q->st.pr->is_var_in_head && !q->st.pr->is_var_in_first_arg
		&& !q->st.pr->is_var_in_idx2_arg;
	bool last_match = is_det || cl->is_first_cut || !has_next_key(q)
		|| (is_next_cut(q->st.instr) && cl->is_fact);
	bool tco = false;

#if 0
	if (last_match) {
		fprintf(stderr, "*** q->no_recov=%d, last_match=%d %s/%u, q->st.cur_ctx=%u,q->st.fp=%u\n",
			q->no_recov, last_match,
			C_STR(q, q->st.key), q->st.key->arity,
			q->st.cur_ctx, q->st.fp
			);
	}
#endif

	if (!q->no_recov
		&& last_match
		&& (q->st.fp == (q->st.cur_ctx + 1))
		) {
		bool barrier = false;
		bool tail_recursive = is_recursive_call(q->st.instr) && is_last_call(q, &barrier);
		bool slots_ok = f->initial_slots <= cl->num_vars;
		bool choices = commit_any_choices(q, barrier ? 2 : 1);
		tco = slots_ok && tail_recursive && !choices;

#if 0
		cell *head = get_head(cl->cells);

		fprintf(stderr,
			"*** %s/%u tco=%d,q->no_recov=%d,last_match=%d,is_det=%d,"
			"tail_recursive=%d,slots_ok=%d,choices=%d,"
			"cl->num_vars=%u,f->initial_slots=%u/%u\n",
			C_STR(q, head), head->arity,
			tco, q->no_recov, last_match, is_det,
			tail_recursive, slots_ok, choices,
			cl->num_vars, f->initial_slots, f->actual_slots);
#endif
	}

	if (!q->st.dbe->owner->is_builtin)
		q->st.m = q->st.dbe->owner->m;

	const bool reused = tco && q->pl->opt;

	if (reused) {
		Trace(q, get_head(save_dbe->cl.cells), q->st.cur_ctx, EXIT);
		reuse_frame(q, cl->num_vars);
	} else {
		push_frame(q);
	}

	if (last_match) {
		leave_predicate_and_drop(q, q->st.pr, false);
		trim_trail(q, reused);


	} else {
		choice *ch = GET_CURR_CHOICE();
		ch->st.dbe = q->st.dbe;
		ch->gen = q->chgen;
	}

	q->st.instr = cl->alt ? cl->alt : get_body(cl->cells);
	if (!q->st.instr) q->st.instr = cl->cells + (cl->cidx-1);
	q->st.iter = NULL;
}

// The three kinds of undo item, disposed of in one place. This dispatch
// existed in two copies that had already drifted apart once - the
// UNDO_RULE case was added to one and not the other, so a retracted
// ground clause was freed as a cell block without clear_clause() and its
// managed cells were never unshared. Now there is one copy.

static void undo_list_drain(list *l)
{
	undo_item *u;

	while ((u = list_pop_back(l)) != NULL) {
		if (u->is_bboard)
			sl_del(u->m->keyval, u->key);
		else if (u->is_rule) {
			clear_clause(&u->r->cl);
			TPL_free(u->r);
		} else {
			unshare_cells(u->c, u->c->num_cells);
			TPL_free(u->c);
		}

		TPL_free(u);
	}
}

// Release the prefetch a choicepoint owns.
//
// run_state is snapshotted whole into every choice raised after
// find_key(), so the handle is aliased by all of them and only the
// choice it was built for may free it. iter_owner names that slot, and a
// choice's slot is simply its own index - which is why this takes the
// choice rather than a caller-computed cp. The three call sites had
// spelled that index two different ways (q->st.cp in retry_choice, where
// the decrement comes after; q->st.cp - 1 in drop_choice, where it comes
// before), which looked like a discrepancy and was not.

static void release_prefetch(query *q, choice *ch, pl_idx cp)
{
	if (!ch->st.iter || (ch->st.iter_owner != cp))
		return;

	// q->st may still alias it - defuse before the free.

	if (q->st.iter == ch->st.iter)
		q->st.iter = NULL;

	sl_done(ch->st.iter);
	ch->st.iter = NULL;
}

int retry_choice(query *q)
{
	while (q->st.cp) {
		undo_me(q);
		pl_idx cp = q->st.cp - 1;
		choice *ch = GET_CURR_CHOICE();
		pop_choice(q);
		undo_list_drain(&ch->undo);

		q->st = ch->st;

		frame *f = GET_CURR_FRAME();
		f->dbgen = ch->dbgen;
		f->chgen = ch->chgen;
		f->initial_slots = ch->initial_slots;
		f->actual_slots = ch->actual_slots;
		f->op = ch->op;
		f->base = ch->base;

		if (ch->reset)
			continue;

		if (ch->catchme_exception || ch->fail_on_retry) {
			// Choice abandoned without drop_choice(); free its prefetch.
			release_prefetch(q, ch, cp);
			leave_predicate(q, ch->st.pr, true);
			continue;
		}

		if (!ch->register_cleanup && q->noretry) {
			release_prefetch(q, ch, cp);
			leave_predicate(q, ch->st.pr, true);
			continue;
		}

		if (ch->register_cleanup && q->noretry)
			q->noretry = false;

		trim_heap(q);

		if (ch->succeed_on_retry) {
			q->st.instr += ch->skip;
			return ch->skip ? -2 : -1;
		}

		return 1;
	}

	trim_heap(q);
	return 0;
}

void drop_choice(query *q)
{
	if (!q->st.cp)
		return;

	pl_idx cp = q->st.cp - 1;
	choice *ch = GET_CHOICE(cp);

	// Free the multi-hit prefetch when the choice it was built for goes.
	// Cuts and last-match commits drop that choice without exhausting the
	// iterator; without this those prefetches are abandoned.

	release_prefetch(q, ch, cp);

	list *undo;

	if (q->st.cp > 1) {
		choice *ch_prev = GET_PREV_CHOICE();
		undo = &ch_prev->undo;
	} else
		undo = &q->undo;

	undo_item *u;

	while ((u = list_pop_front(&ch->undo)) != NULL)
		list_push_back(undo, u);

	pop_choice(q);
}

bool push_choice(query *q)
{
	CHECKED(check_choice(q));
	const frame *f = GET_CURR_FRAME();
	choice *ch = q->choice_next++;
	ch->skip = 0;
	ch->st = q->st;
	q->st.cp++;

	// Keep a record of the frame state, we need to restore
	// it on retry. On cut we commit to it.

	list_init(&ch->undo);
	ch->dbgen = f->dbgen;
	ch->chgen = ch->gen = f->chgen;
	ch->initial_slots = f->initial_slots;
	ch->actual_slots = f->actual_slots;
	ch->op = f->op;
	ch->base = f->base;

	ch->catchme_retry =
		ch->catchme_exception = ch->barrier = ch->register_cleanup =
		ch->block_catcher = ch->fail_on_retry =
		ch->succeed_on_retry = ch->reset = false;

	return true;
}

bool push_succeed_on_retry(query *q, pl_idx skip)
{
	CHECKED(push_choice(q));
	choice *ch = GET_CURR_CHOICE();
	ch->succeed_on_retry = true;
	ch->skip = skip;
	return true;
}

// A barrier is used when making a call, it sets a new
// choice generation so that normal cuts are contained.

bool push_barrier(query *q)
{
	CHECKED(push_choice(q));
	choice *ch = GET_CURR_CHOICE();
	frame *f = GET_CURR_FRAME();
	ch->gen = f->chgen = ++q->chgen;
	ch->barrier = true;
	return true;
}

bool push_succeed_on_retry_with_barrier(query *q, pl_idx skip)
{
	// FIXME: memory waste, but see docs/norecov.md
	frame *f = GET_CURR_FRAME();
	f->no_recov = true;
	CHECKED(push_barrier(q));
	choice *ch = GET_CURR_CHOICE();
	ch->succeed_on_retry = true;
	ch->skip = skip;
	return true;
}

bool push_fail_on_retry_with_barrier(query *q)
{
	CHECKED(push_barrier(q));
	choice *ch = GET_CURR_CHOICE();
	ch->fail_on_retry = true;
	return true;
}

bool push_reset_handler(query *q)
{
	CHECKED(push_fail_on_retry_with_barrier(q));
	choice *ch = GET_CURR_CHOICE();
	ch->reset = true;
	return true;
}

bool push_catcher(query *q, enum q_retry retry)
{
	CHECKED(push_barrier(q));
	choice *ch = GET_CURR_CHOICE();

	if (retry == QUERY_RETRY)
		ch->catchme_retry = true;
	else if (retry == QUERY_EXCEPTION)
		ch->catchme_exception = true;

	rearm_oom_reserve(q);

	return true;
}

// If the call is det then the barrier can be dropped...

bool drop_barrier(query *q, pl_idx cp)
{
	if ((q->st.cp-1) != cp)
		return false;

	const choice *ch = GET_CURR_CHOICE();
	frame *f = GET_CURR_FRAME();
	f->chgen = ch->chgen;
	drop_choice(q);
	return true;
}

void cut(query *q)
{
	const frame *f = GET_CURR_FRAME();

	while (q->st.cp) {
		choice *ch = GET_CURR_CHOICE();

		// A normal cut can't break out of a barrier...

		if (ch->barrier) {
			if (ch->gen <= f->chgen)
				break;
		} else {
			if (ch->gen < f->chgen)
				break;
		}

		// Done...

		leave_predicate(q, ch->st.pr, false);
		drop_choice(q);

		if (ch->register_cleanup && !ch->fail_on_retry) {
			cell *c = FIRST_ARG(ch->st.instr);
			pl_ctx c_ctx = ch->st.cur_ctx;
			c = deref(q, c, c_ctx);
			c_ctx = q->latest_ctx;
			do_cleanup(q, c, c_ctx);
			break;
		}
	}
}

static bool resume_any_choices(const query *q, const frame *f)
{
	if (!q->st.cp)
		return false;

	const choice *ch = GET_CURR_CHOICE();
	return ch->gen >= f->chgen;
}

// Resume at next goal in previous clause...

static bool resume_frame(query *q)
{
	const frame *f = GET_CURR_FRAME();

	if (f->prev == CTX_NUL)
		return false;

#if 0
	printf("*** q->st.cur_ctx=%d, f->no_recov=%d, any_choices=%d\n",
		(unsigned)q->st.cur_ctx,
		(unsigned)f->no_recov, (unsigned)resume_any_choices(q, f));
#endif
	Trace(q, get_head(f->instr), f->prev, EXIT);

	// Call is followed by !: drop callee-internal choices the cut will
	// kill so trim_frame can run. Stop at barriers (cut handles those,
	// including setup_call_cleanup) and at the parent clause choice
	// (gen < f->chgen) - that stays until the real cut.

	if (f->instr && is_next_cut(f->instr)) {
		while (q->st.cp) {
			choice *ch = GET_CURR_CHOICE();

			if (ch->barrier || (ch->gen < f->chgen))
				break;

			leave_predicate(q, ch->st.pr, false);
			drop_choice(q);
		}
	}

	if (q->pl->opt
		&& !f->no_recov
		&& (q->st.fp == (q->st.cur_ctx + 1))
		&& !resume_any_choices(q, f)
		) {
		q->total_recovs++;
		q->st.hp = f->hp;
		q->st.hp_num = f->hp_num;
		trim_frame(q, f);
	}

	q->st.instr = f->instr;
	q->st.cur_ctx = f->prev;
	f = GET_CURR_FRAME();
	q->st.m = f->m;
	return true;
}

// Proceed to next goal in current clause...

static void proceed(query *q)
{
	if (!q->noskip)
		q->st.instr += q->st.instr->num_cells;

	q->noskip = false;

	if (!is_end(q->st.instr))
		return;

	if (q->st.instr->ret_instr) {
		frame *f = GET_CURR_FRAME();
		f->chgen = q->st.instr->chgen;
		q->st.m = q->pl->modmap[q->st.instr->mid];
	}

	q->st.instr = q->st.instr->ret_instr;
}

static bool can_view(query *q, uint64_t dbgen, const rule *r)
{
	if (r->cl.is_deleted)
		return false;

	if (r->dbgen_created > dbgen)
		return false;

	if (r->dbgen_retracted && (r->dbgen_retracted <= dbgen))
		return false;

	return true;
}

static void setup_key(query *q)
{
	cell *save_arg1 = FIRST_ARG(q->st.key), *save_arg2 = NULL;
	cell *arg1 = deref(q, save_arg1, q->st.key_ctx);

	q->st.karg1_is_ground = !is_var(arg1);
	q->st.karg1_is_atomic = is_atomic(arg1);

	if (q->st.key->arity > 1) {
		cell *arg2 = deref(q, save_arg2 = NEXT_ARG(save_arg1), q->st.key_ctx);
		q->st.karg2_is_ground = arg2 && !is_var(arg2);
		q->st.karg2_is_atomic = arg2 && is_atomic(arg2);
	}

	if (q->st.key->arity > 2) {
		cell *arg3 = deref(q, NEXT_ARG(save_arg2), q->st.key_ctx);
		q->st.karg3_is_ground = arg3 && !is_var(arg3);
		q->st.karg3_is_atomic = arg3 && is_atomic(arg3);
	}
}

static void next_key(query *q)
{
	if (q->st.iter_single) {
		// A single-hit lookup has no iterator and must not fall through
		// to the chain walk - the next clause in the chain is not a
		// candidate, it just happens to be adjacent.

		q->st.iter_single = false;
		q->st.dbe = NULL;
		return;
	}

	if (!q->st.iter) {
		q->st.dbe = q->st.dbe->next;
		return;
	}

	if (!sl_next(q->st.iter, (void*)&q->st.dbe)) {
		q->st.dbe = NULL;

		// Drop the handle but do NOT free: release_prefetch() frees when
		// the owning choice goes, and that covers this. Measured with a
		// made/freed counter across the whole suite plus a cut-heavy and
		// an exhaustion-heavy workload - 4000 prefetches, made == freed,
		// nothing leaked. Freeing here as well was the second owner, and
		// needed the alias-clearing dance below it to stay safe.

		q->st.iter = NULL;
	}
}

bool has_next_key(query *q)
{
	if (q->st.iter_single)
		return false;

	if (q->st.iter)
		return sl_has_next(q->st.iter, NULL);

	if (!q->st.dbe->next)
		return false;

	if (!q->st.key->arity)
		return true;

	if (q->st.dbe->cl.is_unique) {
		if ((q->st.key->arity == 1) && q->st.karg1_is_atomic)
			return false;

		if ((q->st.key->arity == 2) && q->st.karg1_is_atomic && q->st.karg2_is_atomic)
			return false;

		if ((q->st.key->arity == 3) && q->st.karg1_is_atomic && q->st.karg2_is_atomic && q->st.karg3_is_atomic)
			return false;
	}

	cell *karg1 = FIRST_ARG(q->st.key), *karg2 = NULL, *karg3 = NULL;
	cell *save_arg1 = karg1;

	if (q->st.karg1_is_ground)
		karg1 = deref(q, save_arg1, q->st.key_ctx);

	if (q->st.karg2_is_ground)
		karg2 = deref(q, NEXT_ARG(save_arg1), q->st.key_ctx);

	if (q->st.karg3_is_ground)
		karg3 = deref(q, NEXT_ARG(NEXT_ARG(save_arg1)), q->st.key_ctx);

	//DUMP_TERM("key ", q->st.key, q->st.key_ctx, 1);

	for (rule *next = q->st.dbe->next; next; next = next->next) {
		cell *dkey = next->cl.cells;

		if ((dkey->val_off == g_neck_s) && (dkey->arity == 2))
			dkey++;

		//DUMP_TERM("next", dkey, q->st.cur_ctx, 0);

		if (karg1) {
			if (index_cmpkey(karg1, FIRST_ARG(dkey), q->st.m, NULL) != 0)
				continue;
		}

		if (karg2) {
			if (index_cmpkey(karg2, NEXT_ARG(FIRST_ARG(dkey)), q->st.m, NULL) != 0)
				continue;
		}

		if (karg3) {
			if (index_cmpkey(karg3, NEXT_ARG(NEXT_ARG(FIRST_ARG(dkey))), q->st.m, NULL) != 0)
				continue;
		}

		if (index_cmpkey(q->st.key, dkey, q->st.m, NULL) == 0)
			return true;
	}

	return false;
}

static bool expand_meta_predicate(query *q, predicate *pr)
{
	int arity = q->st.key->arity;
	cell *tmp = alloc_heap(q, q->st.key->num_cells*3);	// allocate max possible
	CHECKED(tmp);
	cell *save_tmp = tmp;
	tmp += copy_cells(tmp, q->st.key, 1);

	// Expand module-sensitive args...

	for (cell *k = q->st.key+1, *m = pr->meta_args+1; arity--; k += k->num_cells, m += m->num_cells) {
		cell *k0 = deref(q, k, q->st.key_ctx);

		if ((k0->arity == 2) && (k0->val_off == g_colon_s) && is_atom(FIRST_ARG(k0)))
			;
		else if (!is_interned(k0) || is_iso_list(k0))
			;
		else if (is_interned(k0) && ((k0->val_off == g_call_s) || (k0->val_off == g_once_s) || (k0->val_off == g_ignore_s)))
			;
		else if (is_interned(m) && (m->val_off == g_colon_s)) {
			make_instr(tmp, g_colon_s, bif_iso_qualify_2, 2, 1+k->num_cells);
			SET_OP(tmp, OP_XFY); tmp++;
			make_atom(tmp++, new_atom(q->pl, q->st.m->name));
		} else if (is_smallint(m) && is_positive(m) && (get_smallint(m) <= 9)) {
			make_instr(tmp, g_colon_s, bif_iso_qualify_2, 2, 1+k->num_cells);
			SET_OP(tmp, OP_XFY); tmp++;
			make_atom(tmp++, new_atom(q->pl, q->st.m->name));
		}

		tmp += dup_cells_by_ref(tmp, k, q->st.key_ctx, k->num_cells);
	}

	save_tmp->num_cells = tmp - save_tmp;
	q->st.key = save_tmp;
	return true;
}

// --index-check: verify an indexed lookup against the walk it replaced.
//
// The index is allowed to be WIDER than the linear scan - an
// approximation must widen, never narrow - so this asserts the candidate
// set is a SUPERSET of the clauses the linear walk would have offered,
// and says nothing about extras. A narrowing is a wrong-answer bug; a
// widening is only wasted unification.

int g_index_check = 0;
unsigned long g_index_check_lookups = 0, g_index_check_bad = 0;

// The candidate set is snapshotted during the index walk rather than
// read back off the prefetch skiplist. Reading it back is not possible:
// the prefetch is an is_tmp_list, and sl_done() DESTROYS one of those
// rather than recycling the iterator - so merely iterating it to count
// entries freed the list the live query was about to use. The first
// version of this check did exactly that and reported all 1639 entries
// of a 1639-entry set as missing, which is how it was caught.

static bool in_candidates(const rule **got, unsigned num_got, const rule *c)
{
	for (unsigned i = 0; i < num_got; i++) {
		if (got[i] == c)
			return true;
	}

	return false;
}

static void index_check(query *q, predicate *pr, cell *goal, cell *key,
	const rule **got, unsigned num_got, int idx_arg)
{
	// The goal's own generation, NOT GET_CURR_FRAME()->dbgen. find_key()
	// runs before check_frame(), so the current frame is still the
	// CALLER'S and its dbgen predates every clause asserted since the
	// caller was entered. enter_predicate() stamps q->pl->dbgen onto the
	// frame immediately after this, and that is the view the matching
	// loop will use. Getting this wrong is silent: can_view() rejects
	// every clause, the walk finds no candidates to compare against and
	// the check passes everything. It did, until a deliberately
	// mis-filed clause failed to raise it.

	const uint64_t dbgen = q->pl->dbgen;
	unsigned missing = 0;

	g_index_check_lookups++;

	for (const rule *c = pr->head; c; c = c->next) {
		// Same visibility test the matching loop uses, or this reports
		// clauses the goal was never entitled to see.

		if (!can_view(q, dbgen, c))
			continue;

		cell *ch = get_head(((rule*)c)->cl.cells);
		cell *ck = ch;

		// Each index is keyed on its recorded argument, so each has to be
		// checked against the key it was actually built on. The clause
		// HEAD is still what gets printed - an arg on its own says
		// little about which clause went missing.

		if (idx_arg >= 0 && ch->arity)
			ck = get_nth_arg(ch, idx_arg);

		if (index_cmpkey(ck, key, q->st.m, NULL) != 0)
			continue;

		if (in_candidates(got, num_got, c))
			continue;

		if (!missing) {
			fprintf(stderr, "\n*** index-check FAILED for %s/%u (%s)\n",
				C_STR(q, &pr->key), pr->key.arity,
				idx_arg < 0 ? "head" : "argument");
			fprintf(stderr, "***   goal   ");
			DUMP_TERM("", goal, q->st.cur_ctx, 1);
		}

		fprintf(stderr, "***   MISSING db_id=%llu  ",
			(unsigned long long)c->db_id);
		DUMP_TERM("", ch, q->st.cur_ctx, 1);

		// Is the clause reachable by its OWN key? That separates a
		// descent fault from an insertion fault: if the index cannot
		// find it when handed that clause's exact key, the entry is not
		// where the comparator would put it - mis-filed on insert, or
		// lost on a removal. If it IS reachable, the ordering is fine
		// and the query descent went astray.

		sliter *probe = sl_find_key(idx_arg < 0 ? pr->idx0 : idx_arg ? pr->idx2 : pr->idx1, ck);
		const rule *probe_r;
		bool self = false;

		while (probe && sl_next_key(probe, (void*)&probe_r)) {
			if (probe_r == c) {
				self = true;
				break;
			}
		}

		if (probe)
			sl_done(probe);

		fprintf(stderr, "***     reachable by its own key: %s\n",
			self ? "YES (ordering ok, query descent went astray)"
			     : "NO (mis-filed on insert, or lost on removal)");
		fprintf(stderr, "***     cmp(clause,goal)=%d\n",
			index_cmpkey(ck, key, q->st.m, NULL));
		missing++;
	}

	if (missing) {
		fprintf(stderr, "***   indexed set had %u entr%s, %u missing\n",
			num_got, num_got == 1 ? "y" : "ies", missing);
		fprintf(stderr, "***   predicate has %u clauses, head=%s idx1=%s idx2(arg%u)=%s\n",
			(unsigned)pr->cnt, pr->idx0 ? "yes" : "no", pr->idx1 ? "yes" : "no", pr->idx2_arg + 1,
			pr->idx2 ? "yes" : "no");

		// Carry on rather than abort: one run should surface every
		// mismatch in a suite, not just the first. The count at exit is
		// what makes a clean sweep meaningful - zero mismatches over
		// zero verified lookups says nothing at all.

		g_index_check_bad++;
	}
}

static bool find_key(query *q, predicate *pr, cell *key, pl_ctx key_ctx)
{
	q->st.iter = NULL;
	q->st.iter_single = false;
	q->st.karg1_is_ground = q->st.karg2_is_ground = q->st.karg3_is_ground = false;
	q->st.karg1_is_atomic = q->st.karg2_is_atomic = q->st.karg3_is_atomic = false;
	q->st.key = key;
	q->st.key_ctx = key_ctx;

	if (!pr->idx1) {
		q->st.dbe = pr->head;

		if (key->arity) {
			if (pr->is_meta_predicate) {
				if (!expand_meta_predicate(q, pr))
					return false;
			}

			setup_key(q);
		}

		return true;
	}

	INDEX_PROFILE_START(pr);

	if (pr->is_meta_predicate) {
		if (!expand_meta_predicate(q, pr))
			return false;

		key = q->st.key;
		key_ctx = q->st.cur_ctx;
	} else {
		CHECKED(init_tmp_heap(q));
		key = clone_term_to_tmp(q, key, key_ctx);
		key_ctx = q->st.cur_ctx;
	}

	cell *arg1 = key->arity ? FIRST_ARG(key) : NULL;
	skiplist *idx = pr->idx1;
	cell *goal = key;
	int idx_arg = 0;

	// A full-head lookup is exact and much more selective when every
	// clause head is ground. Variable-headed clauses stay out of idx0,
	// because a var-equals-anything comparator is not a skiplist order.
	if (pr->idx0 && !pr->is_var_in_head && is_ground(key)) {
		idx = pr->idx0;
		idx_arg = -1;
		INDEX_PROFILE_MODE(ip, idx0);
	} else if (pr->idx2 && (pr->idx2_arg == 1) && !pr->is_var_in_idx2_arg
			&& is_interned(&pr->key) && !strcmp(C_STR(q, &pr->key), "$predicate_property")) {
		// $predicate_property/3 is populated as
		// $predicate_property(predicate, Name(_, ...), Property).
		// Arg1 consequently has only the two category values (predicate and
		// function), while Arg2 identifies the actual predicate.  Treating
		// this internal catalogue like an ordinary first-arg-indexed relation
		// turns every predicate lookup into a walk of the whole catalogue.
		// idx2 is already maintained precisely for this argument and supports
		// the variable arguments in Name(_, ...) through index_cmpkey's
		// wildcard handling.

		cell *arg2 = get_nth_arg(key, pr->idx2_arg);

		if (!is_var(arg2)) {
			key = arg2;
			idx = pr->idx2;
			idx_arg = pr->idx2_arg;
			INDEX_PROFILE_MODE(ip, idx2);
		} else if (arg1 && (is_var(arg1) || pr->is_var_in_first_arg)) {
			INDEX_PROFILE_MODE(ip, linear);
			INDEX_PROFILE_CANDIDATES(ip, pr->cnt);
			q->st.dbe = pr->head;
			return true;
		} else if (arg1) {
			key = arg1;
			INDEX_PROFILE_MODE(ip, idx1);
		}
	} else if (arg1 && (is_var(arg1) || pr->is_var_in_first_arg)) {
		// idx2 is a floating later-argument index. If Arg1 is unusable,
		// use it when the selected argument in this goal is ground.

		if (!pr->idx2 || pr->is_var_in_idx2_arg) {
			INDEX_PROFILE_MODE(ip, linear);
			INDEX_PROFILE_CANDIDATES(ip, pr->cnt);
			q->st.dbe = pr->head;
			return true;
		}

		cell *arg2 = get_nth_arg(key, pr->idx2_arg);

		if (is_var(arg2)) {
			INDEX_PROFILE_MODE(ip, linear);
			INDEX_PROFILE_CANDIDATES(ip, pr->cnt);
			q->st.dbe = pr->head;
			return true;
		}

		key = arg2;
		idx = pr->idx2;
		idx_arg = pr->idx2_arg;
		INDEX_PROFILE_MODE(ip, idx2);
	} else if (arg1) {
		// idx1 is keyed on Arg1 only (see assert_commit).
		key = arg1;
		INDEX_PROFILE_MODE(ip, idx1);
	}

	if (!arg1) {
		INDEX_PROFILE_MODE(ip, idx1);
	}

	q->st.dbe = NULL;
	sliter *iter;

	if (!(iter = sl_find_key(idx, key))) {
		if (g_index_check)
			index_check(q, pr, goal, key, NULL, 0, idx_arg);

		return false;
	}

	// If the index search has found just one (definite) solution
	// then we can use it with no problems. If more than one then
	// results must be returned in database order, so prefetch all
	// the results and return them sorted as an iterator...

	// Hold the first hit back rather than materialising unconditionally.
	// Most index lookups match exactly one clause - every retract in a
	// key-per-clause table does - and for those there is nothing to sort
	// and nothing to own. Building a temporary skiplist for a single
	// entry cost an allocation per call and, because the iterator is
	// snapshotted into every choicepoint and so cannot safely be freed
	// on a cut, it leaked: 4.7MB over 14000 retracts, measured.

	skiplist *tmp_idx = NULL;
	const rule *first = NULL;
	const rule *r;
	const rule **got = NULL;
	unsigned num_got = 0, max_got = 0;

	while (sl_next_key(iter, (void*)&r)) {
		INDEX_PROFILE_CANDIDATES(ip, 1);
		if (g_index_check) {
			if (num_got == max_got) {
				max_got = max_got ? max_got * 2 : 32;
				got = TPL_realloc(got, max_got * sizeof(*got));
			}

			got[num_got++] = r;
		}

		if (!first) {
			first = r;
			continue;
		}

		if (!tmp_idx) {
			tmp_idx = sl_create(NULL, NULL, NULL);
			sl_set_tmp(tmp_idx);
			sl_app(tmp_idx, (void*)(size_t)first->db_id, (void*)first);
		}

		sl_app(tmp_idx, (void*)(size_t)r->db_id, (void*)r);
	}

	sl_done(iter);

	if (g_index_check) {
		index_check(q, pr, goal, key, got, num_got, idx_arg);
		TPL_free(got);
	}

	if (!first)
		return false;

	if (!tmp_idx) {
		q->st.dbe = (rule*)first;
		q->st.iter = NULL;
		q->st.iter_single = true;
		return true;
	}

	// More than one: results must come back in database order, so the
	// prefetch stands.

	iter = sl_first(tmp_idx);

	if (!sl_next(iter, (void*)&q->st.dbe)) {
		sl_done(iter);
		return false;
	}

	q->st.iter = iter;

	// The goal's alternatives choicepoint has not been raised yet -
	// find_key() runs first - so it will land on the slot q->st.cp is
	// pointing at now. That choice owns the prefetch.

	q->st.iter_owner = q->st.cp;
	return true;
}

// Match HEAD :- BODY.

bool match_rule(query *q, cell *p1, pl_ctx p1_ctx, enum clause_type is_retract)
{
	if (!q->retry) {
		cell *c = deref(q, get_head(p1), p1_ctx);
		pl_ctx c_ctx = q->latest_ctx;
		predicate *pr = NULL;

		if (is_interned(c))
			pr = c->match;
		else if (is_cstring(c))
			convert_to_literal(q->st.m, c);

		if (pr && pr->is_abolished)
			pr = search_predicate(q->st.m, c);

		if (!pr || is_evaluable(c) || is_builtin(c)) {
			pr = search_predicate(q->st.m, c);

			if (pr)
				c->match = pr;
		}

		if (!pr) {
			bool found = false;

			if (get_builtin_term(q->st.m, c, &found, NULL), found)
				return throw_error(q, c, c_ctx, "permission_error", "modify,static_procedure");

			q->st.dbe = NULL;
			return false;
		}

		if (pr->alias) {
			c->val_off = pr->alias->key.val_off;
			pr = pr->alias;
		}

		if (!pr->is_dynamic)
			return throw_error(q, c, c_ctx, "permission_error", "modify,static_procedure");

		find_key(q, pr, c, c_ctx);
		enter_predicate(q, pr);
	} else {
		next_key(q);
	}

	if (!q->st.dbe) {
		leave_predicate(q, q->st.pr, true);
		return false;
	}

	const frame *f = GET_CURR_FRAME();
	cell *p1_body = deref(q, get_logical_body(p1), p1_ctx);
	cell *orig_p1 = p1;

	for (; q->st.dbe; q->st.dbe = q->st.dbe->next) {
		if (!can_view(q, f->dbgen, q->st.dbe))
			continue;

		CHECKED(push_choice(q));
		clause *cl = &q->st.dbe->cl;
		cell *c = cl->cells;
		bool needs_true = false;
		p1 = orig_p1;

		cell *tmp = import_term(q, c, q->st.cur_ctx);
		CHECKED(tmp);
		c = tmp;
		cell *head = get_head(c);
		const cell *c_body = get_logical_body(c);

		if (p1_body && is_var(p1_body) && !c_body) {
			p1 = deref(q, get_head(p1), p1_ctx);
			c = get_head(tmp);
			needs_true = true;
		}

		if (unify(q, p1, p1_ctx, c, q->st.cur_ctx)) {
			if (q->did_throw)
				return true;

			int ok;

			if (needs_true) {
				p1_body = deref(q, p1_body, p1_ctx);
				pl_ctx p1_body_ctx = q->latest_ctx;
				cell tmp;
				make_instr(&tmp, g_true_s, bif_iso_true_0, 0, 0);
				ok = unify(q, p1_body, p1_body_ctx, &tmp, q->st.cur_ctx);
				if (q->did_throw)
					return true;
			} else
				ok = true;

			return ok;
		}

		retry_choice(q);
	}

	leave_predicate_and_drop(q, q->st.pr, true);
	return false;
}

// Match HEAD.
// Match HEAD :- true.

bool match_clause(query *q, cell *p1, pl_ctx p1_ctx, cell **ret_body, enum clause_type is_retract)
{
	if (!q->retry) {
		cell *c = p1;
		pl_ctx c_ctx = p1_ctx;
		predicate *pr = NULL;

		if (is_interned(c))
			pr = c->match;
		else if (is_cstring(c))
			convert_to_literal(q->st.m, c);

		if (pr && pr->is_abolished)
			pr = search_predicate(q->st.m, c);

		if (!pr || is_evaluable(c) || is_builtin(c)) {
			pr = search_predicate(q->st.m, c);

			if (pr)
				c->match = pr;
		}

		if (!pr) {
			bool found = false;

			if (get_builtin_term(q->st.m, p1, &found, NULL), found) {
				if (is_retract != DO_CLAUSE)
					return throw_error(q, p1, p1_ctx, "permission_error", "modify,static_procedure");
				else
					return throw_error(q, p1, p1_ctx, "permission_error", "access,private_procedure");
			}

			q->st.dbe = NULL;
			return false;
		}

		if (pr->alias) {
			c->val_off = pr->alias->key.val_off;
			pr = pr->alias;
		}

		if (!pr->is_dynamic) {
			if (is_retract == DO_CLAUSE) {
				if (!q->access_private)
					return throw_error(q, p1, p1_ctx, "permission_error", "access,private_procedure");
			} else
				return throw_error(q, p1, p1_ctx, "permission_error", "modify,static_procedure");
		}

		find_key(q, pr, c, c_ctx);
		enter_predicate(q, pr);
	} else {
		next_key(q);
	}

	if (!q->st.dbe) {
		leave_predicate(q, q->st.pr, true);
		return false;
	}

	const frame *f = GET_CURR_FRAME();

	for (; q->st.dbe; q->st.dbe = q->st.dbe->next) {
		if (!can_view(q, f->dbgen, q->st.dbe))
			continue;

		clause *cl = &q->st.dbe->cl;
		cell *c = cl->cells;
		cell *body = get_logical_body(c);

		// retract(HEAD) should ignore rules (and directives)

		if ((is_retract == DO_RETRACT) && body)
			continue;

		CHECKED(push_choice(q));
		cell *tmp = import_term(q, c, q->st.cur_ctx);
		CHECKED(tmp);
		cell *head = get_head(tmp);
		body = get_body(tmp);

		if (unify(q, p1, p1_ctx, head, q->st.cur_ctx)) {
			if (q->did_throw)
				return true;

			if (ret_body)
				*ret_body = body;

			return true;
		}

		retry_choice(q);
	}

	leave_predicate(q, q->st.pr, true);
	return false;
}

bool match_head(query *q)
{
	if (!q->retry) {
		cell *c = q->st.instr;
		pl_ctx c_ctx = q->st.cur_ctx;
		predicate *pr = NULL;

		if (is_interned(c))
			pr = c->match;
		else if (is_cstring(c)) {
			convert_to_literal(q->st.m, c);
		}

		if (pr && pr->is_abolished)
			pr = search_predicate(q->st.m, c);

		if (!pr || is_evaluable(c) || is_builtin(c)) {
			pr = search_predicate(q->st.m, c);

			if (pr) {
				c->match = pr;
				// Keep NEXT_CUT / TCO hints; only drop builtin tags.
				c->flags &= ~(FLAG_INTERNED_BUILTIN | FLAG_INTERNED_EVALUABLE);
			}
		}

		if (!pr) {
			if (!is_end(c) && !(is_interned(c) && !strcmp(C_STR(q, c), "initialization"))) {
				if (q->st.m->flags.unknown == UNK_ERROR)
					return throw_error(q, c, c_ctx, "existence_error", "procedure");
				return false;
			} else
				q->error = true;

			return false;
		}

		if (pr->alias) {
			c->val_off = pr->alias->key.val_off;
			pr = pr->alias;
		}

		// A predicate that exists in the module but has no clauses and is
		// neither dynamic nor multifile (e.g. a static predicate left empty
		// after a file reconsult removed its last clause) must be treated as
		// an undefined procedure and honor the `unknown` flag, rather than
		// silently failing.
		if (!pr->head && !pr->is_dynamic && !pr->is_multifile && !pr->is_discontiguous && !pr->is_builtin) {
			if (!is_end(c) && !(is_interned(c) && !strcmp(C_STR(q, c), "initialization"))) {
				if (q->st.m->flags.unknown == UNK_ERROR)
					return throw_error(q, c, c_ctx, "existence_error", "procedure");
				return false;
			} else {
				q->error = true;
				return false;
			}
		}

		find_key(q, pr, c, c_ctx);
		enter_predicate(q, pr);
	} else
		next_key(q);

	if (!q->st.dbe) {
		leave_predicate(q, q->st.pr, true);
		return false;
	}

	CHECKED(check_frame(q, q->st.pr->max_vars));
	CHECKED(push_choice(q));
	const frame *f = GET_CURR_FRAME();

	for (; q->st.dbe; next_key(q)) {
		if (!can_view(q, f->dbgen, q->st.dbe))
			continue;

		clause *cl = &q->st.dbe->cl;
		cell *head = get_head(cl->cells);

		if (cl->num_vars > q->st.pr->max_vars)
			CHECKED(check_slot(q, q->st.pr->max_vars=cl->num_vars));

		try_me(q, cl->num_vars);
		q->st.dbe->attempted++;

		if (unify(q, q->st.key, q->st.key_ctx, head, q->st.fp)) {
			// throw_error() returns true when a handler was armed (via
			// did_throw). Must not commit_frame — that would overwrite the
			// recover goal. find_exception_handler already unwound to the
			// catcher.
			if (q->did_throw)
				return true;

			// Take it now, while it still belongs to the unify above.
			const bool head_has_vars = q->has_vars;

			if (q->error)
				break;

			commit_frame(q, head_has_vars);
			return true;
		}

		undo_me(q);
	}

	leave_predicate_and_drop(q, q->st.pr, true);
	return false;
}

static bool any_outstanding_choices(query *q)
{
	while (q->st.cp) {
		const choice *ch = GET_CURR_CHOICE();

		if (!ch->barrier)
			break;

		pop_choice(q);
	}

	return q->st.cp > 0;
}

void do_cleanup(query *q, cell *c, pl_ctx c_ctx)
{
	cell *tmp = prepare_call(q, CALL_NOSKIP, c, c_ctx, 4);
	ENSURE(tmp);
	pl_idx num_cells = c->num_cells;
	make_instr(tmp+num_cells++, g_cut_s, bif_iso_cut_0, 0, 0);
	make_instr(tmp+num_cells++, g_sys_drop_barrier_s, bif_sys_drop_barrier_1, 1, 1);
	make_uint(tmp+num_cells++, q->st.cp);
	make_call(q, tmp+num_cells);
	q->st.instr = tmp;
}

static bool consultall(query *q, cell *l, pl_ctx l_ctx)
{
	if (is_cyclic_term(q, l, l_ctx))
		return throw_error(q, l, l_ctx, "type_error", "callable");

	PROLOG_LIST_HANDLER(l);

	while (is_list(l)) {
		cell *h = PROLOG_LIST_HEAD(l);
		h = deref(q, h, l_ctx);
		pl_ctx h_ctx = q->latest_ctx;

		if (is_list(h)) {
			if (consultall(q, h, h_ctx) != true)
				return false;
		} else {
			do_load_file(q, h, h_ctx);
		}

		l = PROLOG_LIST_TAIL(l);
		l = deref(q, l, l_ctx);
		l_ctx = q->latest_ctx;
	}

	return true;
}

bool start(query *q)
{
	q->yielded = false;
	bool done = false;

	while (!done && !q->error) {
		if (interrupt_pending(q)) {
			switch (check_interrupt(q)) {
				case 1: return true;
				case -1: q->retry = true;
				default: continue;
			}
		}

#if USE_THREADS
		if (q->thread_ptr) {
			thread *t = q->thread_ptr;

			if (list_count(&t->signals)) {
				do_signal(q, t);
				proceed(q);
			}
		}
#endif

		if (q->retry) {
			switch (retry_choice(q)) {
				case 0: done = true; continue;
				case -1: proceed(q); goto MORE;
				case -2: q->retry = false; break;
			}
		}

		if (!is_callable(q->st.instr)
			&& (q->run_init || !is_list(q->st.instr))) {
			cell *p1 = deref(q, q->st.instr, q->st.cur_ctx);
			pl_ctx p1_ctx = q->latest_ctx;

			if (!bif_call_0(q, p1, p1_ctx)) {
				if (is_var(p1))
					break;

				continue;
			}
		}

		Trace(q, q->st.instr, q->st.cur_ctx, CALL);
		cell *save_cell = q->st.instr;
		pl_ctx save_ctx = q->st.cur_ctx;
		q->cycle_error = q->did_throw = false;
		q->total_goals++;

		if (is_builtin(q->st.instr)) {
			q->total_inferences++;
			bool status;

#if USE_FFI
			if (q->st.instr->bif_ptr->ffi) {
				if (q->st.instr->bif_ptr->evaluable)
					status = wrap_ffi_function(q, q->st.instr->bif_ptr);
				else
					status = wrap_ffi_predicate(q, q->st.instr->bif_ptr);
			} else
#endif
				status = q->st.instr->bif_ptr->fn(q);

			if (q->retry == QUERY_NOOP) {
				q->retry = QUERY_OK;
				continue;
			}

			// throw_error armed a recover goal (noskip). Do not treat
			// status as ordinary success/failure of save_cell.
			if (q->did_throw) {
				proceed(q);
				goto MORE;
			}

			if (!(q->total_goals % YIELD_INTERVAL)) {
				q->s_cnt = 0;

				if (!(q->s_cnt++ % 10000))
					check_pressure(q);

				if (q->yield_at && !q->run_hook) {
					uint64_t now = wall_time_in_usec() / 1000;

					if (now > q->yield_at)  {
						do_yield_then(q, status);
						break;
					}
				}
			}

			if (!status || q->abort) {
				Trace(q, q->st.instr, q->st.cur_ctx, FAIL);
				q->retry = QUERY_RETRY;

				if (q->yielded)
					break;

				q->total_backtracks++;
				continue;
			}

			if (q->run_hook)
				do_post_unify_hook(q, true);

			Trace(q, save_cell, save_ctx, EXIT);
			proceed(q);
		} else if (!q->run_init && is_list(q->st.instr)) {
			if (!consultall(q, q->st.instr, q->st.cur_ctx)) {
				Trace(q, q->st.instr, q->st.cur_ctx, FAIL);
				q->retry = QUERY_RETRY;
				q->total_backtracks++;
				continue;
			}

			Trace(q, save_cell, save_ctx, EXIT);
			proceed(q);
		} else {
			q->total_inferences++;

			if (!match_head(q)) {
				Trace(q, q->st.instr, q->st.cur_ctx, FAIL);
				q->retry = QUERY_RETRY;
				q->total_backtracks++;
				continue;
			}

			// Exception handler armed during head unify — skip hooks;
			// recover goal is already in st.instr (noskip).
			if (q->did_throw) {
				proceed(q);
				goto MORE;
			}

			if (q->run_hook)
				do_post_unify_hook(q, false);
		}

		MORE:

		q->retry = QUERY_OK;

		while (!q->st.instr || is_end(q->st.instr)) {
			if (resume_frame(q)) {
				proceed(q);
				continue;
			}

			if (q->top && !q->run_init && any_outstanding_choices(q)) {
				if (!check_redo(q))
					break;

				return true;
			}

			done = q->status = true;
			break;
		}

		if (q->oom) {
			q->error = true;
			printf("\nresource_error(memory). %%query terminated\n");
			break;
		}
	}

	if (q->halt)
		q->error = false;
	else if (q->do_dump_vars && !q->abort && q->status && !q->error)
		dump_vars(q, false);

	return true;
}

bool execute(query *q, cell *cells, unsigned num_vars)
{
	q->retry = q->halt = q->error = q->abort = false;
	q->pl->did_dump_vars = false;
	q->st.instr = cells;
	q->st.sp = num_vars;
	q->is_redo = false;

	// There is an initial frame (fp=0), so this
	// to the next available frame...

	q->st.fp = 1;

	frame *f = GET_FRAME(0);
	f->initial_slots = f->actual_slots = num_vars;
	f->dbgen = ++q->pl->dbgen;
	return start(q);
}

void query_destroy(query *q)
{
	if (!q)
		return;

	q->done = true;

	for (page *a = q->heap_pages; a;) {
		cell *c = a->cells;

		for (pl_idx i = 0; i < a->idx; i++, c++)
			unshare_cell(c);

		page *save = a;
		a = a->next;
		TPL_free(save->cells);
		TPL_free(save);
	}

	slot *e = q->slots;

	for (pl_idx i = 0; i < q->st.sp; i++, e++) {
		cell *c = &e->c;
		unshare_cell(c);
	}

	for (int i = 0; i < MAX_QUEUES; i++) {
		cell *c = q->queue[i];
		for (pl_idx j = 0; j < q->qp[i]; j++, c++)
			unshare_cell(c);

		TPL_free(q->queue[i]);
	}

	while (q->tasks) {
		query *task = q->tasks->next;
		query_destroy(q->tasks);
		q->tasks = task;
	}

	sched_destroy(q);

	// Choicepoints still live at teardown hold undo items of their own.
	// Draining q->undo alone left them behind, so a query that halted -
	// or simply succeeded - with choicepoints outstanding leaked
	// whatever they were holding. Deepest first, the order backtracking
	// would have taken.

	for (pl_idx i = q->st.cp; i > 0; i--)
		undo_list_drain(&GET_CHOICE(i - 1)->undo);

	undo_list_drain(&q->undo);

	mp_int_clear(&q->tmp_ival);
	mp_rat_clear(&q->tmp_irat);
	query_purge_dirty_list(q);
	parser_destroy(q->p);
	for (trail_page *a = q->trail_pages; a;) {
		trail_page *save = a;
		a = a->next;
		TPL_free(save->entries);
		TPL_free(save);
	}
	for (choice_page *a = q->choice_pages; a;) {
		choice_page *save = a;
		a = a->next;
		TPL_free(save->entries);
		TPL_free(save);
	}
	TPL_free(q->slots);
	for (pl_idx i = 0; i < q->frame_pages_size; i++)
		TPL_free(q->frame_pages[i]);
	TPL_free(q->frame_pages);
	TPL_free(q->tmp_heap);
	TPL_free(q->tabs);
	TPL_free(q->unify_seen);
	release_oom_reserve(q);
	q->pl->q_cnt--;
	TPL_free(q);
}

static query *query_create_(module *m, bool is_toplevel)
{
	static pl_atomic uint64_t g_query_id = 0;

#ifdef INDEX_PROFILE
	if (!g_index_profile_registered) {
		g_index_profile_registered = true;
		atexit(index_profile_report);
	}
#endif

	query *q = TPL_calloc(1, sizeof(query));
	ENSURE(q);
	q->p = parser_create(m);
	q->p->q = q;

	if (!g_query_id) {
		m->pl->threads[0].q = q;
		m->pl->threads[0].is_active = true;
	}

	q->qid = g_query_id++;
	q->pl = m->pl;
	q->pl->q_cnt++;
	q->st.m = m;
	q->trace = m->pl->trace;
	q->flags = m->flags;
	q->get_started = wall_time_in_usec();
	q->cpu_time = q->time_cpu_last_started = q->st.cpu_time = cpu_time_in_usec();
	q->ops_dirty = true;
	q->max_depth = m->pl->def_max_depth;
	q->vgen = 1;
	q->dump_var_num = -1;
	q->dump_var_ctx = -1;
	q->double_quotes = false;

#ifndef __wasi__
	q->rand_seed = getpid() + g_query_id;
#else
	q->rand_seed = clock() + g_query_id;
#endif

	//if (is_threaded) q->trace = 1;

	mp_int_init(&q->tmp_ival);
	mp_rat_init(&q->tmp_irat);

	// Allocate these now...

	q->slots_size = INITIAL_NBR_SLOTS;

	q->frame_pages_size = 1;
	ENSURE(q->frame_pages = TPL_calloc(q->frame_pages_size, sizeof(frame *)), NULL);
	ENSURE(q->frame_pages[0] = TPL_calloc(FRAME_PAGE_SIZE, sizeof(frame)), NULL);
	for (unsigned i = 0; i < FRAME_PAGE_SIZE; i++)
		q->frame_pages[0][i].idx = i;
	ENSURE(q->slots = TPL_calloc(q->slots_size, sizeof(slot)), NULL);

	// Allocate these later as needed...

	q->heap_size = INITIAL_NBR_HEAP_CELLS;
	q->tmph_size = INITIAL_NBR_CELLS;

	for (int i = 0; i < MAX_QUEUES; i++)
		q->q_size[i] = INITIAL_NBR_QUEUE_CELLS;

	frame *f = GET_CURR_FRAME();
	f->prev = CTX_NUL;

	rearm_oom_reserve(q);
	clear_write_options(q);
	return q;
}

query *query_create(module *m)
{
	return query_create_(m, true);
}

query *query_create_threaded(module *m)
{
	query *t = query_create_(m, false);
	t->is_thread = true;
	return t;
}

query *query_create_subquery(query *q, cell *instr)
{
	query *subq = query_create_(q->st.m, false);
	if (!subq) return NULL;
	subq->parent = q;
	subq->st.fp = 1;
	subq->top = q->top;

	cell *tmp = prepare_call(subq, false, instr, q->st.cur_ctx, 1);
	pl_idx num_cells = tmp->num_cells;
	make_end(tmp+num_cells);
	subq->st.instr = tmp;

	frame *fsrc = GET_FRAME(q->st.cur_ctx);
	frame *fdst = get_frame(subq, 0);
	fdst->initial_slots = fdst->actual_slots = fsrc->actual_slots;
	fdst->dbgen = ++q->pl->dbgen;
	subq->st.sp = fdst->actual_slots;
	return subq;
}

query *query_create_task(query *q, cell *instr)
{
	query *t = query_create_subquery(q, instr);
	if (!t) return NULL;
	t->is_task = true;
	return t;
}

// For a goal that has already been cloned and rebased into a numbering
// of its own. query_create_subquery() copies by reference against the
// caller's context, and a context is just a frame index - meaningless in
// a query with its own frames, which is why a caller's bindings never
// reached the task. Here the cells are taken as they stand and the
// frame is sized from the goal itself, the way execute() does it for a
// thread.

query *query_create_task_rebased(query *q, cell *instr, unsigned num_vars)
{
	query *subq = query_create_(q->st.m, false);
	if (!subq) return NULL;
	subq->parent = q;
	subq->st.fp = 1;
	subq->top = q->top;
	subq->is_task = true;

	pl_idx num_cells = instr->num_cells;
	cell *tmp = alloc_heap(subq, num_cells+1);

	if (!tmp) {
		query_destroy(subq);
		return NULL;
	}

	dup_cells(tmp, instr, num_cells);
	make_end(tmp+num_cells);
	subq->st.instr = tmp;

	frame *fdst = get_frame(subq, 0);
	fdst->initial_slots = fdst->actual_slots = num_vars;
	fdst->dbgen = ++q->pl->dbgen;
	subq->st.sp = num_vars;
	return subq;
}