fix(harness): state-integrity pass — reopen order, atomic decision batch, bash anti-bypass
Audit findings 5.1-5.3.
5.1 `phase reopen` now appends `phase_reopened` BEFORE the artifact
teardown: a crash between the two used to leave later-phase artifacts
deleted with the ledger still at the old phase (resume entered a phase
missing its artifacts). The inverse half-state — reopened with stale later
artifacts — is benign. Order locked by tests/test_phase_reopen_order.py.
5.2 New `tht decision add-batch --doc -`: N substantive decisions in ONE
atomic ledger write (meta types and cte_approved stay on `decision add`;
strictest min-phase enforced). reviewer_schema_linking now builds the
complete curation set and persists it with a single add-batch call — a
mid-loop failure can no longer leave the audit ledger half-written, and a
retry cannot duplicate the first K decisions.
5.3 The anti-bypass hook now also blocks BASH mutations of protected
state (`echo >> review_decisions.jsonl`, `sed -i` on the manifest,
`cat > tht-gate.js`, python open('w'), mv/rm/tee/…): FORBIDDEN only
covered tht subcommands and the write/edit hook only covered pi's own
tools. Read-only access (cat/grep/tail/ls) stays allowed.
Also: knownDecisionTypes is defensive — a workflow meta declaring NO
emits at all (older tht, minimal stubs) skips pre-validation instead of
rejecting every substantive type; with emits present, unknown types are
still rejected before the widget (new L1 test).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -25,3 +25,44 @@ test("tht schema introspect senza --refresh passa (cache hit innocuo)", async ()
|
||||
});
|
||||
assert.equal(res, undefined);
|
||||
});
|
||||
|
||||
// Bash mutations of protected state bypass the write/edit hook: block them.
|
||||
const BLOCKED_BASH = [
|
||||
'echo \'{"type":"phase_approved","subject":"phase:4"}\' >> sessions/s1/review_decisions.jsonl',
|
||||
"sed -i '' 's/open/finalized/' sessions/s1/session_manifest.yaml",
|
||||
"cat /tmp/patch.js > .pi/extensions/tht-gate.js",
|
||||
"python3 -c \"open('sessions/s1/review_decisions.jsonl','a').write('x')\"",
|
||||
"mv /tmp/fake.json sessions/s1/cte_plan.json",
|
||||
"rm sessions/s1/session_manifest.yaml",
|
||||
"tee -a sessions/s1/review_decisions.jsonl < /tmp/x",
|
||||
];
|
||||
|
||||
// Read-only access and unrelated redirects stay allowed.
|
||||
const ALLOWED_BASH = [
|
||||
"cat sessions/s1/review_decisions.jsonl",
|
||||
"grep phase_approved sessions/s1/review_decisions.jsonl",
|
||||
"tail -5 sessions/s1/session_manifest.yaml",
|
||||
"ls .pi/extensions",
|
||||
"tht session show s1 > /tmp/out.txt",
|
||||
"echo done > /tmp/scratch.txt",
|
||||
];
|
||||
|
||||
for (const cmd of BLOCKED_BASH) {
|
||||
test(`bash mutation su stato protetto e' bloccata: ${cmd.slice(0, 60)}`, async () => {
|
||||
const installGate = await installGatePromise;
|
||||
const { pi } = createFakePi();
|
||||
installGate(pi);
|
||||
const res = await pi.emit("tool_call", { toolName: "bash", input: { command: cmd } });
|
||||
assert.equal(res?.block, true);
|
||||
});
|
||||
}
|
||||
|
||||
for (const cmd of ALLOWED_BASH) {
|
||||
test(`bash read-only/estraneo passa: ${cmd.slice(0, 60)}`, async () => {
|
||||
const installGate = await installGatePromise;
|
||||
const { pi } = createFakePi();
|
||||
installGate(pi);
|
||||
const res = await pi.emit("tool_call", { toolName: "bash", input: { command: cmd } });
|
||||
assert.equal(res, undefined);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
// validateDecisionTypes: with a full workflow meta (emits declared) an unknown
|
||||
// decision type is rejected BEFORE the widget is shown. (The complementary branch —
|
||||
// a meta with NO emits skips validation entirely — is regression-covered by the
|
||||
// gate_join_review tests, whose minimal meta stub declares no emits.)
|
||||
const test = require("node:test");
|
||||
const assert = require("node:assert");
|
||||
const cp = require("node:child_process");
|
||||
const { createRequire } = require("node:module");
|
||||
const path = require("node:path");
|
||||
|
||||
const GATE = path.join(__dirname, "..", "..", "tht-gate.js");
|
||||
if (typeof globalThis.require === "undefined") {
|
||||
globalThis.require = createRequire(GATE);
|
||||
}
|
||||
|
||||
test("reviewer_select rejects an unknown decision type before showing the widget", async () => {
|
||||
const origExecFileSync = cp.execFileSync;
|
||||
cp.execFileSync = (file, args) => {
|
||||
if (args[0] === "phase" && args[1] === "meta")
|
||||
return JSON.stringify({
|
||||
max_phase: 8,
|
||||
phases: [
|
||||
{ num: 1, id: "F1", emits: ["concept_clarified"] },
|
||||
{ num: 4, id: "F4", emits: ["table_promoted", "join_modified"] },
|
||||
],
|
||||
});
|
||||
if (args[0] === "phase" && args[1] === "show") return "Fase corrente: 4\n";
|
||||
return "";
|
||||
};
|
||||
|
||||
try {
|
||||
const gate = require(GATE);
|
||||
const { createFakePi } = require("./fake_pi_runtime.js");
|
||||
const { pi, ctx, tools } = createFakePi();
|
||||
ctx.cwd = "/nonexistent-thothii-test-cwd";
|
||||
gate.default(pi);
|
||||
|
||||
const uiBefore = ctx.uiCalls.length;
|
||||
const result = await tools.get("reviewer_select").def.execute(
|
||||
"call-1",
|
||||
{
|
||||
session: "s1",
|
||||
title: "t",
|
||||
options: [
|
||||
{ id: "a", label: "A", decision: { type: "tipo_inventato", subject: "x" } },
|
||||
],
|
||||
},
|
||||
null,
|
||||
null,
|
||||
ctx,
|
||||
);
|
||||
|
||||
assert.match(result.content[0].text, /tipo_inventato.*non valido/i);
|
||||
assert.equal(ctx.uiCalls.length, uiBefore, "the widget must NOT be shown");
|
||||
} finally {
|
||||
cp.execFileSync = origExecFileSync;
|
||||
}
|
||||
});
|
||||
@@ -35,16 +35,18 @@ const CATALOG = {
|
||||
|
||||
test("reviewer_schema_linking records table/column decisions and syncs schema_linking", async () => {
|
||||
const calls = [];
|
||||
const inputs = [];
|
||||
// Adaptation #1: stub the shell BEFORE requiring tht-gate.js.
|
||||
const origExecFileSync = cp.execFileSync;
|
||||
cp.execFileSync = (file, args) => {
|
||||
cp.execFileSync = (file, args, opts) => {
|
||||
calls.push(args.join(" "));
|
||||
inputs.push(opts?.input);
|
||||
if (args[0] === "phase" && args[1] === "meta")
|
||||
return JSON.stringify({ phases: [{ num: 4, id: "F4" }] });
|
||||
if (args[0] === "phase" && args[1] === "show") return "Fase corrente: 4\n";
|
||||
if (args[0] === "schema" && args[1] === "columns" && args[2] === "dim_patient")
|
||||
return JSON.stringify(CATALOG);
|
||||
return ""; // decision add, session sync-schema-linking, ...
|
||||
return ""; // decision add-batch, session sync-schema-linking, ...
|
||||
};
|
||||
|
||||
try {
|
||||
@@ -93,27 +95,27 @@ test("reviewer_schema_linking records table/column decisions and syncs schema_li
|
||||
assert.equal(capturedDescriptor.widget, "schema-linking");
|
||||
assert.equal(capturedDescriptor.tables[0].columns.length, 2);
|
||||
|
||||
// cod_paz was selected -> promoted; nome was suggested but deselected -> excluded.
|
||||
assert.ok(
|
||||
calls.some((c) => /decision add .*--type table_promoted --subject dim_patient\b/.test(c)),
|
||||
`expected table_promoted dim_patient in: ${JSON.stringify(calls)}`,
|
||||
);
|
||||
assert.ok(
|
||||
calls.some((c) => /decision add .*--type column_promoted --subject dim_patient\.cod_paz\b/.test(c)),
|
||||
`expected column_promoted dim_patient.cod_paz in: ${JSON.stringify(calls)}`,
|
||||
);
|
||||
assert.ok(
|
||||
calls.some((c) => /decision add .*--type column_excluded --subject dim_patient\.nome\b/.test(c)),
|
||||
`expected column_excluded dim_patient.nome in: ${JSON.stringify(calls)}`,
|
||||
// The complete curation is persisted with ONE atomic ledger write (add-batch):
|
||||
// cod_paz selected -> promoted; nome suggested but deselected -> excluded.
|
||||
const batchIdx = calls.findIndex((c) => c === "decision add-batch --session s1 --doc -");
|
||||
assert.ok(batchIdx !== -1, `expected one decision add-batch in: ${JSON.stringify(calls)}`);
|
||||
const batch = JSON.parse(inputs[batchIdx]);
|
||||
assert.deepEqual(
|
||||
batch.map(({ type, subject }) => ({ type, subject })),
|
||||
[
|
||||
{ type: "table_promoted", subject: "dim_patient" },
|
||||
{ type: "column_promoted", subject: "dim_patient.cod_paz" },
|
||||
{ type: "column_excluded", subject: "dim_patient.nome" },
|
||||
],
|
||||
);
|
||||
assert.equal(calls.filter((c) => c.startsWith("decision add")).length, 1);
|
||||
assert.ok(
|
||||
calls.some((c) => /^session sync-schema-linking s1$/.test(c)),
|
||||
`expected session sync-schema-linking s1 in: ${JSON.stringify(calls)}`,
|
||||
);
|
||||
// sync runs AFTER the decisions are recorded.
|
||||
// sync runs AFTER the atomic batch.
|
||||
const syncIdx = calls.findIndex((c) => c.startsWith("session sync-schema-linking"));
|
||||
const lastDecisionIdx = calls.map((c) => c.startsWith("decision add")).lastIndexOf(true);
|
||||
assert.ok(syncIdx > lastDecisionIdx, "sync must run after all decision adds");
|
||||
assert.ok(syncIdx > batchIdx, "sync must run after the decision batch");
|
||||
|
||||
assert.match(result.content[0].text, /Schema linking registrato/);
|
||||
} finally {
|
||||
|
||||
@@ -153,6 +153,17 @@ const PROTECTED_FILES =
|
||||
// itself. pi's write/edit tools are otherwise unrestricted, so a confused model can
|
||||
// (and did) patch tht-gate.js mid-loop. Block any write under the extensions dir.
|
||||
export const GATE_CODE_FILES = /\.pi[\\/]extensions[\\/]/;
|
||||
// Bash can mutate protected state around the write/edit hook (`echo >> ledger`,
|
||||
// `sed -i` on the manifest, `cat > tht-gate.js`). Block any bash command that names
|
||||
// a protected path in a mutating context; read-only mentions (cat/grep/ls) stay
|
||||
// allowed. Defense against a CONFUSED model, not a sandbox.
|
||||
const PROTECTED_PATH_TOKEN =
|
||||
/(review_decisions\.jsonl|session_manifest\.yaml|cte_plan\.json|\.pi[\\/]extensions)/;
|
||||
const BASH_MUTATION =
|
||||
/(>>?|\btee\b|\bsed\b[^|;&]*\s-i\b|\bmv\b|\bcp\b|\brm\b|\btruncate\b|\bdd\b|open\([^)]*['"][wa]\+?b?['"]|\bchmod\b|\bln\b)/;
|
||||
export function isProtectedBashMutation(cmd) {
|
||||
return PROTECTED_PATH_TOKEN.test(cmd) && BASH_MUTATION.test(cmd);
|
||||
}
|
||||
|
||||
// --- kickoff payloads (verbatim from source L184-212, load-bearing model prose) -
|
||||
const NUOVA_DOMANDA_KICKOFF =
|
||||
@@ -287,15 +298,20 @@ let _knownTypes = null;
|
||||
function knownDecisionTypes(ctx) {
|
||||
if (_knownTypes) return _knownTypes;
|
||||
const meta = phaseMeta(ctx);
|
||||
const types = new Set([
|
||||
const emitted = new Set();
|
||||
for (const p of meta.phases) {
|
||||
for (const t of (p.emits || [])) emitted.add(t);
|
||||
}
|
||||
// A workflow meta with NO emits (older tht, minimal test stubs) gives the gate no
|
||||
// vocabulary to validate against: skip validation instead of rejecting every
|
||||
// substantive type and bricking the gates.
|
||||
if (emitted.size === 0) return null;
|
||||
for (const t of [
|
||||
"phase_approved", "phase_auto_approved", "phase_reopened",
|
||||
"phase_skipped", "decision_retracted",
|
||||
]);
|
||||
for (const p of meta.phases) {
|
||||
for (const t of (p.emits || [])) types.add(t);
|
||||
}
|
||||
_knownTypes = types;
|
||||
return types;
|
||||
]) emitted.add(t);
|
||||
_knownTypes = emitted;
|
||||
return emitted;
|
||||
}
|
||||
function decisionMinPhaseMap(ctx) {
|
||||
const meta = phaseMeta(ctx);
|
||||
@@ -309,6 +325,7 @@ function decisionMinPhaseMap(ctx) {
|
||||
}
|
||||
function validateDecisionTypes(ctx, options, session) {
|
||||
const known = knownDecisionTypes(ctx);
|
||||
if (!known) return null;
|
||||
for (const o of options) {
|
||||
if (o.decision && !known.has(o.decision.type)) {
|
||||
return `Tipo di decisione '${o.decision.type}' non valido. Tipi ammessi: ${[...known].join(", ")}. Correggi e riprova.`;
|
||||
@@ -565,6 +582,15 @@ export default function (pi) {
|
||||
"reviewer: usa i tool reviewer_confirm / reviewer_select.",
|
||||
};
|
||||
}
|
||||
if (isProtectedBashMutation(cmd)) {
|
||||
return {
|
||||
block: true,
|
||||
reason:
|
||||
"I file di stato della sessione e il codice del gate non si modificano " +
|
||||
"da shell: lo stato passa SOLO dai tool del gate (reviewer_*/write_*). " +
|
||||
"La lettura (cat/grep) resta permessa.",
|
||||
};
|
||||
}
|
||||
}
|
||||
if (event.toolName === "write" || event.toolName === "edit") {
|
||||
const path = event.input?.path ?? event.input?.file_path ?? "";
|
||||
@@ -984,16 +1010,20 @@ export default function (pi) {
|
||||
if (resp.control === "freetext")
|
||||
return textResult(`Altro (reviewer): ${resp.text}. Riformula tenendone conto.`);
|
||||
|
||||
// Build the COMPLETE decision set first, persist it with ONE atomic ledger
|
||||
// write (decision add-batch): a per-item loop could fail halfway and leave
|
||||
// the audit ledger half-written, with duplicates on retry.
|
||||
const byId = new Map(enriched.map((t) => [t.id, t]));
|
||||
const toPersist = [];
|
||||
let n = 0;
|
||||
for (const rt of resp.tables ?? []) {
|
||||
const t = byId.get(rt.id);
|
||||
if (!t || !rt.enacted) continue;
|
||||
if (t.kind === "promote") {
|
||||
const e1 = relayIfThtFails(ctx, decisionAddArgs(session, {
|
||||
type: "table_promoted", subject: t.name, detail: t.description, rationale: t.rationale,
|
||||
}), "");
|
||||
if (e1) return e1;
|
||||
toPersist.push({
|
||||
type: "table_promoted", subject: t.name,
|
||||
detail: t.description, rationale: t.rationale,
|
||||
});
|
||||
n++;
|
||||
const sel = new Set(rt.columns ?? []);
|
||||
for (const c of t.columns) {
|
||||
@@ -1001,19 +1031,27 @@ export default function (pi) {
|
||||
? "column_promoted"
|
||||
: (c.suggested ? "column_excluded" : null);
|
||||
if (!type) continue;
|
||||
const e2 = relayIfThtFails(ctx, decisionAddArgs(session, {
|
||||
toPersist.push({
|
||||
type, subject: `${t.name}.${c.name}`, detail: c.description ?? "",
|
||||
}), "");
|
||||
if (e2) return e2;
|
||||
});
|
||||
}
|
||||
} else {
|
||||
const e3 = relayIfThtFails(ctx, decisionAddArgs(session, {
|
||||
type: "table_excluded", subject: t.name, detail: t.description, rationale: t.rationale,
|
||||
}), "");
|
||||
if (e3) return e3;
|
||||
toPersist.push({
|
||||
type: "table_excluded", subject: t.name,
|
||||
detail: t.description, rationale: t.rationale,
|
||||
});
|
||||
n++;
|
||||
}
|
||||
}
|
||||
if (toPersist.length) {
|
||||
const eBatch = relayIfThtFails(
|
||||
ctx,
|
||||
["decision", "add-batch", "--session", session, "--doc", "-"],
|
||||
"Nessuna decisione registrata (batch atomico fallito): correggi e ripresenta il gate.",
|
||||
JSON.stringify(toPersist),
|
||||
);
|
||||
if (eBatch) return eBatch;
|
||||
}
|
||||
|
||||
// Deterministic projection of the ledger into schema_linking.json.
|
||||
const eSync = relayIfThtFails(ctx, ["session", "sync-schema-linking", session], "");
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
"""`decision add-batch`: N decisioni sostanziali in un'unica scrittura atomica.
|
||||
|
||||
Contratto: o TUTTE le decisioni entrano nel ledger o NESSUNA — un item invalido
|
||||
rigetta l'intero batch (il retry del gate non può creare duplicati parziali).
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from tht.cli.decision_cmd import decision_app
|
||||
from tht.decisions import append_decision, append_decisions, list_decisions
|
||||
from tht.phase import current_phase
|
||||
|
||||
|
||||
def _walk_to_phase(session, target):
|
||||
while current_phase(session) < target:
|
||||
append_decision(session, type="phase_approved", subject=f"phase:{current_phase(session)}")
|
||||
|
||||
|
||||
def _configure_command(monkeypatch, sessions):
|
||||
import tht.cli.decision_cmd as mod
|
||||
import tht.cli.session_cmd as session_mod
|
||||
from tht.session.models import SessionManifest, SessionSnapshot
|
||||
|
||||
class _Repository:
|
||||
def get(self, session_id):
|
||||
return SessionSnapshot(
|
||||
manifest=SessionManifest(
|
||||
id=session_id, created_at="2026-01-01T00:00:00Z",
|
||||
question="q", database="d", schema="s",
|
||||
),
|
||||
decisions=list_decisions(sessions / session_id),
|
||||
)
|
||||
|
||||
def append_decisions(self, session_id, decisions):
|
||||
return append_decisions(sessions / session_id, list(decisions))
|
||||
|
||||
class _Cfg:
|
||||
pass
|
||||
|
||||
repository = _Repository()
|
||||
monkeypatch.setattr(mod, "_load_config_or_exit", lambda _c: _Cfg())
|
||||
monkeypatch.setattr(mod, "load_session_or_exit", lambda _cfg, _s: None)
|
||||
monkeypatch.setattr(mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
|
||||
monkeypatch.setattr(mod, "session_repository", lambda _cfg: repository)
|
||||
monkeypatch.setattr(session_mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
|
||||
|
||||
|
||||
SCHEMA_LINKING_BATCH = [
|
||||
{"type": "table_promoted", "subject": "fact_impianto_pmk", "detail": "impianti PMK"},
|
||||
{"type": "column_promoted", "subject": "fact_impianto_pmk.cod_paz"},
|
||||
{"type": "column_excluded", "subject": "fact_impianto_pmk.note"},
|
||||
{"type": "table_excluded", "subject": "dim_obsoleta", "rationale": "non pertinente"},
|
||||
]
|
||||
|
||||
|
||||
def test_add_batch_appends_all_decisions_in_one_write(tmp_path, monkeypatch):
|
||||
session = tmp_path / "s1"
|
||||
session.mkdir()
|
||||
_walk_to_phase(session, 4)
|
||||
_configure_command(monkeypatch, tmp_path)
|
||||
|
||||
result = CliRunner().invoke(
|
||||
decision_app,
|
||||
["add-batch", "--session", "s1", "--doc", "-"],
|
||||
input=json.dumps(SCHEMA_LINKING_BATCH),
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
recorded = [d for d in list_decisions(session) if not d.type.startswith("phase_")]
|
||||
assert [(d.type, d.subject) for d in recorded] == [
|
||||
(item["type"], item["subject"]) for item in SCHEMA_LINKING_BATCH
|
||||
]
|
||||
|
||||
|
||||
def test_add_batch_rejects_the_whole_batch_when_one_item_is_invalid(tmp_path, monkeypatch):
|
||||
session = tmp_path / "s1"
|
||||
session.mkdir()
|
||||
_walk_to_phase(session, 4)
|
||||
_configure_command(monkeypatch, tmp_path)
|
||||
before = len(list_decisions(session))
|
||||
|
||||
bad = SCHEMA_LINKING_BATCH + [{"type": "tipo_inesistente", "subject": "x"}]
|
||||
result = CliRunner().invoke(
|
||||
decision_app,
|
||||
["add-batch", "--session", "s1", "--doc", "-"],
|
||||
input=json.dumps(bad),
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert len(list_decisions(session)) == before
|
||||
|
||||
|
||||
def test_add_batch_rejects_meta_and_cte_approved_types(tmp_path, monkeypatch):
|
||||
session = tmp_path / "s1"
|
||||
session.mkdir()
|
||||
_walk_to_phase(session, 4)
|
||||
_configure_command(monkeypatch, tmp_path)
|
||||
before = len(list_decisions(session))
|
||||
|
||||
for forbidden in ("phase_approved", "decision_retracted", "cte_approved"):
|
||||
result = CliRunner().invoke(
|
||||
decision_app,
|
||||
["add-batch", "--session", "s1", "--doc", "-"],
|
||||
input=json.dumps([{"type": forbidden, "subject": "x"}]),
|
||||
)
|
||||
assert result.exit_code == 1, forbidden
|
||||
assert len(list_decisions(session)) == before
|
||||
|
||||
|
||||
def test_add_batch_enforces_the_strictest_min_phase(tmp_path, monkeypatch):
|
||||
session = tmp_path / "s1"
|
||||
session.mkdir()
|
||||
_walk_to_phase(session, 2) # column_* richiede la fase di schema linking (4)
|
||||
_configure_command(monkeypatch, tmp_path)
|
||||
before = len(list_decisions(session))
|
||||
|
||||
result = CliRunner().invoke(
|
||||
decision_app,
|
||||
["add-batch", "--session", "s1", "--doc", "-"],
|
||||
input=json.dumps([{"type": "column_promoted", "subject": "t.c"}]),
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert len(list_decisions(session)) == before
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Il reopen scrive il ledger PRIMA del teardown (contratto crash-safety).
|
||||
|
||||
Un crash tra le due mutazioni deve lasciare lo stato benigno: `phase_reopened`
|
||||
registrato con artefatti delle fasi successive ancora presenti (la fase foldata
|
||||
vince e il workflow li sovrascrive), MAI artefatti cancellati con il ledger
|
||||
fermo alla fase vecchia (resume entrerebbe in una fase senza i suoi artefatti).
|
||||
"""
|
||||
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from tht.cli.phase_cmd import phase_app
|
||||
from tht.decisions import append_decision, list_decisions
|
||||
from tht.phase import current_phase
|
||||
|
||||
|
||||
def _walk_to_phase(session, target):
|
||||
while current_phase(session) < target:
|
||||
append_decision(session, type="phase_approved", subject=f"phase:{current_phase(session)}")
|
||||
|
||||
|
||||
def _configure(monkeypatch, sessions):
|
||||
import tht.cli.phase_cmd as mod
|
||||
import tht.cli.session_cmd as session_mod
|
||||
from tht.decisions import append_decisions
|
||||
from tht.session.models import SessionManifest, SessionSnapshot
|
||||
|
||||
class _Repository:
|
||||
def get(self, session_id):
|
||||
return SessionSnapshot(
|
||||
manifest=SessionManifest(
|
||||
id=session_id, created_at="2026-01-01T00:00:00Z",
|
||||
question="q", database="d", schema="s",
|
||||
),
|
||||
decisions=list_decisions(sessions / session_id),
|
||||
)
|
||||
|
||||
def append_decisions(self, session_id, decisions):
|
||||
return append_decisions(sessions / session_id, list(decisions))
|
||||
|
||||
repository = _Repository()
|
||||
monkeypatch.setattr(mod, "_cfg", lambda: object())
|
||||
monkeypatch.setattr(session_mod, "load_snapshot_or_exit", lambda _cfg, sid: repository.get(sid))
|
||||
monkeypatch.setattr(session_mod, "session_repository", lambda _cfg: repository)
|
||||
return repository
|
||||
|
||||
|
||||
def _reopened_subjects(session):
|
||||
return [d.subject for d in list_decisions(session) if d.type == "phase_reopened"]
|
||||
|
||||
|
||||
def test_crash_in_teardown_still_records_phase_reopened(tmp_path, monkeypatch):
|
||||
session = tmp_path / "s1"
|
||||
session.mkdir()
|
||||
_walk_to_phase(session, 4)
|
||||
_configure(monkeypatch, tmp_path)
|
||||
|
||||
import tht.teardown as teardown_mod
|
||||
|
||||
def _boom(_repository, _snapshot, _phase):
|
||||
raise RuntimeError("crash window: teardown died after the ledger append")
|
||||
|
||||
monkeypatch.setattr(teardown_mod, "teardown_snapshot", _boom)
|
||||
|
||||
result = CliRunner().invoke(phase_app, ["reopen", "--session", "s1", "--phase", "2"])
|
||||
|
||||
assert result.exit_code != 0
|
||||
# The ledger mutation happened BEFORE the crash: the session is back at phase 2
|
||||
# (with stale later artifacts, which is the benign half-state).
|
||||
assert _reopened_subjects(session) == ["phase:2"]
|
||||
assert current_phase(session) == 2
|
||||
|
||||
|
||||
def test_successful_reopen_records_ledger_and_runs_teardown(tmp_path, monkeypatch):
|
||||
session = tmp_path / "s1"
|
||||
session.mkdir()
|
||||
_walk_to_phase(session, 4)
|
||||
_configure(monkeypatch, tmp_path)
|
||||
|
||||
import tht.teardown as teardown_mod
|
||||
|
||||
calls = []
|
||||
|
||||
class _Report:
|
||||
deleted_files = []
|
||||
|
||||
def _spy(_repository, snapshot, phase):
|
||||
# By the time teardown runs, the ledger already holds the reopen decision.
|
||||
calls.append((phase, _reopened_subjects(session)))
|
||||
return _Report()
|
||||
|
||||
monkeypatch.setattr(teardown_mod, "teardown_snapshot", _spy)
|
||||
|
||||
result = CliRunner().invoke(phase_app, ["reopen", "--session", "s1", "--phase", "2"])
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert calls == [(2, ["phase:2"])]
|
||||
assert current_phase(session) == 2
|
||||
@@ -47,6 +47,57 @@ def add_join_set_cmd(
|
||||
)
|
||||
|
||||
|
||||
@decision_app.command("add-batch")
|
||||
def add_batch_cmd(
|
||||
session: str = typer.Option(..., "--session", help="Id della sessione."),
|
||||
doc: str = typer.Option(..., "--doc", help="Array JSON di decisioni; usa '-' per stdin."),
|
||||
config: Path = CONFIG_OPT,
|
||||
) -> None:
|
||||
"""Registra N decisioni sostanziali con un'unica scrittura atomica del ledger.
|
||||
|
||||
Per i gate che persistono una curation completa (es. schema linking:
|
||||
table_* + column_*): un fallimento a metà non lascia mai il ledger
|
||||
mezzo-scritto — o tutte o nessuna. I tipi meta (phase_*,
|
||||
decision_retracted) e cte_approved restano su `decision add`."""
|
||||
from tht.decisions import DecisionInput
|
||||
|
||||
cfg = _load_config_or_exit(config)
|
||||
load_session_or_exit(cfg, session)
|
||||
excluded = {
|
||||
"phase_approved", "phase_auto_approved", "phase_reopened",
|
||||
"phase_skipped", "decision_retracted", "cte_approved",
|
||||
}
|
||||
try:
|
||||
raw = sys.stdin.read() if doc == "-" else Path(doc).read_text()
|
||||
payload = json.loads(raw)
|
||||
if not isinstance(payload, list) or not payload:
|
||||
raise ValueError("il documento deve essere un array JSON non vuoto")
|
||||
decisions = [DecisionInput.model_validate(item) for item in payload]
|
||||
for decision in decisions:
|
||||
if decision.type in excluded:
|
||||
raise ValueError(
|
||||
f"tipo '{decision.type}' non ammesso in batch (usa 'decision add')"
|
||||
)
|
||||
except (OSError, json.JSONDecodeError, ValidationError, ValueError) as error:
|
||||
typer.secho(
|
||||
f"ERRORE: batch di decisioni non valido: {error}", fg=typer.colors.RED, err=True,
|
||||
)
|
||||
raise typer.Exit(code=1) from error
|
||||
|
||||
from tht.cli.phase_cmd import require_phase_or_exit
|
||||
from tht.workflow import load_workflow
|
||||
|
||||
workflow = load_workflow()
|
||||
require_phase_or_exit(
|
||||
cfg, session, max(workflow.decision_min_phase(d.type) for d in decisions)
|
||||
)
|
||||
records = session_repository(cfg).append_decisions(session, decisions)
|
||||
typer.secho(
|
||||
f"OK: registrate {len(records)} decisioni in un'unica scrittura atomica.",
|
||||
fg=typer.colors.GREEN,
|
||||
)
|
||||
|
||||
|
||||
@decision_app.command("add")
|
||||
def add_cmd(
|
||||
session: str = typer.Option(..., "--session", help="Id della sessione."),
|
||||
|
||||
@@ -122,10 +122,15 @@ def reopen_cmd(
|
||||
from tht.teardown import teardown_snapshot
|
||||
|
||||
repository = session_repository(cfg)
|
||||
report = teardown_snapshot(repository, snapshot, phase)
|
||||
# Ledger FIRST, teardown after: a crash between the two used to leave later-phase
|
||||
# artifacts deleted with the ledger still at the old phase (resume entered a phase
|
||||
# whose expected artifacts were gone). The inverse half-state — reopened with stale
|
||||
# later artifacts — is benign: the folded phase wins and the workflow overwrites
|
||||
# them as it re-progresses.
|
||||
repository.append_decisions(
|
||||
session, [{"type": "phase_reopened", "subject": f"phase:{phase}"}]
|
||||
)
|
||||
report = teardown_snapshot(repository, snapshot, phase)
|
||||
for f in report.deleted_files:
|
||||
typer.echo(f" eliminato artefatto: {f}")
|
||||
typer.echo(f"Tornati alla Fase {phase} ({load_workflow().phase_name(phase)}).")
|
||||
|
||||
Reference in New Issue
Block a user