Vito
89d94cdb2b
* fix(history): join outer-whitespace raw text to the history of the prompt it saves (#438) The Library stores text.strip() while a run keeps its exact raw text. The history window, run detail scope id, newest pair run and prompt hints now compare the raw text's stored form (one rule, utils/hashing.py, mirrored in SQLite with the same code points). Hints also keep matching the exact raw text, so older rows that kept outer whitespace stay hinted and protected. * fix(history): join case-variant raw text to the history of the prompt it counts toward (#460) The prompt hash folds case, so a run of "A Lighthouse" counts toward the saved "a lighthouse" but never joined its history. History now compares the hash's own key (prompt_key = strip + str.lower()); SQL calls Python for it through pm_prompt_key/pm_prompt_hash registered on every managed connection, since SQLite's lower() folds ASCII only. Hints match by text first (#438) and fall back to the dedup hash only when no text matches, so no existing hint becomes ambiguous. A history position from before this change still resumes. * test(history): a real-host case-variant re-queue joins the prompt's history (#460)
444 lines
17 KiB
Python
444 lines
17 KiB
Python
"""Immutable execution capture and exact, epoch-fenced library navigation hints."""
|
|
|
|
from contextlib import ExitStack
|
|
import json
|
|
import sqlite3
|
|
from typing import Callable, ContextManager, Iterator
|
|
|
|
try:
|
|
from ..utils.execution_history_events import RunFinish, encode_command
|
|
from ..utils.hashing import (
|
|
generate_prompt_hash,
|
|
stored_prompt_text,
|
|
stored_prompt_text_sql,
|
|
)
|
|
from ..utils.execution_contexts import (
|
|
CAPTURE_REASONS,
|
|
ContextBundle,
|
|
content_digest,
|
|
validate_bundle,
|
|
)
|
|
except ImportError:
|
|
from utils.execution_history_events import RunFinish, encode_command
|
|
from utils.hashing import (
|
|
generate_prompt_hash,
|
|
stored_prompt_text,
|
|
stored_prompt_text_sql,
|
|
)
|
|
from utils.execution_contexts import (
|
|
CAPTURE_REASONS,
|
|
ContextBundle,
|
|
content_digest,
|
|
validate_bundle,
|
|
)
|
|
|
|
|
|
FIELDS = ("raw", "effective", "prepend", "append")
|
|
ROLES = ("positive", "negative")
|
|
|
|
|
|
def _prepare(bundle: ContextBundle) -> list[dict]:
|
|
validate_bundle(bundle)
|
|
if bundle.reason not in CAPTURE_REASONS:
|
|
raise ValueError("Invalid capture reason")
|
|
prepared = []
|
|
complete = bool(bundle.contexts)
|
|
for context in bundle.contexts:
|
|
evidence = {"format_version": 1, "role_provenance": context.role_provenance}
|
|
snapshots = []
|
|
for role in ROLES:
|
|
snapshot = getattr(context, role)
|
|
if snapshot is None:
|
|
snapshots.append(None)
|
|
evidence[role] = None
|
|
complete = False
|
|
continue
|
|
values = tuple(getattr(snapshot, field).value for field in FIELDS)
|
|
complete = complete and all(value is not None for value in values)
|
|
snapshots.append((content_digest("snapshot", values), values))
|
|
evidence[role] = {
|
|
field: {
|
|
"provenance": getattr(snapshot, field).provenance,
|
|
"source": getattr(snapshot, field).source,
|
|
"reason": getattr(snapshot, field).reason,
|
|
}
|
|
for field in FIELDS
|
|
}
|
|
if all(snapshot is None for snapshot in snapshots):
|
|
raise ValueError("Context must contain a snapshot")
|
|
context_id = content_digest(
|
|
"context", tuple(s[0] if s else None for s in snapshots)
|
|
)
|
|
encoded = json.dumps(
|
|
evidence,
|
|
ensure_ascii=False,
|
|
allow_nan=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
)
|
|
if len(encoded.encode("utf-8")) > 8192:
|
|
raise ValueError("Oversized context evidence")
|
|
prepared.append(
|
|
{
|
|
"anchor_id": context.anchor_id,
|
|
"context_id": context_id,
|
|
"snapshots": snapshots,
|
|
"evidence_json": encoded,
|
|
}
|
|
)
|
|
if bundle.state == "complete" and not complete:
|
|
raise ValueError("Incomplete scalar capture")
|
|
return sorted(prepared, key=lambda item: item["anchor_id"])
|
|
|
|
|
|
def _existing_row(
|
|
conn: sqlite3.Connection, table: str, columns: tuple[str, ...], values: tuple
|
|
) -> bool:
|
|
row = conn.execute(
|
|
f"SELECT {','.join(columns[1:])} FROM {table} WHERE {columns[0]}=?",
|
|
(values[0],),
|
|
).fetchone()
|
|
if row is None:
|
|
return False
|
|
if tuple(row) != values[1:]:
|
|
raise ValueError("Execution context digest collision")
|
|
return True
|
|
|
|
|
|
def _content_rows(item: dict) -> Iterator[tuple[str, tuple[str, ...], tuple]]:
|
|
snapshot_ids = []
|
|
for snapshot in item["snapshots"]:
|
|
if snapshot is None:
|
|
snapshot_ids.append(None)
|
|
continue
|
|
snapshot_id, values = snapshot
|
|
text_ids = []
|
|
for value in values:
|
|
text_id = None if value is None else content_digest("text", (value,))
|
|
if value is not None:
|
|
yield "execution_context_texts", ("text_id", "text_value"), (
|
|
text_id,
|
|
value,
|
|
)
|
|
text_ids.append(text_id)
|
|
yield "execution_prompt_snapshots", (
|
|
"snapshot_id",
|
|
*(f"{field}_text_id" for field in FIELDS),
|
|
), (snapshot_id, *text_ids)
|
|
snapshot_ids.append(snapshot_id)
|
|
yield "execution_contexts", (
|
|
"context_id",
|
|
"positive_snapshot_id",
|
|
"negative_snapshot_id",
|
|
), (item["context_id"], *snapshot_ids)
|
|
|
|
|
|
def _insert_exact(
|
|
conn: sqlite3.Connection, table: str, columns: tuple[str, ...], values: tuple
|
|
) -> None:
|
|
if not _existing_row(conn, table, columns, values):
|
|
names = ",".join(columns)
|
|
placeholders = ",".join("?" for _ in columns)
|
|
conn.execute(f"INSERT INTO {table} ({names}) VALUES ({placeholders})", values)
|
|
|
|
|
|
def _role_hints(role: str) -> str:
|
|
return (
|
|
f"SELECT r.{role}_prompt_id, tx.text_value, "
|
|
f"{stored_prompt_text_sql('tx.text_value')} "
|
|
"FROM execution_run_contexts r "
|
|
"JOIN execution_contexts c ON c.context_id = r.context_id "
|
|
f"JOIN execution_prompt_snapshots s ON s.snapshot_id = c.{role}_snapshot_id "
|
|
"JOIN execution_context_texts tx ON tx.text_id = s.raw_text_id "
|
|
f"WHERE r.{role}_prompt_id IS NOT NULL AND r.reference_epoch = ("
|
|
"SELECT storage_epoch FROM execution_history_metadata WHERE singleton_id = 1)"
|
|
)
|
|
|
|
|
|
# Set form of the read-time prompt-hint rule in get_execution_contexts():
|
|
# a stored role hint is live only in the current storage epoch, and only when
|
|
# the hinted prompt is the one match of _prompt_hint: by text (the raw
|
|
# snapshot text or its stored form, outer whitespace stripped, #438), or, only
|
|
# when no prompt has either text, by dedup hash (case folded too, #460).
|
|
# The hash is a fallback, so a hint #438 kept is never made ambiguous.
|
|
# Needs register_prompt_functions (pm_prompt_hash). The hint rows are the
|
|
# outer loop (CROSS JOIN), so each prompts lookup is by index.
|
|
# Defines the CTE ``live_hint_prompts(prompt_id)``; prefix it with WITH.
|
|
LIVE_HINT_CTES = (
|
|
"stored_hints(prompt_id, raw, stored) AS ("
|
|
+ _role_hints("positive")
|
|
+ " UNION "
|
|
+ _role_hints("negative")
|
|
+ "), hint_matches(prompt_id, raw, stored, match_id) AS ("
|
|
"SELECT h.prompt_id, h.raw, h.stored, p.id FROM stored_hints h "
|
|
"CROSS JOIN prompts p ON p.text COLLATE BINARY IN (h.raw, h.stored) "
|
|
"UNION ALL "
|
|
"SELECT h.prompt_id, h.raw, h.stored, p.id FROM stored_hints h "
|
|
"CROSS JOIN prompts p ON p.hash = pm_prompt_hash(h.raw) "
|
|
"WHERE NOT EXISTS (SELECT 1 FROM prompts q "
|
|
"WHERE q.text COLLATE BINARY IN (h.raw, h.stored))"
|
|
"), live_hint_prompts(prompt_id) AS ("
|
|
"SELECT prompt_id FROM hint_matches GROUP BY prompt_id, raw, stored "
|
|
"HAVING COUNT(*) = 1 AND MIN(match_id) = prompt_id)"
|
|
)
|
|
LIVE_HINT_PROMPT_IDS_SQL = (
|
|
"WITH " + LIVE_HINT_CTES + " SELECT DISTINCT prompt_id FROM live_hint_prompts"
|
|
)
|
|
|
|
_BY_TEXT = "SELECT id FROM prompts WHERE text COLLATE BINARY IN (?,?) LIMIT 2"
|
|
|
|
|
|
def _text_matches(conn: sqlite3.Connection, raw: str) -> list:
|
|
return [row[0] for row in conn.execute(_BY_TEXT, (raw, stored_prompt_text(raw)))]
|
|
|
|
|
|
def _prompt_hint(conn: sqlite3.Connection, raw: str | None) -> int | None:
|
|
"""The one prompt that stores ``raw`` (#438), else the one it counts toward.
|
|
|
|
By text first: ``raw`` or ``raw.strip()`` (the exact form keeps hints of
|
|
older rows that kept outer whitespace). Only when no prompt has either
|
|
text, by the dedup hash, which also folds case (#460).
|
|
"""
|
|
if raw is None:
|
|
return None
|
|
matches = _text_matches(conn, raw)
|
|
if not matches:
|
|
matches = [
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT id FROM prompts WHERE hash=? LIMIT 2",
|
|
(generate_prompt_hash(raw),),
|
|
)
|
|
]
|
|
return matches[0] if len(matches) == 1 else None
|
|
|
|
|
|
def _hints(
|
|
conn: sqlite3.Connection, positive: str | None, negative: str | None
|
|
) -> tuple:
|
|
pos_id, neg_id = _prompt_hint(conn, positive), _prompt_hint(conn, negative)
|
|
stored = None if negative is None else stored_prompt_text(negative)
|
|
pair_id = None
|
|
# A known negative that names no prompt because several share its text
|
|
negative_ambiguous = (
|
|
stored is not None and neg_id is None and bool(_text_matches(conn, negative))
|
|
)
|
|
if pos_id is not None and stored is not None and not negative_ambiguous:
|
|
rows = conn.execute(
|
|
"SELECT pp.id FROM prompt_pairs pp "
|
|
"WHERE pp.positive_prompt_id=? AND ((pp.negative_prompt_id IS NULL AND "
|
|
"?='') OR pp.negative_prompt_id=?) LIMIT 2",
|
|
(pos_id, stored, neg_id),
|
|
).fetchall()
|
|
if len(rows) == 1:
|
|
pair_id = rows[0][0]
|
|
return pos_id, neg_id, pair_id
|
|
|
|
|
|
def _values(conn: sqlite3.Connection, snapshot_id: str | None) -> dict | None:
|
|
if snapshot_id is None:
|
|
return None
|
|
row = conn.execute(
|
|
"SELECT r.text_value,e.text_value,p.text_value,a.text_value "
|
|
"FROM execution_prompt_snapshots s "
|
|
"LEFT JOIN execution_context_texts r ON r.text_id=s.raw_text_id "
|
|
"LEFT JOIN execution_context_texts e ON e.text_id=s.effective_text_id "
|
|
"LEFT JOIN execution_context_texts p ON p.text_id=s.prepend_text_id "
|
|
"LEFT JOIN execution_context_texts a ON a.text_id=s.append_text_id "
|
|
"WHERE s.snapshot_id=?",
|
|
(snapshot_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
raise sqlite3.DatabaseError("Missing execution context snapshot")
|
|
return dict(zip(FIELDS, row))
|
|
|
|
|
|
def _write_capture(
|
|
conn: sqlite3.Connection,
|
|
run_id: str,
|
|
expected_epoch: str,
|
|
bundle: ContextBundle,
|
|
prepared: list[dict],
|
|
) -> bool:
|
|
capture = conn.execute(
|
|
"SELECT state,reason,format_version FROM execution_context_capture "
|
|
"WHERE run_id=?",
|
|
(run_id,),
|
|
).fetchone()
|
|
if capture is not None:
|
|
for item in prepared:
|
|
for table, columns, values in _content_rows(item):
|
|
_existing_row(conn, table, columns, values)
|
|
links = conn.execute(
|
|
"SELECT anchor_id,context_id,evidence_json FROM "
|
|
"execution_run_contexts WHERE run_id=? ORDER BY anchor_id",
|
|
(run_id,),
|
|
).fetchall()
|
|
expected = [
|
|
(item["anchor_id"], item["context_id"], item["evidence_json"])
|
|
for item in prepared
|
|
]
|
|
# Compare JSON objects so serialization order never creates a conflict.
|
|
original = [(row[0], row[1], json.loads(row[2])) for row in links]
|
|
proposed = [
|
|
(anchor, context, json.loads(evidence))
|
|
for anchor, context, evidence in expected
|
|
]
|
|
if capture[0] != "conflicting" and (
|
|
tuple(capture) != (bundle.state, bundle.reason, bundle.format_version)
|
|
or original != proposed
|
|
):
|
|
conn.execute(
|
|
"UPDATE execution_context_capture SET "
|
|
"state='conflicting',reason='conflicting' WHERE run_id=?",
|
|
(run_id,),
|
|
)
|
|
return True
|
|
for item in prepared:
|
|
for table, columns, values in _content_rows(item):
|
|
_insert_exact(conn, table, columns, values)
|
|
raw = [snapshot[1][0] if snapshot else None for snapshot in item["snapshots"]]
|
|
positive_id, negative_id, pair_id = _hints(conn, *raw)
|
|
conn.execute(
|
|
"INSERT INTO "
|
|
"execution_run_contexts(run_id,anchor_id,context_id,evidence_json,"
|
|
"positive_prompt_id,negative_prompt_id,pair_id,reference_epoch,"
|
|
"capture_version) VALUES(?,?,?,?,?,?,?,?,1)",
|
|
(
|
|
run_id,
|
|
item["anchor_id"],
|
|
item["context_id"],
|
|
item["evidence_json"],
|
|
positive_id,
|
|
negative_id,
|
|
pair_id,
|
|
expected_epoch,
|
|
),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO "
|
|
"execution_context_capture(run_id,state,reason,format_version) "
|
|
"VALUES(?,?,?,?)",
|
|
(run_id, bundle.state, bundle.reason, bundle.format_version),
|
|
)
|
|
return True
|
|
|
|
|
|
class ExecutionContextOperations:
|
|
"""Compose with ExecutionHistoryOperations on PromptDatabase."""
|
|
|
|
def write_execution_contexts(
|
|
self, run_id: str, expected_epoch: str, bundle: ContextBundle
|
|
) -> bool:
|
|
prepared = _prepare(bundle)
|
|
with self._execution_transaction(expected_epoch) as conn:
|
|
if (
|
|
conn.execute(
|
|
"SELECT 1 FROM execution_runs WHERE run_id=? AND storage_epoch=?",
|
|
(run_id, expected_epoch),
|
|
).fetchone()
|
|
is None
|
|
):
|
|
return False
|
|
return _write_capture(conn, run_id, expected_epoch, bundle, prepared)
|
|
|
|
def write_execution_contexts_for_finish(
|
|
self,
|
|
command: RunFinish,
|
|
bundle: ContextBundle,
|
|
*,
|
|
generation_guard: Callable[[], ContextManager[bool]] | None = None,
|
|
) -> bool:
|
|
encode_command(command)
|
|
if type(command) is not RunFinish:
|
|
raise ValueError("Expected execution finish command")
|
|
prepared = _prepare(bundle)
|
|
# Enter the guard after BEGIN, and keep it until COMMIT/ROLLBACK exits.
|
|
with ExitStack() as guards:
|
|
with self._execution_transaction(command.storage_epoch) as conn:
|
|
if generation_guard is not None and not guards.enter_context(
|
|
generation_guard()
|
|
):
|
|
return False
|
|
row = conn.execute(
|
|
"SELECT run_id FROM execution_runs WHERE session_id=? "
|
|
"AND queue_item_id=? AND storage_epoch=?",
|
|
(command.session_id, command.queue_item_id, command.storage_epoch),
|
|
).fetchone()
|
|
if row is None:
|
|
return False
|
|
return _write_capture(
|
|
conn, row[0], command.storage_epoch, bundle, prepared
|
|
)
|
|
|
|
def get_execution_contexts(self, run_id: str) -> dict:
|
|
"""Return first-capture scalars; hints are None unless currently valid.
|
|
|
|
reference_epoch remains historical evidence even when hints are invalidated.
|
|
Reads never rebind a first-capture reference or alter stored capture data.
|
|
"""
|
|
conn = self.model.get_connection()
|
|
own_transaction = not conn.in_transaction
|
|
if own_transaction:
|
|
conn.execute("BEGIN")
|
|
try:
|
|
capture = conn.execute(
|
|
"SELECT state,reason,format_version FROM execution_context_capture "
|
|
"WHERE run_id=?",
|
|
(run_id,),
|
|
).fetchone()
|
|
if capture is None:
|
|
return {
|
|
"state": "not_captured",
|
|
"reason": "missing",
|
|
"format_version": None,
|
|
"contexts": [],
|
|
}
|
|
epoch = self.get_execution_epoch()
|
|
contexts = []
|
|
rows = conn.execute(
|
|
"SELECT r.*,c.positive_snapshot_id,c.negative_snapshot_id FROM "
|
|
"execution_run_contexts r JOIN execution_contexts c ON "
|
|
"c.context_id=r.context_id WHERE r.run_id=? ORDER BY r.anchor_id",
|
|
(run_id,),
|
|
)
|
|
for row in rows:
|
|
positive = _values(conn, row["positive_snapshot_id"])
|
|
negative = _values(conn, row["negative_snapshot_id"])
|
|
hints = (None, None, None)
|
|
if row["reference_epoch"] == epoch:
|
|
current = _hints(
|
|
conn,
|
|
positive["raw"] if positive else None,
|
|
negative["raw"] if negative else None,
|
|
)
|
|
hints = tuple(
|
|
row[key] if row[key] == value else None
|
|
for key, value in zip(
|
|
("positive_prompt_id", "negative_prompt_id", "pair_id"),
|
|
current,
|
|
)
|
|
)
|
|
contexts.append(
|
|
{
|
|
"anchor_id": row["anchor_id"],
|
|
"context_id": row["context_id"],
|
|
"positive": positive,
|
|
"negative": negative,
|
|
"evidence": json.loads(row["evidence_json"]),
|
|
"positive_prompt_id": hints[0],
|
|
"negative_prompt_id": hints[1],
|
|
"pair_id": hints[2],
|
|
"reference_epoch": row["reference_epoch"],
|
|
}
|
|
)
|
|
return {
|
|
"state": capture[0],
|
|
"reason": capture[1],
|
|
"format_version": capture[2],
|
|
"contexts": contexts,
|
|
}
|
|
finally:
|
|
if own_transaction:
|
|
conn.rollback()
|