#!/usr/bin/env python3 """Isolated evidence-near Negative Act Form classification experiment.""" from __future__ import annotations import argparse import json import time from pathlib import Path from typing import Any from .experiment_h import ( DEFAULT_ENDPOINT, DEFAULT_MODEL, DerivationValidationError, OBSERVATION_KEYS, build_ollama_payload, call_ollama, ) GOLD_SCHEMA_VERSION = "experimental-negative-act-form-gold-v0" RECOGNITION_KEYS = {"observation_id", "negative_act_form", "normalized_action_text"} NEGATIVE_ACT_FORMS = { "explicit_non_pursuit", "personal_preference", "recommendation", "temporary_non_action", "none", } FORBIDDEN_LLM_KEYS = { "rejection_form", "explicitly_rejected", "status", "decision", "outcome", "topic_status", "responsible_person", "responsibility", "owner", "requested_actor", "action_item", "protocol", "protocol_category", "confidence", "relation", "relations", "graph", "unresolved_issue", } PROMPT_TEMPLATE = """Classify only the negative semantic form expressed by the candidate observation, using earlier supplied V3-style observations only as local context for pronouns or shortened references. The candidate observation is {candidate_observation_id}. Choose exactly one negative_act_form: - explicit_non_pursuit: explicitly states that an action, option, collaboration, or course will not be continued or pursued. This is stronger than preference, advice, or temporary delay. - personal_preference: the speaker states what they personally would or would not do, without establishing collective non-pursuit. - recommendation: the speaker advises for or against an action without establishing abandonment. - temporary_non_action: the action is postponed, deferred, or explicitly not done for now without abandonment. - none: none of those four forms is present, including mere concern, uncertainty, negative sentiment, or factual negation. Do not collapse non-pursuit into temporary non-action. Do not convert a personal conditional preference into collective non-pursuit. Do not convert advice into non-pursuit. Speaker identity does not change personal preference into collective non-pursuit. When the form is not none, return concise normalized action meaning. Resolve a pronoun only from the supplied local context. If its target is genuinely ambiguous, return none rather than guessing. When the form is none, normalized_action_text must be null. Keep normalized action text in the observation language. Do not derive or output rejection, status, decision, outcome, topic closure, responsibility, ownership, Action Item, protocol category, confidence, relations, graphs, or unresolved issues. Return exactly this JSON shape and no additional fields: {{ "observation_id": "{candidate_observation_id}", "negative_act_form": "explicit_non_pursuit | personal_preference | recommendation | temporary_non_action | none", "normalized_action_text": "concise action meaning" | null }} V3-style observations: {observations_json} """ def _exact_keys(value: dict[str, Any], required: set[str], location: str) -> None: missing = required - value.keys() unknown = value.keys() - required if missing: raise DerivationValidationError(f"{location} missing required keys: {sorted(missing)}") if unknown: raise DerivationValidationError(f"{location} has unknown keys: {sorted(unknown)}") def _nonempty_text(value: Any, location: str) -> str: if not isinstance(value, str) or not value.strip(): raise DerivationValidationError(f"{location} must be a non-empty string") return value.strip() def _validate_observations(observations: Any) -> None: if not isinstance(observations, list) or not observations: raise DerivationValidationError("observations must be a non-empty list") seen_observations: set[str] = set() seen_evidence: set[str] = set() for index, observation in enumerate(observations): location = f"observations[{index}]" if not isinstance(observation, dict): raise DerivationValidationError(f"{location} must be an object") _exact_keys(observation, OBSERVATION_KEYS, location) observation_id = _nonempty_text(observation["observation_id"], f"{location}.observation_id") evidence_id = _nonempty_text(observation["evidence_id"], f"{location}.evidence_id") if observation_id in seen_observations or evidence_id in seen_evidence: raise DerivationValidationError("observation and evidence provenance must be unique") seen_observations.add(observation_id) seen_evidence.add(evidence_id) _nonempty_text(observation["content"], f"{location}.content") _nonempty_text(observation["speaker"], f"{location}.speaker") for field in ("named_person", "addressee"): if observation[field] is not None: _nonempty_text(observation[field], f"{location}.{field}") def load_gold_cases(path: Path) -> list[dict[str, Any]]: data = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(data, dict): raise DerivationValidationError("Gold fixture must be an object") _exact_keys(data, {"schema_version", "cases"}, "Gold fixture") if data["schema_version"] != GOLD_SCHEMA_VERSION: raise DerivationValidationError("unexpected Gold fixture schema_version") cases = data["cases"] if not isinstance(cases, list) or not cases: raise DerivationValidationError("Gold fixture cases must be a non-empty list") seen: set[str] = set() for case in cases: _exact_keys(case, {"case_id", "description", "observations", "expected"}, "Gold case") case_id = _nonempty_text(case["case_id"], "Gold case.case_id") if case_id in seen: raise DerivationValidationError(f"duplicate case ID: {case_id}") seen.add(case_id) _validate_observations(case["observations"]) if len(case["observations"]) not in (1, 2): raise DerivationValidationError("Negative Act cases require one or two observations") return cases def build_prompt(case: dict[str, Any]) -> str: observations = case["observations"] _validate_observations(observations) candidate_id = observations[-1]["observation_id"] return PROMPT_TEMPLATE.format( candidate_observation_id=candidate_id, observations_json=json.dumps(observations, ensure_ascii=False, indent=2), ) def parse_model_json(raw_text: str) -> dict[str, Any]: data = json.loads(raw_text) if not isinstance(data, dict): raise DerivationValidationError("semantic classification must be an object") return data def _reject_forbidden_keys(value: Any, location: str = "output") -> None: if isinstance(value, dict): forbidden = FORBIDDEN_LLM_KEYS.intersection(value) if forbidden: raise DerivationValidationError(f"{location} contains forbidden semantic keys: {sorted(forbidden)}") for key, item in value.items(): _reject_forbidden_keys(item, f"{location}.{key}") elif isinstance(value, list): for index, item in enumerate(value): _reject_forbidden_keys(item, f"{location}[{index}]") def validate_classification(data: Any, observations: list[dict[str, Any]]) -> dict[str, Any]: _validate_observations(observations) if not isinstance(data, dict): raise DerivationValidationError("semantic classification must be an object") _reject_forbidden_keys(data) _exact_keys(data, RECOGNITION_KEYS, "output") observation_id = _nonempty_text(data["observation_id"], "output.observation_id") if observation_id not in {item["observation_id"] for item in observations}: raise DerivationValidationError("classification references unknown observation") form = data["negative_act_form"] if form not in NEGATIVE_ACT_FORMS: raise DerivationValidationError("negative_act_form has an unsupported value") action_text = data["normalized_action_text"] if form == "none": if action_text is not None: raise DerivationValidationError("none form requires null normalized_action_text") else: _nonempty_text(action_text, "output.normalized_action_text") return data def _concepts_present(text: str | None, concepts: list[list[str]]) -> bool: if not concepts: return text is None if not isinstance(text, str): return False folded = text.casefold() return all(any(alias.casefold() in folded for alias in alternatives) for alternatives in concepts) def evaluate_case(case: dict[str, Any], classification: dict[str, Any]) -> dict[str, Any]: validate_classification(classification, case["observations"]) expected = case["expected"] observation_correct = classification["observation_id"] == expected["observation_id"] form_correct = classification["negative_act_form"] == expected["negative_act_form"] action_correct = _concepts_present(classification["normalized_action_text"], expected["action_concepts"]) unsupported_strengthening = expected["negative_act_form"] == "none" and classification["negative_act_form"] != "none" classification_label = "PASS" if observation_correct and form_correct and action_correct else ("PARTIAL" if observation_correct and form_correct else "FAIL") return { "case_id": case["case_id"], "classification": classification_label, "expected_negative_act_form": expected["negative_act_form"], "actual_negative_act_form": classification["negative_act_form"], "observation_id_correct": observation_correct, "normalized_action_meaning_correct": action_correct, "unsupported_semantic_strengthening": unsupported_strengthening, "normative_leakage": False, } def _write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def run_experiment(args: argparse.Namespace) -> dict[str, Any]: cases = load_gold_cases(args.cases) args.output.mkdir(parents=True, exist_ok=False) _write_json(args.output / "gold_cases.json", {"schema_version": GOLD_SCHEMA_VERSION, "cases": cases}) evaluations: list[dict[str, Any]] = [] successful_calls = 0 technical_failures = 0 started = time.perf_counter() for case in cases: case_dir = args.output / case["case_id"].lower() case_dir.mkdir() observations = case["observations"] _write_json(case_dir / "v3_style_input_observations.json", observations) prompt = build_prompt(case) (case_dir / "prompt.txt").write_text(prompt, encoding="utf-8") try: raw, metadata = call_ollama(args.endpoint, args.model, prompt, args.timeout, args.num_ctx, args.num_predict) successful_calls += 1 except Exception as exc: # one recorded attempt; never retry technical_failures += 1 failure = {"case_id": case["case_id"], "classification": "FAIL", "technical_failure": True, "error_type": type(exc).__name__, "error": str(exc)} _write_json(case_dir / "ollama_metadata.json", {"model": args.model, "configuration": {"temperature": 0, "think": False, "num_ctx": args.num_ctx, "num_predict": args.num_predict, "retries": 0}, "technical_failure": failure}) _write_json(case_dir / "structural_validation.json", {"valid": False, "error": str(exc)}) _write_json(case_dir / "evaluation.json", failure) evaluations.append(failure) continue (case_dir / "raw_model_response.txt").write_text(raw + "\n", encoding="utf-8") _write_json(case_dir / "ollama_metadata.json", metadata) try: parsed = parse_model_json(raw) _write_json(case_dir / "parsed_semantic_classification.json", parsed) evaluation = evaluate_case(case, parsed) validation = {"valid": True, "error": None} except (DerivationValidationError, json.JSONDecodeError) as exc: validation = {"valid": False, "error_type": type(exc).__name__, "error": str(exc)} evaluation = {"case_id": case["case_id"], "classification": "FAIL", "error": str(exc), "normative_leakage": "forbidden" in str(exc)} _write_json(case_dir / "structural_validation.json", validation) _write_json(case_dir / "evaluation.json", evaluation) evaluations.append(evaluation) summary = { "experiment": "negative_act_form_v0", "model": args.model, "successful_llm_call_count": successful_calls, "technical_failed_call_count": technical_failures, "runtime_seconds": round(time.perf_counter() - started, 3), "counts": {label: sum(item["classification"] == label for item in evaluations) for label in ("PASS", "PARTIAL", "FAIL")}, "evaluations": evaluations, } _write_json(args.output / "summary.json", summary) return summary def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run isolated Negative Act Form experiment") parser.add_argument("cases", type=Path) parser.add_argument("-o", "--output", type=Path, required=True) parser.add_argument("--model", default=DEFAULT_MODEL) parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT) parser.add_argument("--timeout", type=int, default=300) parser.add_argument("--num-ctx", type=int, default=16384) parser.add_argument("--num-predict", type=int, default=1024) return parser.parse_args() def main() -> int: summary = run_experiment(parse_args()) print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if summary["counts"]["FAIL"] == 0 and summary["technical_failed_call_count"] == 0 else 1 if __name__ == "__main__": raise SystemExit(main())