From 679d209738010e05345db506c7b4207e673a6606 Mon Sep 17 00:00:00 2001 From: witbrock Date: Tue, 28 Jul 2026 11:19:39 +0100 Subject: [PATCH 1/4] Harden POSIX scheduled backups --- run.sh | 751 +++++++++++++++++++++++---- scripts/backup_von_db.py | 19 +- tests/test_backup_von_db.py | 91 ++++ tests/test_run_sh_start_behaviour.py | 733 ++++++++++++++++++++++++++ 4 files changed, 1482 insertions(+), 112 deletions(-) diff --git a/run.sh b/run.sh index 5576c9162..7e4d0af09 100755 --- a/run.sh +++ b/run.sh @@ -131,6 +131,64 @@ if [ "$AGENT_TEST_INSTANCE" -eq 1 ]; then fi fi +load_env_from_dotenv() { + local env_path="$1" + if [ -z "$env_path" ] || [ ! -f "$env_path" ]; then + return 0 + fi + + local applied=0 + local line="" + local trimmed="" + local key="" + local value="" + local first_char="" + local last_char="" + + while IFS= read -r line || [ -n "$line" ]; do + # Handle CRLF files safely. + line="${line%$'\r'}" + + trimmed="$line" + trimmed="${trimmed#"${trimmed%%[![:space:]]*}"}" + trimmed="${trimmed%"${trimmed##*[![:space:]]}"}" + if [ -z "$trimmed" ]; then + continue + fi + if [[ "$trimmed" == \#* ]]; then + continue + fi + + if [[ "$trimmed" =~ ^([A-Za-z_][A-Za-z0-9_]*)[[:space:]]*=[[:space:]]*(.*)$ ]]; then + key="${BASH_REMATCH[1]}" + value="${BASH_REMATCH[2]}" + + value="${value#"${value%%[![:space:]]*}"}" + value="${value%"${value##*[![:space:]]}"}" + + if [ "${#value}" -ge 2 ]; then + first_char="${value:0:1}" + last_char="${value:${#value}-1:1}" + if [[ "$first_char" == '"' && "$last_char" == '"' ]] || [[ "$first_char" == "'" && "$last_char" == "'" ]]; then + value="${value:1:${#value}-2}" + fi + fi + + printf -v "$key" '%s' "$value" + export "$key" + applied=$((applied + 1)) + fi + done < "$env_path" + + if [ "$applied" -gt 0 ]; then + log "Loaded $applied .env override(s) from $env_path" + fi +} + +# Backup path and policy globals below must see the same .env values as the +# eventual backup process. Load them before resolving roots or migrating data. +load_env_from_dotenv "${ROOT}/.env" + PID_FILE="${RUN_DIR}/von_${PORT}.pid" CURRENT_LOG="${LOGS_DIR}/von_${PORT}_current.log" TS="$(date +%Y%m%d_%H%M%S 2>/dev/null || date +%Y%m%d_%H%M%S)" @@ -453,63 +511,6 @@ if [ "$ACTION" = "start" ] && [ "$BACKUP_FLAGS_SET" -eq 1 ]; then ACTION="backup" fi -load_env_from_dotenv() { - local env_path="$1" - if [ -z "$env_path" ] || [ ! -f "$env_path" ]; then - return 0 - fi - - local applied=0 - local line="" - local trimmed="" - local key="" - local value="" - local first_char="" - local last_char="" - - while IFS= read -r line || [ -n "$line" ]; do - # Handle CRLF files safely. - line="${line%$'\r'}" - - trimmed="$line" - trimmed="${trimmed#"${trimmed%%[![:space:]]*}"}" - trimmed="${trimmed%"${trimmed##*[![:space:]]}"}" - if [ -z "$trimmed" ]; then - continue - fi - if [[ "$trimmed" == \#* ]]; then - continue - fi - - if [[ "$trimmed" =~ ^([A-Za-z_][A-Za-z0-9_]*)[[:space:]]*=[[:space:]]*(.*)$ ]]; then - key="${BASH_REMATCH[1]}" - value="${BASH_REMATCH[2]}" - - value="${value#"${value%%[![:space:]]*}"}" - value="${value%"${value##*[![:space:]]}"}" - - if [ "${#value}" -ge 2 ]; then - first_char="${value:0:1}" - last_char="${value:${#value}-1:1}" - if [[ "$first_char" == '"' && "$last_char" == '"' ]] || [[ "$first_char" == "'" && "$last_char" == "'" ]]; then - value="${value:1:${#value}-2}" - fi - fi - - printf -v "$key" '%s' "$value" - export "$key" - applied=$((applied + 1)) - fi - done < "$env_path" - - if [ "$applied" -gt 0 ]; then - log "Loaded $applied .env override(s) from $env_path" - fi -} - -# Load .env overrides without executing arbitrary shell content. -load_env_from_dotenv "${ROOT}/.env" - if [ "$AGENT_TEST_INSTANCE" -eq 1 ]; then export VON_AGENT_TEST_INSTANCE=1 else @@ -2128,10 +2129,512 @@ cron_due() { return 1 } +resolve_daily_backup_success() { + local receipt_path="$1" + local sentinel_path="$2" + local backup_root="$3" + local expected_db_name="${VON_DB_NAME:-von_db}" + if [ -z "$expected_db_name" ]; then + expected_db_name="von_db" + fi + local py + py="$(python_cmd)" + if [ -z "$py" ]; then + log "[daily-backup] ERROR: Python is unavailable for backup receipt validation." + return 1 + fi + + "$py" - "$receipt_path" "$sentinel_path" "$backup_root" "$expected_db_name" <<'PY' +import datetime +import json +import os +import sys +from pathlib import Path + +receipt_path = Path(sys.argv[1]) +sentinel_path = Path(sys.argv[2]) +backup_root = Path(sys.argv[3]) +expected_db_name = sys.argv[4] +utc = datetime.timezone.utc +epoch = datetime.datetime(1970, 1, 1, tzinfo=utc) +max_future_skew = datetime.timedelta(minutes=5) +receipt_schema = "backup_success_receipt.v1" +expected_tag = "auto-daily" + + +def warn(message): + print(f"[daily-backup] WARN: {message}", file=sys.stderr) + + +try: + resolved_backup_root = backup_root.resolve(strict=True) +except OSError as exc: + warn(f"configured backup root '{backup_root}' is unavailable: {exc}") + raise SystemExit(1) + +if not resolved_backup_root.is_dir(): + warn(f"configured backup root '{resolved_backup_root}' is not a directory") + raise SystemExit(1) + + +def parse_completed(value): + if not isinstance(value, str) or not value.strip(): + raise ValueError("missing completed_at_utc") + text = value.strip() + parsed = datetime.datetime.fromisoformat( + text[:-1] + "+00:00" if text.endswith("Z") else text + ) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=utc) + parsed = parsed.astimezone(utc) + if parsed < epoch: + raise ValueError("completed_at_utc predates the supported epoch") + if parsed > datetime.datetime.now(utc) + max_future_skew: + raise ValueError("completed_at_utc is implausibly far in the future") + return parsed + + +def artifact_size(path): + try: + if path.is_file(): + return path.stat().st_size + if path.is_dir(): + for item in path.rglob("*"): + if item.is_file() and item.stat().st_size > 0: + return 1 + except OSError: + return 0 + return 0 + + +def read_receipt(path, *, require_valid_artifact, require_sidecar_binding): + if not path.is_file(): + return None + try: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("receipt payload is not an object") + if payload.get("schema_version") != receipt_schema: + raise ValueError("unsupported or missing schema_version") + if payload.get("tag") != expected_tag: + raise ValueError(f"tag must be {expected_tag!r}") + if payload.get("db_name") != expected_db_name: + raise ValueError(f"db_name must be {expected_db_name!r}") + completed = parse_completed(payload.get("completed_at_utc")) + out_root_text = payload.get("out_root") + if not isinstance(out_root_text, str) or not out_root_text.strip(): + raise ValueError("missing out_root") + if Path(out_root_text).expanduser().resolve(strict=True) != resolved_backup_root: + raise ValueError("out_root does not match the configured backup root") + artifact_text = payload.get("final_artifact_path") + if not isinstance(artifact_text, str) or not artifact_text.strip(): + raise ValueError("missing final_artifact_path") + artifact_path = Path(artifact_text).expanduser().resolve(strict=True) + if artifact_path.parent != resolved_backup_root: + raise ValueError( + "final_artifact_path must be a direct child of the configured backup root" + ) + if require_valid_artifact and artifact_size(artifact_path) <= 0: + raise ValueError("backup artefact is missing or empty") + reported_size = payload.get("artifact_size_bytes") + if ( + not isinstance(reported_size, int) + or isinstance(reported_size, bool) + or reported_size <= 0 + ): + raise ValueError("artifact_size_bytes must be a positive integer") + collection_count = payload.get("collection_count") + if ( + not isinstance(collection_count, int) + or isinstance(collection_count, bool) + or collection_count <= 0 + ): + raise ValueError("collection_count must be a positive integer") + if require_sidecar_binding: + expected_sidecar = artifact_path.with_name( + artifact_path.name + ".backup_receipt.json" + ) + if path.resolve(strict=True) != expected_sidecar.resolve(strict=False): + raise ValueError("sidecar name does not match final_artifact_path") + return { + "path": path, + "payload": payload, + "completed": completed, + "artifact_path": artifact_path, + "tag": str(payload.get("tag") or ""), + } + except Exception as exc: + warn(f"ignoring invalid backup receipt '{path}': {exc}") + return None + + +def write_private_atomic(path, content): + path.parent.mkdir(parents=True, exist_ok=True) + temp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") + fd = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(content) + os.replace(temp_path, path) + os.chmod(path, 0o600) + finally: + try: + temp_path.unlink() + except FileNotFoundError: + pass + + +launcher = read_receipt( + receipt_path, + require_valid_artifact=True, + require_sidecar_binding=False, +) +newest = None +if resolved_backup_root.is_dir(): + for sidecar in resolved_backup_root.glob("*.backup_receipt.json"): + candidate = read_receipt( + sidecar, + require_valid_artifact=True, + require_sidecar_binding=True, + ) + if candidate is None: + continue + if newest is None or candidate["completed"] > newest["completed"]: + newest = candidate + +source = "" +selected = None +if newest is not None and ( + launcher is None or newest["completed"] > launcher["completed"] +): + launcher_summary = ( + str(launcher["completed"].isoformat()) if launcher is not None else "missing" + ) + warn( + "authoritative launcher receipt is stale/missing " + f"({launcher_summary}); repairing from newest validated " + f"backup artefact receipt '{newest['path']}'" + ) + payload_text = json.dumps(newest["payload"], indent=2, ensure_ascii=False) + "\n" + write_private_atomic(receipt_path, payload_text) + completed_text = ( + newest["completed"].isoformat(timespec="microseconds").replace("+00:00", "Z") + ) + write_private_atomic(sentinel_path, completed_text + "\n") + selected = newest + source = "artifact_receipt_repaired" +elif launcher is not None: + selected = launcher + source = "launcher_receipt" + +if selected is None and sentinel_path.is_file(): + try: + legacy_text = sentinel_path.read_text(encoding="utf-8").strip() + legacy_completed = parse_completed(legacy_text) + warn( + "falling back to legacy last_backup_utc.txt sentinel because " + "no valid structured backup receipt is available" + ) + selected = {"completed": legacy_completed} + source = "legacy_sentinel" + except Exception as exc: + warn(f"ignoring invalid legacy backup sentinel '{sentinel_path}': {exc}") + +if selected is not None: + completed = selected["completed"] + canonical = completed.isoformat(timespec="microseconds").replace("+00:00", "Z") + delta = completed - epoch + epoch_us = ( + delta.days * 86400 * 1_000_000 + + delta.seconds * 1_000_000 + + delta.microseconds + ) + print(f"{canonical}|{epoch_us}|{source}") +PY +} + +daily_backup_record_iso() { + local record="${1:-}" + if [ -n "$record" ]; then + printf '%s' "${record%%|*}" + fi +} + +daily_backup_record_epoch_us() { + local record="${1:-}" + if [ -z "$record" ] || [ "${record#*|}" = "$record" ]; then + return 0 + fi + local remainder="${record#*|}" + printf '%s' "${remainder%%|*}" +} + +daily_backup_record_source() { + local record="${1:-}" + if [ -z "$record" ] || [ "${record#*|}" = "$record" ]; then + return 0 + fi + local remainder="${record#*|}" + if [ "${remainder#*|}" = "$remainder" ]; then + return 0 + fi + printf '%s' "${remainder#*|}" +} + +daily_backup_success_advanced() { + local current_record="${1:-}" + local baseline_record="${2:-}" + local current_source + current_source="$(daily_backup_record_source "$current_record")" + if [ "$current_source" != "launcher_receipt" ] && [ "$current_source" != "artifact_receipt_repaired" ]; then + return 1 + fi + local current_epoch + local baseline_epoch + current_epoch="$(daily_backup_record_epoch_us "$current_record")" + baseline_epoch="$(daily_backup_record_epoch_us "$baseline_record")" + if ! printf '%s' "$current_epoch" | grep -qE '^[0-9]+$'; then + return 1 + fi + if [ -z "$baseline_epoch" ]; then + return 0 + fi + printf '%s' "$baseline_epoch" | grep -qE '^[0-9]+$' || return 1 + [ "$current_epoch" -gt "$baseline_epoch" ] +} + +process_start_marker() { + local pid="$1" + local marker="" + marker="$(ps -o lstart= -p "$pid" 2>/dev/null || true)" + marker="${marker#"${marker%%[![:space:]]*}"}" + marker="${marker%"${marker##*[![:space:]]}"}" + printf '%s' "$marker" +} + +cleanup_daily_backup_lock() { + local lock_dir="$1" + local owner_token="$2" + local owner_pid="$3" + local pid_file="${RUN_DIR}/daily_backup.pid" + local observed_token="" + observed_token="$(tr -d '\r\n' < "${lock_dir}/owner" 2>/dev/null || true)" + if [ -z "$observed_token" ] || [ "$observed_token" != "$owner_token" ]; then + return 0 + fi + local observed_pid="" + observed_pid="$(tr -d '\r\n' < "$pid_file" 2>/dev/null || true)" + if [ "$observed_pid" = "$owner_pid" ]; then + rm -f "$pid_file" 2>/dev/null || true + fi + rm -f \ + "${lock_dir}/baseline_success" \ + "${lock_dir}/launcher.pid" \ + "${lock_dir}/launcher.start" \ + "${lock_dir}/owner" \ + "${lock_dir}/ready" \ + 2>/dev/null || true + rmdir "$lock_dir" 2>/dev/null || true +} + +acquire_daily_backup_lock() { + local lock_dir="$1" + local baseline_record="$2" + local owner_token="$3" + local pid_file="${RUN_DIR}/daily_backup.pid" + local owner_start="" + owner_start="$(process_start_marker "$$")" + if [ -z "$owner_start" ]; then + log "[daily-backup] ERROR: could not establish the backup owner's process identity." + return 1 + fi + if ! ( + umask 077 + mkdir "$lock_dir" 2>/dev/null + ); then + return 1 + fi + chmod 700 "$lock_dir" 2>/dev/null || true + if ! ( + umask 077 + printf '%s\n' "$owner_token" > "${lock_dir}/owner" + printf '%s\n' "$$" > "${lock_dir}/launcher.pid" + printf '%s\n' "$owner_start" > "${lock_dir}/launcher.start" + printf '%s\n' "$baseline_record" > "${lock_dir}/baseline_success" + printf '%s\n' "$$" > "$pid_file" + : > "${lock_dir}/ready" + ); then + cleanup_daily_backup_lock "$lock_dir" "$owner_token" "$$" + rm -f \ + "${lock_dir}/baseline_success" \ + "${lock_dir}/launcher.pid" \ + "${lock_dir}/launcher.start" \ + "${lock_dir}/owner" \ + "${lock_dir}/ready" \ + 2>/dev/null || true + rmdir "$lock_dir" 2>/dev/null || true + return 1 + fi + return 0 +} + +wait_for_existing_daily_backup() { + local lock_dir="$1" + local fallback_baseline="$2" + local receipt_path="$3" + local sentinel_path="$4" + local out_dir="$5" + local attempts=0 + while [ "$attempts" -lt 100 ] && [ ! -f "${lock_dir}/ready" ]; do + [ -d "$lock_dir" ] || break + sleep 0.05 + attempts=$((attempts + 1)) + done + + local owner_token="" + local owner_pid="" + local owner_start="" + local baseline_record="" + owner_token="$(tr -d '\r\n' < "${lock_dir}/owner" 2>/dev/null || true)" + owner_pid="$(tr -d '\r\n' < "${lock_dir}/launcher.pid" 2>/dev/null || true)" + owner_start="$(tr -d '\r\n' < "${lock_dir}/launcher.start" 2>/dev/null || true)" + baseline_record="$(tr -d '\r\n' < "${lock_dir}/baseline_success" 2>/dev/null || true)" + if [ -z "$baseline_record" ]; then + baseline_record="$fallback_baseline" + fi + if [ -z "$owner_token" ] || [ -z "$owner_start" ] || ! printf '%s' "$owner_pid" | grep -qE '^[0-9]+$'; then + if [ ! -d "$lock_dir" ]; then + local completed_after_disappearance="" + completed_after_disappearance="$(resolve_daily_backup_success "$receipt_path" "$sentinel_path" "$out_dir")" || return 1 + if daily_backup_success_advanced "$completed_after_disappearance" "$baseline_record"; then + log "[daily-backup] Existing scheduled backup completed with validated success evidence." + return 0 + fi + fi + if [ -n "$owner_token" ]; then + cleanup_daily_backup_lock "$lock_dir" "$owner_token" "$owner_pid" + else + local current_token="" + current_token="$(tr -d '\r\n' < "${lock_dir}/owner" 2>/dev/null || true)" + if [ -z "$current_token" ]; then + rm -f \ + "${lock_dir}/baseline_success" \ + "${lock_dir}/launcher.pid" \ + "${lock_dir}/launcher.start" \ + "${lock_dir}/owner" \ + "${lock_dir}/ready" \ + 2>/dev/null || true + rmdir "$lock_dir" 2>/dev/null || true + fi + fi + log "[daily-backup] ERROR: existing backup lock has no valid owner metadata." + return 1 + fi + + local wait_timeout_seconds=21600 + if printf '%s' "${VON_BACKUP_WAIT_TIMEOUT_SECONDS:-}" | grep -qE '^[0-9]+$' && + [ "${VON_BACKUP_WAIT_TIMEOUT_SECONDS}" -ge 1 ]; then + wait_timeout_seconds="${VON_BACKUP_WAIT_TIMEOUT_SECONDS}" + fi + local wait_started_epoch + wait_started_epoch="$(date -u +%s)" + log "[daily-backup] Waiting for existing scheduled backup PID=$owner_pid." + while [ -d "$lock_dir" ]; do + local current_token="" + current_token="$(tr -d '\r\n' < "${lock_dir}/owner" 2>/dev/null || true)" + if [ "$current_token" != "$owner_token" ]; then + break + fi + local current_start="" + current_start="$(process_start_marker "$owner_pid")" + if [ -z "$current_start" ] || [ "$current_start" != "$owner_start" ]; then + break + fi + local wait_now_epoch + wait_now_epoch="$(date -u +%s)" + if [ $((wait_now_epoch - wait_started_epoch)) -ge "$wait_timeout_seconds" ]; then + log "[daily-backup] ERROR: timed out after ${wait_timeout_seconds}s waiting for existing scheduled backup PID=$owner_pid." + return 1 + fi + sleep 0.2 + done + + local completed_record="" + completed_record="$(resolve_daily_backup_success "$receipt_path" "$sentinel_path" "$out_dir")" || return 1 + if daily_backup_success_advanced "$completed_record" "$baseline_record"; then + cleanup_daily_backup_lock "$lock_dir" "$owner_token" "$owner_pid" + log "[daily-backup] Existing scheduled backup completed with validated success evidence." + return 0 + fi + cleanup_daily_backup_lock "$lock_dir" "$owner_token" "$owner_pid" + log "[daily-backup] ERROR: existing scheduled backup ended without newly validated success evidence." + return 1 +} + +launch_daily_backup_detached() { + local detacher_py + detacher_py="$(python_cmd)" + local bash_exe + bash_exe="$(command -v bash 2>/dev/null || true)" + if [ -z "$detacher_py" ] || [ -z "$bash_exe" ]; then + log "[daily-backup] ERROR: cannot detach scheduled backup without Python and Bash." + return 1 + fi + local stdout_log="${RUN_DIR}/daily_backup_launcher.log" + local stderr_log="${RUN_DIR}/daily_backup_launcher.err.log" + local detached_pid="" + detached_pid="$( + ( + umask 077 + launch_detached_process \ + "$detacher_py" \ + "$stdout_log" \ + "$stderr_log" \ + "$ROOT" \ + "$bash_exe" \ + "${ROOT}/run.sh" \ + scheduled-backup \ + -NoBackupMigrate + ) 2>>"$stderr_log" || true + )" + if ! printf '%s' "$detached_pid" | grep -qE '^[0-9]+$'; then + log "[daily-backup] ERROR: detached scheduled-backup launcher did not return a PID." + return 1 + fi + chmod 600 "$stdout_log" "$stderr_log" 2>/dev/null || true + log "[daily-backup] Launched detached scheduled-backup PID=$detached_pid." + return 0 +} + run_daily_backup_if_due() { + local mode="${1:-background}" + if ! is_truthy "${VON_ENABLE_DAILY_BACKUP:-}"; then + log "[daily-backup] Skip: automatic backups disabled. Set VON_ENABLE_DAILY_BACKUP=1 on a designated backup host to enable." + return 0 + fi if is_truthy "${VON_DISABLE_DAILY_BACKUP:-}"; then + log "[daily-backup] Skip: disabled via VON_DISABLE_DAILY_BACKUP." return 0 fi + + local backup_script="${ROOT}/scripts/backup_von_db.py" + if [ ! -f "$backup_script" ]; then + log "[daily-backup] ERROR: backup script missing: $backup_script" + return 1 + fi + local out_dir + out_dir="$(resolve_backup_out_dir "$BACKUP_ROOT" "$LOCAL_BACKUPS" "daily-backup")" + if ! backup_output_path_allowed "$out_dir" "daily-backup" 1; then + log "[daily-backup] WARN: skipping scheduled backup until VON_BACKUP_ROOT points outside repo (or VON_ALLOW_BACKUP_IN_REPO=1)." + return 1 + fi + + local receipt="${RUN_DIR}/last_successful_backup_receipt.json" + local sentinel="${RUN_DIR}/last_backup_utc.txt" + local last_record="" + last_record="$(resolve_daily_backup_success "$receipt" "$sentinel" "$out_dir")" || return 1 + local last + last="$(daily_backup_record_iso "$last_record")" local schedule="${VON_BACKUP_SCHEDULE:-}" if [ -n "$schedule" ]; then schedule="${schedule#\"}" @@ -2144,11 +2647,6 @@ run_daily_backup_if_due() { if [ "$interval_hours" -lt 1 ]; then interval_hours=24 fi - local sentinel="${RUN_DIR}/last_backup_utc.txt" - local last="" - if [ -f "$sentinel" ]; then - last="$(tr -d '\r\n' < "$sentinel" 2>/dev/null || true)" - fi local now now="$(date -u +"%Y-%m-%dT%H:%M:%SZ")" local due=1 @@ -2166,16 +2664,15 @@ run_daily_backup_if_due() { fi fi if [ -z "$schedule" ]; then - if [ -n "$last" ]; then - local last_epoch - last_epoch="$(parse_iso_epoch "$last")" - if [ -n "$last_epoch" ]; then - local now_epoch - now_epoch="$(date -u +%s)" - local elapsed_hours=$(( (now_epoch - last_epoch) / 3600 )) - if [ "$elapsed_hours" -lt "$interval_hours" ]; then - due=0 - fi + local last_epoch_us + last_epoch_us="$(daily_backup_record_epoch_us "$last_record")" + if printf '%s' "$last_epoch_us" | grep -qE '^[0-9]+$'; then + local now_epoch + now_epoch="$(date -u +%s)" + local last_epoch=$((last_epoch_us / 1000000)) + local elapsed_hours=$(( (now_epoch - last_epoch) / 3600 )) + if [ "$elapsed_hours" -lt "$interval_hours" ]; then + due=0 fi fi fi @@ -2183,47 +2680,57 @@ run_daily_backup_if_due() { return 0 fi - local pid_file="${RUN_DIR}/daily_backup.pid" - if [ -f "$pid_file" ]; then - local pid - pid="$(tr -d '\r\n' < "$pid_file" 2>/dev/null || true)" - if [ -n "$pid" ] && process_exists "$pid"; then - return 0 - fi - rm -f "$pid_file" 2>/dev/null || true + if [ "$mode" != "wait" ]; then + launch_daily_backup_detached + return $? fi - local backup_script="${ROOT}/scripts/backup_von_db.py" - if [ ! -f "$backup_script" ]; then - log "[daily-backup] WARN: backup script missing: $backup_script (skipping)" - return 0 + local lock_dir="${RUN_DIR}/daily_backup.lock" + local owner_token="$$-${RANDOM}-${RANDOM}" + if ! acquire_daily_backup_lock "$lock_dir" "$last_record" "$owner_token"; then + wait_for_existing_daily_backup "$lock_dir" "$last_record" "$receipt" "$sentinel" "$out_dir" + return $? fi + + trap 'cleanup_daily_backup_lock "$lock_dir" "$owner_token" "$$"' EXIT + log "[daily-backup] Running scheduled backup under owner lock (interval ${interval_hours}h)." local pdm pdm="$(pdm_cmd)" - local out_dir - out_dir="$(resolve_backup_out_dir "$BACKUP_ROOT" "$LOCAL_BACKUPS" "daily-backup")" - if ! backup_output_path_allowed "$out_dir" "daily-backup" 1; then - log "[daily-backup] WARN: skipping scheduled backup until VON_BACKUP_ROOT points outside repo (or VON_ALLOW_BACKUP_IN_REPO=1)." - return 0 - fi - log "[daily-backup] Launching background backup (interval ${interval_hours}h)..." - ( - set +e + local output_log="${RUN_DIR}/daily_backup_last_output.log" + local backup_exit=0 + if ( cd "$ROOT" || exit 1 + umask 077 export MONGO_ALLOW_LOCAL_FALLBACK=0 - local output_log="${RUN_DIR}/daily_backup_last_output.log" - "$pdm" run python "$backup_script" --apply --out-dir "$out_dir" --tag auto-daily >> "$output_log" 2>&1 - local exit_code=$? - if [ "$exit_code" -ne 0 ]; then - log "[daily-backup] ERROR exit=$exit_code" - exit 0 + "$pdm" run python "$backup_script" \ + --apply \ + --out-dir "$out_dir" \ + --tag auto-daily \ + --launcher-receipt-path "$receipt" \ + --legacy-sentinel-path "$sentinel" \ + >> "$output_log" 2>&1 + ); then + backup_exit=0 + else + backup_exit=$? + fi + + if [ "$backup_exit" -eq 0 ]; then + local completed_record="" + completed_record="$(resolve_daily_backup_success "$receipt" "$sentinel" "$out_dir")" || backup_exit=1 + if [ "$backup_exit" -eq 0 ] && ! daily_backup_success_advanced "$completed_record" "$last_record"; then + log "[daily-backup] ERROR: backup command exited successfully without newly validated success evidence." + backup_exit=1 fi - run_code_mention_scan "daily-backup" - run_code_predicate_sync "daily-backup" - date -u +"%Y-%m-%dT%H:%M:%SZ" > "$sentinel" 2>/dev/null || true - log "[daily-backup] Completed." - ) & - printf '%s\n' "$!" > "$pid_file" 2>/dev/null || true + fi + if [ "$backup_exit" -eq 0 ]; then + log "[daily-backup] Completed with validated success evidence." + else + log "[daily-backup] ERROR exit=$backup_exit" + fi + cleanup_daily_backup_lock "$lock_dir" "$owner_token" "$$" + trap - EXIT + return "$backup_exit" } trigger_test_db_refresh() { @@ -2840,16 +3347,28 @@ run_backup() { return 1 fi log "[backup] Starting backup (mode=$mode tag=$BACKUP_TAG out=$out_dir)" + local backup_exit=0 if [ "$BACKUP_DRY_RUN" -eq 1 ]; then - "$pdm" run python "$backup_script" --out-dir "$out_dir" --tag "$BACKUP_TAG" + if ( + umask 077 + "$pdm" run python "$backup_script" --out-dir "$out_dir" --tag "$BACKUP_TAG" + ); then + backup_exit=0 + else + backup_exit=$? + fi else - "$pdm" run python "$backup_script" --apply --out-dir "$out_dir" --tag "$BACKUP_TAG" + if ( + umask 077 + "$pdm" run python "$backup_script" --apply --out-dir "$out_dir" --tag "$BACKUP_TAG" + ); then + backup_exit=0 + else + backup_exit=$? + fi fi - local backup_exit=$? if [ "$backup_exit" -eq 0 ]; then log "[backup] OK" - run_code_mention_scan "manual-backup" - run_code_predicate_sync "manual-backup" if [ "$AGENT_TEST_INSTANCE" -eq 1 ]; then log "[backup-migrate] disabled via -AgentTest" elif [ "$NO_BACKUP_MIGRATE" -eq 0 ]; then @@ -2979,7 +3498,7 @@ show_help() { cat <<'TXT' Von Launcher Help Usage: ./run.sh [action] [options] - Actions: start | foreground | stop | status | restart | logs | check | backup | restore-backup | autoupdate | rag-worker | help + Actions: start | foreground | stop | status | restart | logs | check | backup | scheduled-backup | restore-backup | autoupdate | rag-worker | help Options: -Port Server port (default 5001 on macOS, 5000 elsewhere; -AgentTest defaults to 5010) -AgentTest Isolated coding-agent test instance mode @@ -3014,6 +3533,7 @@ Von Launcher Help Backup safety environment variables: VON_ENABLE_BACKUP_ACTION=1 + VON_ENABLE_DAILY_BACKUP=1 VON_ENABLE_RESTORE_ACTION=1 VON_ALLOW_BACKUP_IN_REPO=1 @@ -3028,6 +3548,7 @@ Von Launcher Help ./run.sh restart -AgentTest -HealthTimeoutSec 180 ./run.sh logs -Tail 200 -Follow ./run.sh backup -BackupDryRun + ./run.sh scheduled-backup -NoBackupMigrate ./run.sh restore-backup -RestoreBackupPath /path/to/last_successful_backup_receipt.json ./run.sh autoupdate -UpdateIntervalMinutes 30 -UpdateBranch main TXT @@ -3063,6 +3584,14 @@ case "$ACTION" in backup) run_backup ;; + scheduled-backup) + if run_daily_backup_if_due wait; then + exit 0 + else + scheduled_backup_exit=$? + exit "$scheduled_backup_exit" + fi + ;; restore-backup) run_restore_backup ;; diff --git a/scripts/backup_von_db.py b/scripts/backup_von_db.py index 72a9359fb..5c41e3076 100644 --- a/scripts/backup_von_db.py +++ b/scripts/backup_von_db.py @@ -12,6 +12,7 @@ - VON_BACKUP_COMPRESSION_ENABLED (truthy/falsey) - VON_BACKUP_ENCRYPTION_ENABLED (truthy/falsey) - VON_BACKUP_ENCRYPTION_KEY (Fernet key; required if encryption enabled) +- VON_BACKUP_MONGODUMP_TIMEOUT_SECONDS (int; default 21600 / 6 hours) """ from __future__ import annotations @@ -33,6 +34,7 @@ BACKUP_SUCCESS_RECEIPT_SCHEMA_VERSION = "backup_success_receipt.v1" BACKUP_RECEIPT_SIDECAR_SUFFIX = ".backup_receipt.json" +DEFAULT_MONGODUMP_TIMEOUT_SECONDS = 6 * 60 * 60 def _env_truthy(value: str | None) -> bool: @@ -375,7 +377,22 @@ def _run_mongodump(*, mongo_uri: str, db_name: str, out_path: Path) -> None: "--out", str(out_path), ] - subprocess.run(cmd, check=True) + timeout_seconds = _env_int( + os.environ.get("VON_BACKUP_MONGODUMP_TIMEOUT_SECONDS"), + default=DEFAULT_MONGODUMP_TIMEOUT_SECONDS, + ) + if timeout_seconds < 1: + timeout_seconds = DEFAULT_MONGODUMP_TIMEOUT_SECONDS + try: + subprocess.run(cmd, check=True, timeout=timeout_seconds) + except subprocess.TimeoutExpired: + raise RuntimeError( + f"mongodump exceeded the configured {timeout_seconds}s timeout" + ) from None + except subprocess.CalledProcessError as exc: + raise RuntimeError( + f"mongodump failed with exit code {exc.returncode}" + ) from None def main(argv: list[str] | None = None) -> int: diff --git a/tests/test_backup_von_db.py b/tests/test_backup_von_db.py index 4f3fb96a4..6ba45fb72 100644 --- a/tests/test_backup_von_db.py +++ b/tests/test_backup_von_db.py @@ -3,6 +3,7 @@ import json import os import time +import traceback from pathlib import Path import pytest @@ -22,6 +23,96 @@ def _touch(path: Path, *, age_seconds: int) -> None: os.utime(path, (ts, ts)) +@pytest.mark.parametrize( + ("configured_timeout", "expected_timeout"), + [ + (None, backup_von_db.DEFAULT_MONGODUMP_TIMEOUT_SECONDS), + ("90", 90), + ("0", backup_von_db.DEFAULT_MONGODUMP_TIMEOUT_SECONDS), + ("invalid", backup_von_db.DEFAULT_MONGODUMP_TIMEOUT_SECONDS), + ], +) +def test_mongodump_uses_bounded_timeout( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + configured_timeout: str | None, + expected_timeout: int, +) -> None: + if configured_timeout is None: + monkeypatch.delenv("VON_BACKUP_MONGODUMP_TIMEOUT_SECONDS", raising=False) + else: + monkeypatch.setenv( + "VON_BACKUP_MONGODUMP_TIMEOUT_SECONDS", + configured_timeout, + ) + observed: dict[str, object] = {} + + def _fake_run( + command: list[str], + *, + check: bool, + timeout: int, + ) -> None: + observed.update(command=command, check=check, timeout=timeout) + + monkeypatch.setattr(backup_von_db.subprocess, "run", _fake_run) + + backup_von_db._run_mongodump( + mongo_uri="mongodb://example.invalid/", + db_name="von_db", + out_path=tmp_path, + ) + + assert observed["check"] is True + assert observed["timeout"] == expected_timeout + + +@pytest.mark.parametrize( + ("failure_kind", "expected_message"), + [ + ("timeout", "exceeded the configured"), + ("nonzero", "failed with exit code 19"), + ], +) +def test_mongodump_failure_traceback_does_not_expose_uri_credentials( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, + expected_message: str, +) -> None: + mongo_uri = "mongodb://backup-user:super-secret-password@example.invalid/von_db" + + def _fake_run( + command: list[str], + *, + check: bool, + timeout: int, + ) -> None: + if failure_kind == "timeout": + raise backup_von_db.subprocess.TimeoutExpired(command, timeout) + raise backup_von_db.subprocess.CalledProcessError(19, command) + + monkeypatch.setattr(backup_von_db.subprocess, "run", _fake_run) + + with pytest.raises(RuntimeError) as caught: + backup_von_db._run_mongodump( + mongo_uri=mongo_uri, + db_name="von_db", + out_path=tmp_path, + ) + + rendered = "".join( + traceback.format_exception( + caught.type, + caught.value, + caught.tb, + ) + ) + assert expected_message in rendered + assert mongo_uri not in rendered + assert "super-secret-password" not in rendered + + def test_retention_deletes_old_backups(tmp_path: Path) -> None: out_root = tmp_path / "backups" old_dir = out_root / "von_db_20000101_000000Z_auto-daily" diff --git a/tests/test_run_sh_start_behaviour.py b/tests/test_run_sh_start_behaviour.py index c04bb712f..a50d315bd 100644 --- a/tests/test_run_sh_start_behaviour.py +++ b/tests/test_run_sh_start_behaviour.py @@ -1,8 +1,11 @@ from __future__ import annotations import json +import os import shutil import subprocess +import time +from datetime import datetime, timedelta, timezone from pathlib import Path import pytest @@ -37,6 +40,736 @@ def shlex_quote(value: str) -> str: return "'" + value.replace("'", "'\"'\"'") + "'" +def _prepare_backup_launcher( + tmp_path: Path, + *, + enable_daily_backup: bool, + backup_exit: int = 0, + backup_delay: float = 0.15, + artifact_size: int = 16, +) -> dict[str, Path]: + launcher_root = tmp_path / "launcher" + launcher_root.mkdir() + run_sh = launcher_root / "run.sh" + shutil.copy2(REPO_ROOT / "run.sh", run_sh) + + backup_script = launcher_root / "scripts" / "backup_von_db.py" + backup_script.parent.mkdir() + backup_script.write_text("# fake backup entrypoint\n", encoding="utf-8") + + args_path = tmp_path / "backup_args.txt" + calls_path = tmp_path / "backup_calls.txt" + created_path = tmp_path / "created_by_backup.txt" + fake_pdm = launcher_root / ".venv" / "bin" / "pdm" + fake_pdm.parent.mkdir(parents=True) + fake_pdm.write_text( + f"""#!/usr/bin/env python3 +import datetime +import json +import sys +import time +from pathlib import Path + +args = sys.argv[1:] +args_path = Path({str(args_path)!r}) +calls_path = Path({str(calls_path)!r}) +created_path = Path({str(created_path)!r}) +args_path.write_text("\\n".join(args) + "\\n", encoding="utf-8") +with calls_path.open("a", encoding="utf-8") as handle: + handle.write("call\\n") +created_path.write_text("created\\n", encoding="utf-8") +time.sleep({backup_delay!r}) + +if {backup_exit} == 0 and "--launcher-receipt-path" in args: + def option(name): + return args[args.index(name) + 1] + + out_dir = Path(option("--out-dir")) + out_dir.mkdir(parents=True, exist_ok=True) + artifact = out_dir / "fake_auto_daily.zip" + artifact.write_bytes(b"x" * {artifact_size}) + completed_at = datetime.datetime.now(datetime.timezone.utc).isoformat().replace( + "+00:00", "Z" + ) + payload = {{ + "schema_version": "backup_success_receipt.v1", + "completed_at_utc": completed_at, + "db_name": "von_db", + "tag": "auto-daily", + "out_root": str(out_dir), + "backup_root": str(artifact), + "final_artifact_path": str(artifact), + "artifact_kind": "zip", + "compressed": True, + "encrypted": False, + "artifact_size_bytes": artifact.stat().st_size, + "collection_count": 1, + }} + receipt_text = json.dumps(payload, indent=2) + "\\n" + Path(option("--launcher-receipt-path")).write_text( + receipt_text, encoding="utf-8" + ) + Path(option("--legacy-sentinel-path")).write_text( + completed_at + "\\n", encoding="utf-8" + ) + artifact.with_name(artifact.name + ".backup_receipt.json").write_text( + receipt_text, encoding="utf-8" + ) + +raise SystemExit({backup_exit}) +""", + encoding="utf-8", + ) + fake_pdm.chmod(0o755) + + backup_root = tmp_path / "external-backups" + (launcher_root / ".env").write_text( + "\n".join( + [ + f"VON_ENABLE_DAILY_BACKUP={'1' if enable_daily_backup else '0'}", + "VON_ENABLE_BACKUP_ACTION=1", + "VON_DB_NAME=von_db", + f'VON_BACKUP_ROOT="{backup_root}"', + "VON_BACKUP_INTERVAL_HOURS=24", + "", + ] + ), + encoding="utf-8", + ) + return { + "root": launcher_root, + "run_sh": run_sh, + "backup_script": backup_script, + "backup_root": backup_root, + "args": args_path, + "calls": calls_path, + "created": created_path, + } + + +def _run_prepared_launcher( + prepared: dict[str, Path], action: str +) -> subprocess.CompletedProcess[str]: + bash = shutil.which("bash") + if not bash: + pytest.skip("bash is required for run.sh launcher-path tests") + return subprocess.run( + [bash, str(prepared["run_sh"]), action, "-NoBackupMigrate"], + cwd=prepared["root"], + text=True, + capture_output=True, + timeout=20, + check=False, + ) + + +def _write_test_backup_receipt( + receipt_path: Path, + *, + artifact_path: Path, + completed_at: datetime, + tag: str = "auto-daily", + schema_version: str = "backup_success_receipt.v1", + db_name: str = "von_db", + out_root: Path | None = None, +) -> None: + receipt_path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "schema_version": schema_version, + "completed_at_utc": completed_at.astimezone(timezone.utc) + .isoformat() + .replace("+00:00", "Z"), + "db_name": db_name, + "tag": tag, + "out_root": str(out_root or artifact_path.parent), + "backup_root": str(artifact_path), + "final_artifact_path": str(artifact_path), + "artifact_kind": "zip", + "compressed": True, + "encrypted": False, + "artifact_size_bytes": artifact_path.stat().st_size, + "collection_count": 1, + } + receipt_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + + +def _process_start_marker(pid: int) -> str: + return subprocess.check_output( + ["ps", "-o", "lstart=", "-p", str(pid)], + text=True, + ).strip() + + +def _write_test_backup_lock( + prepared: dict[str, Path], + *, + owner_pid: int, + owner_start: str, + include_owner_metadata: bool = True, +) -> Path: + lock_dir = prepared["root"] / ".run" / "daily_backup.lock" + lock_dir.mkdir(parents=True) + if include_owner_metadata: + (lock_dir / "owner").write_text("test-owner\n", encoding="utf-8") + (lock_dir / "launcher.pid").write_text( + f"{owner_pid}\n", + encoding="utf-8", + ) + (lock_dir / "launcher.start").write_text( + owner_start + "\n", + encoding="utf-8", + ) + (lock_dir / "baseline_success").write_text("\n", encoding="utf-8") + (lock_dir / "ready").write_text("", encoding="utf-8") + return lock_dir + + +def _run_concurrent_scheduled_backups( + prepared: dict[str, Path], +) -> tuple[tuple[int, str, str], tuple[int, str, str]]: + bash = shutil.which("bash") + if not bash: + pytest.skip("bash is required for run.sh launcher-path tests") + command = [ + bash, + str(prepared["run_sh"]), + "scheduled-backup", + "-NoBackupMigrate", + ] + first = subprocess.Popen( + command, + cwd=prepared["root"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + ready_path = prepared["root"] / ".run" / "daily_backup.lock" / "ready" + deadline = time.monotonic() + 5 + while ( + time.monotonic() < deadline + and not ready_path.exists() + and first.poll() is None + ): + time.sleep(0.01) + assert ready_path.exists(), "first scheduled backup did not acquire its owner lock" + lock_dir = ready_path.parent + assert lock_dir.stat().st_mode & 0o777 == 0o700 + for private_name in ( + "owner", + "launcher.pid", + "launcher.start", + "baseline_success", + "ready", + ): + assert (lock_dir / private_name).stat().st_mode & 0o777 == 0o600 + + second = subprocess.Popen( + command, + cwd=prepared["root"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + first_out, first_err = first.communicate(timeout=20) + second_out, second_err = second.communicate(timeout=20) + return ( + (first.returncode, first_out, first_err), + (second.returncode, second_out, second_err), + ) + + +def test_run_sh_loads_dotenv_before_backup_path_globals(tmp_path: Path) -> None: + launcher_root = tmp_path / "launcher" + launcher_root.mkdir() + run_sh = launcher_root / "run.sh" + shutil.copy2(REPO_ROOT / "run.sh", run_sh) + backup_root = tmp_path / "external-backups" + (launcher_root / ".env").write_text( + "\n".join( + [ + f'VON_BACKUP_ROOT="{backup_root}"', + "VON_ALLOW_BACKUP_IN_REPO=1", + "", + ] + ), + encoding="utf-8", + ) + + bash = shutil.which("bash") + if not bash: + pytest.skip("bash is required for run.sh launcher-path tests") + result = subprocess.run( + [ + bash, + "-c", + ( + f". {shlex_quote(str(run_sh))} help -NoBackupMigrate >/dev/null\n" + "printf '%s\\n%s\\n' \"$BACKUP_ROOT\" \"$ALLOW_REPO_BACKUP_OUTPUT\"" + ), + ], + cwd=launcher_root, + text=True, + capture_output=True, + timeout=20, + check=True, + ) + + assert result.stdout.splitlines() == [str(backup_root), "1"] + + +def test_run_sh_scheduled_backup_requires_explicit_host_opt_in( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=False) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0 + assert not prepared["args"].exists() + assert "automatic backups disabled" in result.stdout + + +def test_run_sh_scheduled_backup_waits_and_passes_receipt_paths( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["created"].exists() + assert prepared["created"].stat().st_mode & 0o777 == 0o600 + launcher_receipt = ( + prepared["root"] / ".run" / "last_successful_backup_receipt.json" + ) + artifact = prepared["backup_root"] / "fake_auto_daily.zip" + sidecar = artifact.with_name(artifact.name + ".backup_receipt.json") + assert launcher_receipt.stat().st_mode & 0o777 == 0o600 + assert artifact.stat().st_mode & 0o777 == 0o600 + assert sidecar.stat().st_mode & 0o777 == 0o600 + assert prepared["args"].read_text(encoding="utf-8").splitlines() == [ + "run", + "python", + str(prepared["backup_script"]), + "--apply", + "--out-dir", + str(prepared["backup_root"]), + "--tag", + "auto-daily", + "--launcher-receipt-path", + str(prepared["root"] / ".run" / "last_successful_backup_receipt.json"), + "--legacy-sentinel-path", + str(prepared["root"] / ".run" / "last_backup_utc.txt"), + ] + + +def test_run_sh_scheduled_backup_returns_backup_failure( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher( + tmp_path, + enable_daily_backup=True, + backup_exit=23, + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 23 + assert prepared["created"].exists() + assert "[daily-backup] ERROR exit=23" in result.stdout + + +def test_run_sh_concurrent_scheduled_backups_launch_one_dump( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher( + tmp_path, + enable_daily_backup=True, + backup_delay=0.4, + ) + + first, second = _run_concurrent_scheduled_backups(prepared) + + assert first[0] == 0, first[1] + first[2] + assert second[0] == 0, second[1] + second[2] + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert "Waiting for existing scheduled backup" in second[1] + assert "validated success evidence" in second[1] + + +def test_run_sh_concurrent_waiter_fails_without_new_success_evidence( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher( + tmp_path, + enable_daily_backup=True, + backup_exit=23, + backup_delay=0.4, + ) + + first, second = _run_concurrent_scheduled_backups(prepared) + + assert first[0] == 23, first[1] + first[2] + assert second[0] == 1, second[1] + second[2] + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert "without newly validated success evidence" in second[1] + + +def test_run_sh_rejects_advanced_receipt_with_empty_artifact( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher( + tmp_path, + enable_daily_backup=True, + artifact_size=0, + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 1 + assert "without newly validated success evidence" in result.stdout + assert "backup artefact is missing or empty" in result.stderr + + +def test_run_sh_repairs_launcher_receipt_from_valid_auto_daily_sidecar( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + artifact = prepared["backup_root"] / "existing_auto_daily.zip" + artifact.parent.mkdir(parents=True) + artifact.write_bytes(b"existing validated backup") + sidecar = artifact.with_name(artifact.name + ".backup_receipt.json") + completed_at = datetime.now(timezone.utc) - timedelta(minutes=5) + _write_test_backup_receipt( + sidecar, + artifact_path=artifact, + completed_at=completed_at, + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + launcher_receipt = ( + prepared["root"] / ".run" / "last_successful_backup_receipt.json" + ) + legacy_sentinel = prepared["root"] / ".run" / "last_backup_utc.txt" + assert result.returncode == 0, result.stdout + result.stderr + assert not prepared["calls"].exists() + assert json.loads(launcher_receipt.read_text(encoding="utf-8"))[ + "final_artifact_path" + ] == str(artifact) + assert launcher_receipt.stat().st_mode & 0o777 == 0o600 + assert legacy_sentinel.stat().st_mode & 0o777 == 0o600 + assert "repairing from newest validated" in result.stderr + + +@pytest.mark.parametrize( + ("receipt_overrides", "warning_fragment"), + [ + ({"schema_version": "unknown.v1"}, "schema_version"), + ({"tag": "manual"}, "tag must be"), + ({"db_name": "other_db"}, "db_name must be"), + ], +) +def test_run_sh_ignores_untrusted_launcher_receipt_identity( + tmp_path: Path, + receipt_overrides: dict[str, str], + warning_fragment: str, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + artifact = prepared["backup_root"] / "untrusted.zip" + artifact.parent.mkdir(parents=True) + artifact.write_bytes(b"not authoritative") + launcher_receipt = ( + prepared["root"] / ".run" / "last_successful_backup_receipt.json" + ) + _write_test_backup_receipt( + launcher_receipt, + artifact_path=artifact, + completed_at=datetime.now(timezone.utc), + **receipt_overrides, + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert warning_fragment in result.stderr + + +def test_run_sh_ignores_receipt_for_artifact_outside_backup_root( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + prepared["backup_root"].mkdir(parents=True) + outside_artifact = tmp_path / "outside.zip" + outside_artifact.write_bytes(b"outside configured root") + launcher_receipt = ( + prepared["root"] / ".run" / "last_successful_backup_receipt.json" + ) + _write_test_backup_receipt( + launcher_receipt, + artifact_path=outside_artifact, + out_root=prepared["backup_root"], + completed_at=datetime.now(timezone.utc), + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert "direct child of the configured backup root" in result.stderr + + +def test_run_sh_ignores_receipt_that_names_backup_root_as_artifact( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + prepared["backup_root"].mkdir(parents=True) + (prepared["backup_root"] / "not-an-artifact.txt").write_text( + "not a backup artefact", + encoding="utf-8", + ) + launcher_receipt = ( + prepared["root"] / ".run" / "last_successful_backup_receipt.json" + ) + _write_test_backup_receipt( + launcher_receipt, + artifact_path=prepared["backup_root"], + out_root=prepared["backup_root"], + completed_at=datetime.now(timezone.utc), + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert "direct child of the configured backup root" in result.stderr + + +def test_run_sh_ignores_misnamed_artifact_sidecar(tmp_path: Path) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + artifact = prepared["backup_root"] / "existing.zip" + artifact.parent.mkdir(parents=True) + artifact.write_bytes(b"existing backup") + misnamed_sidecar = prepared["backup_root"] / "different.backup_receipt.json" + _write_test_backup_receipt( + misnamed_sidecar, + artifact_path=artifact, + completed_at=datetime.now(timezone.utc), + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert "sidecar name does not match" in result.stderr + + +def test_run_sh_future_receipt_cannot_suppress_backup(tmp_path: Path) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + artifact = prepared["backup_root"] / "future.zip" + artifact.parent.mkdir(parents=True) + artifact.write_bytes(b"future dated backup") + launcher_receipt = ( + prepared["root"] / ".run" / "last_successful_backup_receipt.json" + ) + _write_test_backup_receipt( + launcher_receipt, + artifact_path=artifact, + completed_at=datetime.now(timezone.utc) + timedelta(days=1), + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + assert "implausibly far in the future" in result.stderr + + +def test_run_sh_uses_legacy_sentinel_only_without_valid_receipt( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + sentinel = prepared["root"] / ".run" / "last_backup_utc.txt" + sentinel.parent.mkdir(parents=True) + sentinel.write_text( + datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + "\n", + encoding="utf-8", + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert not prepared["calls"].exists() + assert "falling back to legacy" in result.stderr + + +def test_run_sh_recovers_lock_missing_owner_metadata(tmp_path: Path) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + lock_dir = _write_test_backup_lock( + prepared, + owner_pid=os.getpid(), + owner_start="", + include_owner_metadata=False, + ) + + stale_result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert stale_result.returncode == 1 + assert "no valid owner metadata" in stale_result.stdout + assert not lock_dir.exists() + + retry_result = _run_prepared_launcher(prepared, "scheduled-backup") + assert retry_result.returncode == 0, retry_result.stdout + retry_result.stderr + assert prepared["calls"].read_text(encoding="utf-8").splitlines() == ["call"] + + +def test_run_sh_rejects_reused_pid_with_different_process_start( + tmp_path: Path, +) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + lock_dir = _write_test_backup_lock( + prepared, + owner_pid=os.getpid(), + owner_start="not-the-current-process-start", + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 1 + assert "without newly validated success evidence" in result.stdout + assert not lock_dir.exists() + assert not prepared["calls"].exists() + + +def test_run_sh_bounds_wait_for_live_backup_owner(tmp_path: Path) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=True) + env_path = prepared["root"] / ".env" + env_path.write_text( + env_path.read_text(encoding="utf-8") + + "VON_BACKUP_WAIT_TIMEOUT_SECONDS=1\n", + encoding="utf-8", + ) + lock_dir = _write_test_backup_lock( + prepared, + owner_pid=os.getpid(), + owner_start=_process_start_marker(os.getpid()), + ) + + result = _run_prepared_launcher(prepared, "scheduled-backup") + + assert result.returncode == 1 + assert "timed out after 1s" in result.stdout + assert lock_dir.exists() + assert not prepared["calls"].exists() + shutil.rmtree(lock_dir) + + +def test_run_sh_manual_backup_uses_owner_only_umask(tmp_path: Path) -> None: + prepared = _prepare_backup_launcher(tmp_path, enable_daily_backup=False) + + result = _run_prepared_launcher(prepared, "backup") + + assert result.returncode == 0, result.stdout + result.stderr + assert prepared["created"].exists() + assert prepared["created"].stat().st_mode & 0o777 == 0o600 + + +def test_run_sh_background_backup_delegates_to_detached_public_action() -> None: + payload = _run_bash_probe( + r""" +tmpdir="$(mktemp -d)" +detach_args="$tmpdir/detach-args.txt" +logs=() +log() { logs+=("$*"); } +python_cmd() { command -v python3; } +resolve_daily_backup_success() { return 0; } +launch_detached_process() { + printf '%s\n' "$*" > "$detach_args" + printf '4242' +} + +VON_ENABLE_DAILY_BACKUP=1 +unset VON_DISABLE_DAILY_BACKUP +RUN_DIR="$tmpdir/.run" +BACKUP_ROOT="$tmpdir/backups" +LOCAL_BACKUPS="$tmpdir/local" +ALLOW_REPO_BACKUP_OUTPUT=0 +mkdir -p "$RUN_DIR" + +if run_daily_backup_if_due; then status=0; else status=$?; fi +args="$(cat "$detach_args")" +LOGS="$(printf '%s\n' "${logs[@]}")" ARGS="$args" python3 - < None: + payload = _run_bash_probe( + r""" +tmpdir="$(mktemp -d)" +fake_pdm="$tmpdir/fake-pdm" +cat > "$fake_pdm" <<'SH' +#!/usr/bin/env bash +exit 0 +SH +chmod +x "$fake_pdm" + +semantic_calls=() +log() { :; } +pdm_cmd() { printf '%s' "$fake_pdm"; } +run_code_mention_scan() { semantic_calls+=("mention:$1"); } +run_code_predicate_sync() { semantic_calls+=("predicate:$1"); } + +VON_ENABLE_BACKUP_ACTION=1 +BACKUP_OUT_DIR="$tmpdir/backups" +BACKUP_ROOT="$BACKUP_OUT_DIR" +LOCAL_BACKUPS="$tmpdir/local" +BACKUP_TAG="test" +ALLOW_REPO_BACKUP_OUTPUT=0 +AGENT_TEST_INSTANCE=1 +NO_BACKUP_MIGRATE=1 + +BACKUP_DRY_RUN=1 +if run_backup; then dry_status=0; else dry_status=$?; fi +BACKUP_DRY_RUN=0 +if run_backup; then apply_status=0; else apply_status=$?; fi + +python3 - < None: payload = _run_bash_probe( r""" From 9e8871ff5d999dd442797c49e5242aafc0bcf08c Mon Sep 17 00:00:00 2001 From: witbrock Date: Tue, 28 Jul 2026 11:26:12 +0100 Subject: [PATCH 2/4] Keep Mongo backup credentials out of process arguments --- scripts/backup_von_db.py | 48 ++++++++++++++++++++++--------------- tests/test_backup_von_db.py | 24 ++++++++++++++++++- 2 files changed, 52 insertions(+), 20 deletions(-) diff --git a/scripts/backup_von_db.py b/scripts/backup_von_db.py index 5c41e3076..af032a81b 100644 --- a/scripts/backup_von_db.py +++ b/scripts/backup_von_db.py @@ -23,6 +23,7 @@ import os import subprocess import sys +import tempfile from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path @@ -368,31 +369,40 @@ def _upload_backup_to_blob( def _run_mongodump(*, mongo_uri: str, db_name: str, out_path: Path) -> None: - cmd = [ - "mongodump", - "--uri", - mongo_uri, - "--db", - db_name, - "--out", - str(out_path), - ] timeout_seconds = _env_int( os.environ.get("VON_BACKUP_MONGODUMP_TIMEOUT_SECONDS"), default=DEFAULT_MONGODUMP_TIMEOUT_SECONDS, ) if timeout_seconds < 1: timeout_seconds = DEFAULT_MONGODUMP_TIMEOUT_SECONDS - try: - subprocess.run(cmd, check=True, timeout=timeout_seconds) - except subprocess.TimeoutExpired: - raise RuntimeError( - f"mongodump exceeded the configured {timeout_seconds}s timeout" - ) from None - except subprocess.CalledProcessError as exc: - raise RuntimeError( - f"mongodump failed with exit code {exc.returncode}" - ) from None + with tempfile.TemporaryDirectory(prefix="von-mongodump-") as config_dir: + config_root = Path(config_dir) + os.chmod(config_root, 0o700) + config_path = config_root / "config.yml" + config_path.write_text( + f"uri: {json.dumps(mongo_uri)}\n", + encoding="utf-8", + ) + os.chmod(config_path, 0o600) + cmd = [ + "mongodump", + "--config", + str(config_path), + "--db", + db_name, + "--out", + str(out_path), + ] + try: + subprocess.run(cmd, check=True, timeout=timeout_seconds) + except subprocess.TimeoutExpired: + raise RuntimeError( + f"mongodump exceeded the configured {timeout_seconds}s timeout" + ) from None + except subprocess.CalledProcessError as exc: + raise RuntimeError( + f"mongodump failed with exit code {exc.returncode}" + ) from None def main(argv: list[str] | None = None) -> int: diff --git a/tests/test_backup_von_db.py b/tests/test_backup_von_db.py index 6ba45fb72..fb4afd4ee 100644 --- a/tests/test_backup_von_db.py +++ b/tests/test_backup_von_db.py @@ -53,7 +53,15 @@ def _fake_run( check: bool, timeout: int, ) -> None: - observed.update(command=command, check=check, timeout=timeout) + config_path = Path(command[command.index("--config") + 1]) + observed.update( + command=command, + check=check, + timeout=timeout, + config_path=config_path, + config_mode=config_path.stat().st_mode & 0o777, + config_text=config_path.read_text(encoding="utf-8"), + ) monkeypatch.setattr(backup_von_db.subprocess, "run", _fake_run) @@ -65,6 +73,11 @@ def _fake_run( assert observed["check"] is True assert observed["timeout"] == expected_timeout + assert "--uri" not in observed["command"] + assert "mongodb://example.invalid/" not in observed["command"] + assert observed["config_mode"] == 0o600 + assert "mongodb://example.invalid/" in observed["config_text"] + assert not Path(observed["config_path"]).exists() @pytest.mark.parametrize( @@ -81,6 +94,7 @@ def test_mongodump_failure_traceback_does_not_expose_uri_credentials( expected_message: str, ) -> None: mongo_uri = "mongodb://backup-user:super-secret-password@example.invalid/von_db" + observed_config_paths: list[Path] = [] def _fake_run( command: list[str], @@ -88,6 +102,12 @@ def _fake_run( check: bool, timeout: int, ) -> None: + assert "--uri" not in command + assert mongo_uri not in command + config_path = Path(command[command.index("--config") + 1]) + observed_config_paths.append(config_path) + assert config_path.stat().st_mode & 0o777 == 0o600 + assert mongo_uri in config_path.read_text(encoding="utf-8") if failure_kind == "timeout": raise backup_von_db.subprocess.TimeoutExpired(command, timeout) raise backup_von_db.subprocess.CalledProcessError(19, command) @@ -111,6 +131,8 @@ def _fake_run( assert expected_message in rendered assert mongo_uri not in rendered assert "super-secret-password" not in rendered + assert observed_config_paths + assert not observed_config_paths[0].exists() def test_retention_deletes_old_backups(tmp_path: Path) -> None: From 7503f7e10204ac779decd5e9cbfdfdebffc5a7c2 Mon Sep 17 00:00:00 2001 From: witbrock Date: Tue, 28 Jul 2026 11:32:34 +0100 Subject: [PATCH 3/4] Keep Mongo restore credentials out of process arguments --- scripts/restore_von_db.py | 66 +++++++++++++++----- tests/test_restore_von_db.py | 116 +++++++++++++++++++++++++++++++++++ 2 files changed, 167 insertions(+), 15 deletions(-) diff --git a/scripts/restore_von_db.py b/scripts/restore_von_db.py index 40fe1d2f3..55ef57447 100644 --- a/scripts/restore_von_db.py +++ b/scripts/restore_von_db.py @@ -33,6 +33,18 @@ from backup_von_db import BACKUP_RECEIPT_SIDECAR_SUFFIX +DEFAULT_MONGORESTORE_TIMEOUT_SECONDS = 6 * 60 * 60 + + +def _env_int(value: str | None, *, default: int) -> int: + if value is None: + return default + try: + return int(value.strip()) + except Exception: + return default + + def _redact_mongo_uri(uri: str) -> str: if not uri or "@" not in uri: return uri @@ -258,21 +270,45 @@ def _run_mongorestore( target_db_name: str, drop_target: bool, ) -> None: - cmd = [ - "mongorestore", - "--uri", - mongo_uri, - "--dir", - str(dump_root), - "--nsFrom", - f"{source_db_name}.*", - "--nsTo", - f"{target_db_name}.*", - "--stopOnError", - ] - if drop_target: - cmd.append("--drop") - subprocess.run(cmd, check=True) + timeout_seconds = _env_int( + os.environ.get("VON_BACKUP_MONGORESTORE_TIMEOUT_SECONDS"), + default=DEFAULT_MONGORESTORE_TIMEOUT_SECONDS, + ) + if timeout_seconds < 1: + timeout_seconds = DEFAULT_MONGORESTORE_TIMEOUT_SECONDS + with tempfile.TemporaryDirectory(prefix="von-mongorestore-") as config_dir: + config_root = Path(config_dir) + os.chmod(config_root, 0o700) + config_path = config_root / "config.yml" + config_path.write_text( + f"uri: {json.dumps(mongo_uri)}\n", + encoding="utf-8", + ) + os.chmod(config_path, 0o600) + cmd = [ + "mongorestore", + "--config", + str(config_path), + "--dir", + str(dump_root), + "--nsFrom", + f"{source_db_name}.*", + "--nsTo", + f"{target_db_name}.*", + "--stopOnError", + ] + if drop_target: + cmd.append("--drop") + try: + subprocess.run(cmd, check=True, timeout=timeout_seconds) + except subprocess.TimeoutExpired: + raise RuntimeError( + f"mongorestore exceeded the configured {timeout_seconds}s timeout" + ) from None + except subprocess.CalledProcessError as exc: + raise RuntimeError( + f"mongorestore failed with exit code {exc.returncode}" + ) from None def main(argv: list[str] | None = None) -> int: diff --git a/tests/test_restore_von_db.py b/tests/test_restore_von_db.py index 07c3363a1..a5c57d32c 100644 --- a/tests/test_restore_von_db.py +++ b/tests/test_restore_von_db.py @@ -1,9 +1,11 @@ from __future__ import annotations import json +import traceback import zipfile from pathlib import Path +import pytest from cryptography.fernet import Fernet from scripts import restore_von_db @@ -102,6 +104,120 @@ def _fake_run_mongorestore( assert str(captured["mongo_uri"]).startswith("mongodb://") +@pytest.mark.parametrize( + ("configured_timeout", "expected_timeout"), + [ + (None, restore_von_db.DEFAULT_MONGORESTORE_TIMEOUT_SECONDS), + ("90", 90), + ("0", restore_von_db.DEFAULT_MONGORESTORE_TIMEOUT_SECONDS), + ("invalid", restore_von_db.DEFAULT_MONGORESTORE_TIMEOUT_SECONDS), + ], +) +def test_mongorestore_uses_private_config_and_bounded_timeout( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + configured_timeout: str | None, + expected_timeout: int, +) -> None: + if configured_timeout is None: + monkeypatch.delenv( + "VON_BACKUP_MONGORESTORE_TIMEOUT_SECONDS", + raising=False, + ) + else: + monkeypatch.setenv( + "VON_BACKUP_MONGORESTORE_TIMEOUT_SECONDS", + configured_timeout, + ) + mongo_uri = "mongodb://restore-user:restore-secret@example.invalid/von_db" + observed: dict[str, object] = {} + + def _fake_run( + command: list[str], + *, + check: bool, + timeout: int, + ) -> None: + config_path = Path(command[command.index("--config") + 1]) + observed.update( + command=command, + check=check, + timeout=timeout, + config_path=config_path, + config_mode=config_path.stat().st_mode & 0o777, + config_text=config_path.read_text(encoding="utf-8"), + ) + + monkeypatch.setattr(restore_von_db.subprocess, "run", _fake_run) + + restore_von_db._run_mongorestore( + mongo_uri=mongo_uri, + dump_root=tmp_path, + source_db_name="von_db", + target_db_name="von_db_restore_probe", + drop_target=True, + ) + + assert observed["check"] is True + assert observed["timeout"] == expected_timeout + assert "--uri" not in observed["command"] + assert mongo_uri not in observed["command"] + assert "--drop" in observed["command"] + assert observed["config_mode"] == 0o600 + assert mongo_uri in observed["config_text"] + assert not Path(observed["config_path"]).exists() + + +@pytest.mark.parametrize( + ("failure_kind", "expected_message"), + [ + ("timeout", "exceeded the configured"), + ("nonzero", "failed with exit code 19"), + ], +) +def test_mongorestore_failure_traceback_does_not_expose_uri_credentials( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, + expected_message: str, +) -> None: + mongo_uri = "mongodb://restore-user:restore-secret@example.invalid/von_db" + + def _fake_run( + command: list[str], + *, + check: bool, + timeout: int, + ) -> None: + assert "--uri" not in command + assert mongo_uri not in command + if failure_kind == "timeout": + raise restore_von_db.subprocess.TimeoutExpired(command, timeout) + raise restore_von_db.subprocess.CalledProcessError(19, command) + + monkeypatch.setattr(restore_von_db.subprocess, "run", _fake_run) + + with pytest.raises(RuntimeError) as caught: + restore_von_db._run_mongorestore( + mongo_uri=mongo_uri, + dump_root=tmp_path, + source_db_name="von_db", + target_db_name="von_db_restore_probe", + drop_target=False, + ) + + rendered = "".join( + traceback.format_exception( + caught.type, + caught.value, + caught.tb, + ) + ) + assert expected_message in rendered + assert mongo_uri not in rendered + assert "restore-secret" not in rendered + + def test_dry_run_from_encrypted_zip_uses_fernet_key( tmp_path: Path, monkeypatch ) -> None: From d0d5a139877820c8c1cc0541b8cb6a483149eb75 Mon Sep 17 00:00:00 2001 From: witbrock Date: Tue, 28 Jul 2026 11:23:24 +0100 Subject: [PATCH 4/4] JVNAUTOSCI-2608 restore represented workflow affordances --- .../integrations/internal_mcp/catalogue.py | 329 +++++++-- src/backend/server/routes/von_routes.py | 1 + src/backend/services/adaptive_turn_service.py | 665 +++++++++++++---- .../workflow_turn_capability_service.py | 667 ++++++++++++++++++ tests/backend/test_adaptive_turn_service.py | 274 +++++++ ...p_add_relationship_predicate_validation.py | 218 +++++- .../test_von_generate_workflow_instances.py | 22 + .../test_workflow_turn_capability_service.py | 243 +++++++ 8 files changed, 2252 insertions(+), 167 deletions(-) create mode 100644 src/backend/services/workflow_turn_capability_service.py create mode 100644 tests/backend/test_workflow_turn_capability_service.py diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 01a02a2a4..a5bab45ec 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -2610,8 +2610,11 @@ def _add_relationship(**kwargs): source_id = kwargs.get("source_id") predicate = kwargs.get("predicate") + predicate_ref = kwargs.get("predicate_ref") target = kwargs.get("target") predicate_if_missing = kwargs.get("predicate_if_missing") + predicate_resolution: dict[str, Any] | None = None + predicate_value_kind = "concept" mutation_dispatched = False predicate_dependency: dict[str, Any] | None = None predicate_dependency_changed = False @@ -2625,6 +2628,83 @@ def _add_relationship(**kwargs): else None ) + if predicate_ref is not None: + if isinstance(predicate, str) and predicate.strip(): + return make_error_response( + "conflicting_predicate_reference", + "Provide predicate or predicate_ref, not both.", + details={"conflicting_fields": ["predicate", "predicate_ref"]}, + ) + if not isinstance(predicate_ref, Mapping): + return make_error_response( + "invalid_predicate_reference", + "predicate_ref must be an object.", + details={"value_type": type(predicate_ref).__name__}, + ) + reference_concept_id = predicate_ref.get("concept_id") + reference_name = predicate_ref.get("name") + has_concept_id = bool( + isinstance(reference_concept_id, str) + and reference_concept_id.strip() + ) + has_name = bool( + isinstance(reference_name, str) and reference_name.strip() + ) + if has_concept_id == has_name: + return make_error_response( + "invalid_predicate_reference", + ( + "predicate_ref must provide exactly one of concept_id or " + "name." + ), + details={ + "required_choice": ["concept_id", "name"], + }, + ) + on_missing = str(predicate_ref.get("on_missing") or "fail").strip().lower() + if on_missing not in {"fail", "create_typed_predicate"}: + return make_error_response( + "invalid_predicate_reference", + ( + "predicate_ref.on_missing must be 'fail' or " + "'create_typed_predicate'." + ), + details={"on_missing": on_missing}, + ) + if has_concept_id and on_missing == "create_typed_predicate": + return make_error_response( + "invalid_predicate_reference", + ( + "Creating a missing predicate requires predicate_ref.name; " + "an exact concept_id reference is resolve-only." + ), + details={"on_missing": on_missing}, + ) + predicate_value_kind = str( + predicate_ref.get("value_kind") or "concept" + ).strip().lower() + if predicate_value_kind not in {"concept", "text"}: + return make_error_response( + "invalid_predicate_reference", + "predicate_ref.value_kind must be 'concept' or 'text'.", + details={"value_kind": predicate_value_kind}, + ) + if has_concept_id: + predicate = str(reference_concept_id).strip() + predicate_resolution = { + "status": "explicit_concept_id", + "requested": predicate, + "resolved_concept_id": predicate, + } + else: + predicate = str(reference_name).strip() + if on_missing == "create_typed_predicate": + predicate_if_missing = { + "name": predicate, + "description": predicate_ref.get("description"), + "value_kind": predicate_value_kind, + } + def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: try: raise_if_internal_mcp_cancelled() @@ -2686,10 +2766,11 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: if not predicate: return make_error_response( "missing_parameter", - "Missing 'predicate' parameter", - details={"missing": ["predicate"]}, + "Missing predicate parameter or predicate_ref.", + details={"missing": ["predicate_or_predicate_ref"]}, suggestions=[ - "Provide a predicate like 'instance_of', 'typeOf', or a concept ID like '#V#hasAffiliation'" + "Provide predicate, or a predicate_ref with exactly one of " + "concept_id or name." ], ) if not target: @@ -2718,7 +2799,7 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: details={ "value_type": type(predicate_if_missing).__name__, "required_fields": ["name"], - "optional_fields": ["description"], + "optional_fields": ["description", "value_kind"], }, ) raw_dependency_name = predicate_if_missing.get("name") @@ -2745,12 +2826,22 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: predicate_dependency_description = ( raw_dependency_description.strip() or None ) + predicate_value_kind = str( + predicate_if_missing.get("value_kind") or predicate_value_kind + ).strip().lower() + if predicate_value_kind not in {"concept", "text"}: + return make_error_response( + "invalid_predicate_if_missing", + "predicate_if_missing.value_kind must be 'concept' or 'text'.", + details={"value_kind": predicate_value_kind}, + ) try: repo = ConceptsRepository from ...services.relationship_write_service import ( add_relationship, + is_structural_predicate, normalise_structural_predicate, validate_predicate_concept, ) @@ -2781,6 +2872,133 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: predicate_str[3:] if predicate_str.startswith("#V#") else predicate_str ) well_known_text_predicates = {"hasContent", "hasDescription", "hasName"} + if ( + not predicate_str.startswith("#V#") + and predicate_normalised not in well_known_text_predicates + and not is_structural_predicate(predicate_str) + ): + from ...services.concept_resolution_service import ( + resolve_concept_by_name, + ) + + resolution_payload = resolve_concept_by_name( + name=predicate_str, + instance_of="#V#predicate", + match_code_strings=False, + max_results=5, + ) + resolution_status = str( + resolution_payload.get("status") or "not_found" + ).strip() + resolved_predicate_id = resolution_payload.get( + "resolved_concept_id" + ) + predicate_resolution = { + "status": resolution_status, + "requested": predicate_str, + "resolved_concept_id": resolved_predicate_id, + "match": resolution_payload.get("match"), + "candidates": list( + resolution_payload.get("candidates") or [] + )[:5], + } + if ( + resolution_status == "resolved" + and isinstance(resolved_predicate_id, str) + and resolved_predicate_id.startswith("#V#") + ): + predicate_str = resolved_predicate_id + predicate_normalised = predicate_str[3:] + elif resolution_status == "ambiguous": + candidates = list( + resolution_payload.get("candidates") or [] + )[:5] + response = make_error_response( + "predicate_reference_ambiguous", + ( + "The predicate name identifies more than one accessible " + "predicate. Select one exact concept ID." + ), + details={ + "predicate": predicate_str, + "candidates": candidates, + }, + ) + response.update( + { + "effect_status": "not_started", + "changed": False, + "mutation_outcome": "not_started", + "predicate_resolution": predicate_resolution, + "recovery_affordances": [ + { + "action_type": "retry_with_predicate_concept_id", + "predicate_ref": { + "concept_id": candidate.get("concept_id") + }, + } + for candidate in candidates + if isinstance(candidate, Mapping) + and isinstance(candidate.get("concept_id"), str) + ], + } + ) + return response + elif predicate_dependency_name is not None: + canonical_new_predicate_id = canonicalise_vontology_concept_id( + predicate_dependency_name + ) + if canonical_new_predicate_id is None: + return make_error_response( + "predicate_reference_not_canonicalisable", + ( + "The requested predicate name cannot form a canonical " + "Vontology concept ID." + ), + details={"predicate": predicate_dependency_name}, + ) + predicate_str = canonical_new_predicate_id + predicate_normalised = predicate_str[3:] + predicate_resolution.update( + { + "status": "not_found_create_requested", + "resolved_concept_id": predicate_str, + } + ) + else: + response = make_error_response( + "predicate_reference_not_found", + ( + "No accessible predicate matched that name. The " + "relationship was not started." + ), + details={"predicate": predicate_str}, + ) + response.update( + { + "effect_status": "not_started", + "changed": False, + "mutation_outcome": "not_started", + "predicate_resolution": predicate_resolution, + "recovery_affordances": [ + { + "action_type": "retry_with_predicate_concept_id", + "predicate_ref": { + "concept_id": "#V#exact_predicate_id" + }, + }, + { + "action_type": "create_typed_predicate_then_retry", + "predicate_ref": { + "name": predicate_str, + "on_missing": "create_typed_predicate", + "value_kind": "concept", + }, + }, + ], + } + ) + return response if predicate_dependency_name is not None: canonical_dependency_id = canonicalise_vontology_concept_id( predicate_dependency_name @@ -2835,30 +3053,31 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: related_concept_ids=[predicate_str], ) - target_id = str(target).strip() - target_doc = repo.find_one( - {"concept_id": target_id}, - {"concept_id": 1}, - ) - if not target_doc: - return make_error_response( - "target_concept_not_found", - ( - f"Target concept '{target_id}' was not found. The " - "predicate dependency and relationship were not " - "created." - ), - details={ - "role": "target", - "concept_id": target_id, - "predicate": predicate_str, - }, - suggestions=[ - "Create or resolve the target concept before retrying " - "the relationship." - ], - related_concept_ids=[target_id], + if predicate_value_kind == "concept": + target_id = str(target).strip() + target_doc = repo.find_one( + {"concept_id": target_id}, + {"concept_id": 1}, ) + if not target_doc: + return make_error_response( + "target_concept_not_found", + ( + f"Target concept '{target_id}' was not found. The " + "predicate dependency and relationship were not " + "created." + ), + details={ + "role": "target", + "concept_id": target_id, + "predicate": predicate_str, + }, + suggestions=[ + "Create or resolve the target concept before retrying " + "the relationship." + ], + related_concept_ids=[target_id], + ) predicate_concept: dict[str, Any] = { "name": predicate_dependency_name, @@ -2870,7 +3089,11 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: mutation_dispatched = True try: creation_result = _create_concepts( - parent_id="#V#predicate", + parent_id=( + "#V#binary_text_predicate" + if predicate_value_kind == "text" + else "#V#predicate" + ), concepts=[predicate_concept], duplicate_resolution_mode=( _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY @@ -3047,6 +3270,11 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: if predicate_dependency is not None else {} ), + **( + {"predicate_resolution": predicate_resolution} + if predicate_resolution is not None + else {} + ), } # Determine if this is a text predicate (binary_text_predicate instance) @@ -3092,6 +3320,11 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: if predicate_dependency is not None else {} ), + **( + {"predicate_resolution": predicate_resolution} + if predicate_resolution is not None + else {} + ), } # Concept-to-concept relationships use the single authoritative pathway. @@ -3122,6 +3355,8 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: ) if predicate_dependency is not None: response["predicate_dependency"] = predicate_dependency + if predicate_resolution is not None: + response["predicate_resolution"] = predicate_resolution if ( predicate_dependency_changed or predicate_dependency_partial_failures @@ -3176,6 +3411,8 @@ def _cancellation_after_predicate_dependency() -> dict[str, Any] | None: response["changed"] = bool(response["added"] or predicate_dependency_changed) if predicate_dependency is not None: response["predicate_dependency"] = predicate_dependency + if predicate_resolution is not None: + response["predicate_resolution"] = predicate_resolution if "inverse_predicate" in result or "inverse_modified" in result: response["inverse"] = { "predicate": result.get("inverse_predicate"), @@ -9038,21 +9275,24 @@ def _add_relationship_input_schema() -> Schema: return Schema( required={ "source_id": str, - "predicate": str, "target": str, }, optional={ + "predicate": (str, type(None)), + "predicate_ref": (dict, type(None)), "predicate_if_missing": (dict, type(None)), "namespace": (str, type(None)), }, allow_unknown=True, description=( - "add_relationship input: source_id (str), predicate (str, structural " - "relationship name or exact #V# predicate concept ID), target (str). " - "For a missing dynamic predicate only, predicate_if_missing may be " - "{name: str, description?: str}; its name must canonically identify " - "the exact requested predicate. The capability creates or reuses and " - "verifies that predicate before attempting the edge." + "add_relationship input: source_id and target plus exactly one useful " + "predicate reference. predicate accepts a structural relationship, " + "natural-language predicate name, or exact #V# predicate ID. " + "predicate_ref is the typed form: {concept_id} or {name, " + "on_missing?: 'fail'|'create_typed_predicate', value_kind?: " + "'concept'|'text', description?}. Legacy predicate_if_missing remains " + "supported for exact dynamic predicates. Creation is attempted only " + "when explicitly requested, then verified before the edge." ), ) @@ -9074,6 +9314,8 @@ def _add_relationship_output_schema() -> Schema: "changed": (bool, type(None)), "partial_failures": (list, type(None)), "predicate_dependency": (dict, type(None)), + "predicate_resolution": (dict, type(None)), + "recovery_affordances": (list, type(None)), "text_value_id": (str, type(None)), "relation_id": (str, type(None)), "error": (str, type(None)), @@ -32280,11 +32522,12 @@ def _build_default_catalogue_core_definitions() -> List[MethodDefinition]: ordinary_turn_mutation_subject_argument="source_id", description=( "Add one concept-to-concept or concept-to-text relationship. " - "Structural predicates use their existing aliases; a custom " - "predicate uses its exact #V# concept ID. If that exact custom " - "predicate is missing, predicate_if_missing={name, description?} " - "can canonically create or reuse and verify it before this " - "relationship attempt." + "Structural predicates use their aliases. A custom predicate may " + "be supplied by exact #V# ID or resolved natural-language name. " + "Use predicate_ref for typed ambiguity/not-found recovery and " + "explicitly request creation with on_missing=" + "'create_typed_predicate' plus value_kind='concept' or 'text'. " + "Creation/reuse is verified before the relationship attempt." ), ), MethodDefinition( @@ -36291,8 +36534,8 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin input_schema=Schema( required={"workflow_id": str}, optional={ - "user_id": str, - "org_id": str, + "user_id": (str, type(None)), + "org_id": (str, type(None)), "namespace": (str, type(None)), "inputs": (dict, type(None)), "max_retries": int, @@ -36328,6 +36571,8 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin description="Structured awaited workflow execution result and telemetry.", ), category="write", + timeout_sec=120.0, + effect_admission_window_sec=5.0, description=( "Create a durable workflow instance via the verified submission pathway and optionally " "wait for a bounded terminal result with structured telemetry and optional trace retrieval." diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index 33b7b08b4..09a22f279 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -10975,6 +10975,7 @@ def _progress_update(info: dict[str, Any]) -> None: if request_gmail_profile else None ), + workflow_launch_inputs=request_workflow_launch_inputs, progress_tracker=progress_tracker, turn_id=request_id, ) diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py index a8fcf2642..3daf962f6 100644 --- a/src/backend/services/adaptive_turn_service.py +++ b/src/backend/services/adaptive_turn_service.py @@ -84,6 +84,37 @@ class AdaptiveTurnResult: effect_finality_fallback: bool = False +@dataclass(frozen=True) +class _PreparedCapabilityCall: + """One model-requested capability after trusted argument binding.""" + + index: int + call: ToolCall + capability_name: str + execution_method_name: str + arguments: Mapping[str, Any] + is_effect: bool + minimum_effect_window_seconds: float + capability_kind: str = "registered_tool" + represented_workflow_id: str | None = None + binding_diagnostics: Mapping[str, Any] | None = None + + +@dataclass(frozen=True) +class _ContainedCapabilityResult: + """A capability result retained until request-thread evidence commit.""" + + raw_payload: Any + transport_result: Any + arguments: Mapping[str, Any] + capability_name: str + is_effect: bool + execution_method_name: str + capability_kind: str = "registered_tool" + represented_workflow_id: str | None = None + binding_diagnostics: Mapping[str, Any] | None = None + + def _positive_float_env(name: str, default: float) -> float: try: value = float(os.getenv(name, str(default))) @@ -1022,8 +1053,10 @@ def _tool_definitions() -> list[ToolDefinition]: description=( "Inspect the capabilities delegated to this turn. Search by " "ordinary words, request exact names, or page through the complete " - "catalogue. Returns canonical argument schemas. This is discovery, " - "not a requirement to use any particular capability." + "catalogue. A natural-language query also searches actor-accessible " + "represented workflows and returns executable matches as bound " + "capabilities. Returns canonical argument schemas. This is " + "discovery, not a requirement to use any particular capability." ), input_schema={ "type": "object", @@ -1044,7 +1077,9 @@ def _tool_definitions() -> list[ToolDefinition]: f"{_CAPABILITY_TOOL_NAME}. You may also invoke a known capability " "directly without first searching. Reads return provenance-bearing " "evidence. Effects return a server-generated receipt and evidence " - "handle; inspect canonical state before claiming persistence." + "handle; represented-workflow capability identities and actor scope " + "are bound by the server. Inspect canonical state before claiming " + "persistence." ), input_schema={ "type": "object", @@ -1135,7 +1170,9 @@ def _scope_message( "- Remaining effect-capable window before the protected final-answer " f"reserve: {max(0.0, remaining_effect_capable_seconds):.3f} seconds.\n" f"- {_CAPABILITY_TOOL_NAME} exposes the complete delegated catalogue " - "without interpreting the user's intent.\n" + "without interpreting the user's intent; a natural-language query also " + "retrieves actor-accessible executable represented workflows as bound " + "capabilities.\n" f"- {_INVOKE_TOOL_NAME} invokes any named delegated capability.\n" f"- {_EVIDENCE_INDEX_TOOL_NAME} pages every evidence handle recorded " "for this turn.\n" @@ -1256,6 +1293,9 @@ def _capability_catalogue( gateway: InternalMCPGateway, delegated_names: Sequence[str], payload: Mapping[str, Any], + *, + workflow_capabilities: Sequence[Any] = (), + workflow_discovery: Mapping[str, Any] | None = None, ) -> dict[str, Any]: query = str(payload.get("query") or "").strip().lower() requested_names = payload.get("names") @@ -1356,6 +1396,7 @@ def _capability_catalogue( ) capability = { "name": name, + "kind": "registered_tool", "description": description, "input_schema": _model_visible_input_schema(definition), "query_match": query_match, @@ -1387,6 +1428,68 @@ def _capability_catalogue( capability, ) ) + represented_workflow_total = 0 + for workflow_capability in workflow_capabilities: + to_catalogue_entry = getattr( + workflow_capability, + "to_catalogue_entry", + None, + ) + if not callable(to_catalogue_entry): + continue + capability = to_catalogue_entry() + if not isinstance(capability, Mapping): + continue + capability = dict(capability) + name = str(capability.get("name") or "").strip() + if not name: + continue + represented_workflow_total += 1 + name_key = name.lower() + if exact_names and name_key not in exact_names: + continue + description = str(capability.get("description") or "").strip() + searchable = ( + f"{name_key.replace('_', ' ')} " + f"{str(capability.get('display_name') or '').lower()} " + f"{str(capability.get('workflow_id') or '').lower()} " + f"{description.lower()}" + ) + if query_tokens: + literal_matches = sum( + 1 for token in query_tokens if token in searchable + ) + query_match = bool( + literal_matches + or float(capability.get("relevance_score") or 0.0) > 0.0 + ) + semantic_score = int( + max( + 0.0, + min( + 1.0, + float(capability.get("relevance_score") or 0.0), + ), + ) + * 1000 + ) + score = semantic_score + (literal_matches * 10) + else: + query_match = True + score = int( + max( + 0.0, + min( + 1.0, + float(capability.get("relevance_score") or 0.0), + ), + ) + * 1000 + ) + capability["query_match"] = query_match + if query_match: + matched_total += 1 + ranked.append((-score, name_key, capability)) ranked.sort(key=lambda item: (item[0], item[1])) selected = [item[2] for item in ranked] page = selected[offset : offset + limit] @@ -1396,9 +1499,15 @@ def _capability_catalogue( "success": True, "delegation": "bounded_capabilities", "total": len(selected), - "delegated_total": registered_delegated_total, + "delegated_total": ( + registered_delegated_total + represented_workflow_total + ), + "registered_tool_total": registered_delegated_total, + "represented_workflow_total": represented_workflow_total, "matched_total": matched_total, - "ranking": "literal_query_match_then_name", + "ranking": ( + "represented_semantic_relevance_then_literal_query_match_then_name" + ), "catalogue_scope": ( "requested_exact_names" if exact_names @@ -1407,6 +1516,32 @@ def _capability_catalogue( "offset": offset, "next_offset": next_offset if next_offset < len(selected) else None, "capabilities": page, + **( + {"workflow_discovery": dict(workflow_discovery)} + if isinstance(workflow_discovery, Mapping) + else {} + ), + **( + { + "recovery_affordances": [ + { + "action_type": "query_represented_workflow_capabilities", + "tool": _CAPABILITY_TOOL_NAME, + "arguments": { + "query": ( + "Describe the work product or reusable workflow " + "capability needed." + ) + }, + } + ] + } + if exact_names.intersection( + {"workflow_create_instance", "workflow_execute"} + ) + and not page + else {} + ), } @@ -1621,6 +1756,22 @@ def _effect_id(*, turn_id: str | None, call_id: str, capability_name: str) -> st return f"effect_{hashlib.sha256(material).hexdigest()[:24]}" +def _effect_request_signature( + capability_name: str, + arguments: Mapping[str, Any], +) -> str: + """Identify an unchanged effect request independently of model call IDs.""" + + return hashlib.sha256( + _json_bytes( + { + "capability_name": str(capability_name).strip().lower(), + "arguments": dict(arguments), + } + ) + ).hexdigest() + + def _effect_result_target_ids(raw_payload: Any) -> list[str]: """Extract bounded concrete mutation targets from a handler receipt.""" @@ -1798,6 +1949,7 @@ def execute_adaptive_turn( user_concept_id: str | None = None, org_concept_id: str | None = None, trusted_argument_values: Mapping[str, Any] | None = None, + workflow_launch_inputs: Mapping[str, Any] | None = None, progress_tracker: Any = None, turn_id: str | None = None, turn_budget_seconds: float | None = None, @@ -1869,6 +2021,8 @@ def execute_adaptive_turn( trusted_argument_values=trusted_values, ) delegated_lookup = {name.lower(): name for name in delegated_names} + workflow_capabilities_by_name: dict[str, Any] = {} + latest_workflow_discovery: dict[str, Any] | None = None available_tools = _tool_definitions() final_synthesis_tools = [ tool @@ -1917,6 +2071,8 @@ def execute_adaptive_turn( effect_states: dict[str, dict[str, Any]] = {} effect_state_generation = 0 last_partial_effect_generation = 0 + successful_effect_mutation_generation = 0 + terminal_failed_effect_requests: dict[str, tuple[int, str | None]] = {} def remember_effect_state( effect_id: str, @@ -2417,7 +2573,10 @@ def enter_answer_only(reason: str) -> None: model=model, system_message=_scope_message( scope, - delegated_count=len(delegated_names), + delegated_count=( + len(delegated_names) + + len(workflow_capabilities_by_name) + ), final_synthesis=final_synthesis, answer_only=answer_only, remaining_effect_capable_seconds=( @@ -2707,13 +2866,8 @@ def enter_answer_only(reason: str) -> None: continuation = response.continuation batch_results: list[ToolResult | None] = [None] * len(calls) - raw_capability_results: dict[ - int, - tuple[Any, Any, dict[str, Any], str, bool], - ] = {} - actual_capabilities: list[ - tuple[int, ToolCall, str, dict[str, Any], bool, float] - ] = [] + raw_capability_results: dict[int, _ContainedCapabilityResult] = {} + actual_capabilities: list[_PreparedCapabilityCall] = [] for index, call in enumerate(calls): _check_cancellation(progress_tracker) @@ -2779,14 +2933,71 @@ def enter_answer_only(reason: str) -> None: ) continue if call.tool_name == _CAPABILITY_TOOL_NAME: - output = ( - _capability_catalogue(gateway, delegated_names, call.payload) - if gateway is not None - else _error_payload( + if gateway is not None: + workflow_query = str(call.payload.get("query") or "").strip() + if workflow_query: + from src.backend.services.workflow_turn_capability_service import ( + discover_turn_workflow_capabilities, + ) + + remaining_discovery_seconds = max( + 0.1, + min(5.0, research_deadline - clock()), + ) + try: + requested_workflow_limit = int( + call.payload.get("limit") or 5 + ) + except (TypeError, ValueError): + requested_workflow_limit = 5 + try: + discovered_workflows, workflow_discovery = ( + discover_turn_workflow_capabilities( + workflow_query, + namespace=scope.namespace, + user_concept_id=scope.user_concept_id, + organisation_concept_id=( + scope.organisation_concept_id + ), + turn_id=turn_id or "ordinary-turn", + max_results=min( + 10, + max(1, requested_workflow_limit), + ), + timeout_seconds=remaining_discovery_seconds, + ) + ) + except Exception as exc: # noqa: BLE001 + discovered_workflows = [] + workflow_discovery = { + "schema_version": ( + "workflow_turn_capability_discovery.v1" + ), + "status": "failed", + "query": workflow_query, + "match_count": 0, + "error_type": type(exc).__name__, + "error": str(exc)[:500], + } + for workflow_capability in discovered_workflows: + workflow_capabilities_by_name[ + workflow_capability.name.lower() + ] = workflow_capability + latest_workflow_discovery = dict(workflow_discovery) + output = _capability_catalogue( + gateway, + delegated_names, + call.payload, + workflow_capabilities=tuple( + workflow_capabilities_by_name.values() + ), + workflow_discovery=latest_workflow_discovery, + ) + else: + output = _error_payload( "capability_gateway_unavailable", "The capability gateway is unavailable.", ) - ) batch_results[index] = ToolResult( call_id=call.call_id, tool_name=call.tool_name, @@ -2869,8 +3080,11 @@ def enter_answer_only(reason: str) -> None: requested_name = str(call.payload.get("name") or "").strip() canonical_name = delegated_lookup.get(requested_name.lower()) + workflow_capability = workflow_capabilities_by_name.get( + requested_name.lower() + ) arguments = call.payload.get("arguments") - if canonical_name is None: + if canonical_name is None and workflow_capability is None: output = _error_payload( "capability_not_delegated", f"{requested_name!r} is not a delegated capability.", @@ -2893,73 +3107,210 @@ def enter_answer_only(reason: str) -> None: ) continue assert gateway is not None - definition = gateway.get_method_definition(canonical_name) + capability_name = ( + canonical_name + if canonical_name is not None + else str(workflow_capability.name) + ) + execution_method_name = ( + canonical_name + if canonical_name is not None + else "workflow_execute" + ) + definition = gateway.get_method_definition(execution_method_name) + is_workflow_capability = workflow_capability is not None is_effect = bool( - definition is not None - and definition.category == "write" - and definition.ordinary_turn_effect + is_workflow_capability + or ( + definition is not None + and definition.category == "write" + and definition.ordinary_turn_effect + ) ) if not isinstance(arguments, Mapping): - raw_capability_results[index] = ( - _error_payload( + raw_capability_results[index] = _ContainedCapabilityResult( + raw_payload=_error_payload( "invalid_capability_arguments", "arguments must be an object.", ), - None, - {}, - canonical_name, - is_effect, + transport_result=None, + arguments={}, + capability_name=capability_name, + is_effect=is_effect, + execution_method_name=execution_method_name, + capability_kind=( + "represented_workflow" + if is_workflow_capability + else "registered_tool" + ), + represented_workflow_id=( + str(workflow_capability.workflow_id) + if is_workflow_capability + else None + ), ) continue - trusted_arguments = _trusted_tool_payload( - gateway=gateway, - tool_name=canonical_name, - model_payload=arguments, - trusted_argument_values=trusted_values, - ) + binding_diagnostics: Mapping[str, Any] | None = None + if is_workflow_capability: + if definition is None: + raw_capability_results[index] = _ContainedCapabilityResult( + raw_payload=_error_payload( + "workflow_execution_capability_unavailable", + ( + "The represented workflow was discovered, but " + "the verified workflow execution method is not " + "registered." + ), + ), + transport_result=None, + arguments={}, + capability_name=capability_name, + is_effect=True, + execution_method_name=execution_method_name, + capability_kind="represented_workflow", + represented_workflow_id=str( + workflow_capability.workflow_id + ), + ) + continue + if not scope.user_concept_id or not scope.namespace: + raw_capability_results[index] = _ContainedCapabilityResult( + raw_payload=_error_payload( + "workflow_actor_authority_required", + ( + "Represented workflow invocation requires an " + "authenticated actor and namespace." + ), + ), + transport_result=None, + arguments={}, + capability_name=capability_name, + is_effect=True, + execution_method_name=execution_method_name, + capability_kind="represented_workflow", + represented_workflow_id=str( + workflow_capability.workflow_id + ), + ) + continue + from src.backend.services.workflow_turn_capability_service import ( + build_workflow_execution_arguments, + ) + + try: + ( + trusted_arguments, + binding_diagnostics, + ) = build_workflow_execution_arguments( + workflow_capability, + arguments, + prompt=prompt, + context=current_context, + request_workflow_launch_inputs=workflow_launch_inputs, + user_concept_id=scope.user_concept_id, + organisation_concept_id=scope.organisation_concept_id, + namespace=scope.namespace, + maximum_wait_seconds=max( + 0.1, + final_answer_deadline - clock() - 1.0, + ), + ) + except (TypeError, ValueError) as exc: + raw_capability_results[index] = _ContainedCapabilityResult( + raw_payload=_error_payload( + "invalid_workflow_capability_arguments", + str(exc), + ), + transport_result=None, + arguments={}, + capability_name=capability_name, + is_effect=True, + execution_method_name=execution_method_name, + capability_kind="represented_workflow", + represented_workflow_id=str( + workflow_capability.workflow_id + ), + ) + continue + else: + trusted_arguments = _trusted_tool_payload( + gateway=gateway, + tool_name=execution_method_name, + model_payload=arguments, + trusted_argument_values=trusted_values, + ) minimum_effect_window = 0.0 if is_effect: - resolved_effect_window = gateway.get_method_effect_admission_window_sec( - canonical_name + resolved_effect_window = ( + 5.0 + if is_workflow_capability + else gateway.get_method_effect_admission_window_sec( + execution_method_name + ) ) if resolved_effect_window is None: resolved_effect_window = gateway.get_method_timeout_sec( - canonical_name + execution_method_name ) minimum_effect_window = max( 0.0, float(resolved_effect_window or 0.0), ) actual_capabilities.append( - ( - index, - call, - canonical_name, - trusted_arguments, - is_effect, - minimum_effect_window, + _PreparedCapabilityCall( + index=index, + call=call, + capability_name=capability_name, + execution_method_name=execution_method_name, + arguments=trusted_arguments, + is_effect=is_effect, + minimum_effect_window_seconds=minimum_effect_window, + capability_kind=( + "represented_workflow" + if is_workflow_capability + else "registered_tool" + ), + represented_workflow_id=( + str(workflow_capability.workflow_id) + if is_workflow_capability + else None + ), + binding_diagnostics=binding_diagnostics, ) ) def invoke_and_contain( - item: tuple[int, ToolCall, str, dict[str, Any], bool, float], + item: _PreparedCapabilityCall, *, deadline_monotonic: float, effect_window_denial: Mapping[str, Any] | None = None, - ) -> tuple[ - int, - tuple[Any, Any, dict[str, Any], str, bool], - ]: - ( - index, - call, - canonical_name, - arguments, - is_effect, - _minimum_effect_window, - ) = item + ) -> tuple[int, _ContainedCapabilityResult]: + nonlocal successful_effect_mutation_generation + index = item.index + call = item.call + canonical_name = item.capability_name + execution_method_name = item.execution_method_name + arguments = dict(item.arguments) + is_effect = item.is_effect + + def contained( + raw_payload: Any, + transport_result: Any = None, + ) -> _ContainedCapabilityResult: + return _ContainedCapabilityResult( + raw_payload=raw_payload, + transport_result=transport_result, + arguments=arguments, + capability_name=canonical_name, + is_effect=is_effect, + execution_method_name=execution_method_name, + capability_kind=item.capability_kind, + represented_workflow_id=item.represented_workflow_id, + binding_diagnostics=item.binding_diagnostics, + ) + assert gateway is not None - definition = gateway.get_method_definition(canonical_name) + definition = gateway.get_method_definition(execution_method_name) subject_argument = ( definition.ordinary_turn_mutation_subject_argument if definition is not None and is_effect @@ -2971,29 +3322,61 @@ def invoke_and_contain( else None ) if effect_denial is not None: - return index, ( - effect_denial, - None, - arguments, - canonical_name, - is_effect, - ) + return index, contained(effect_denial) if subject_argument: if not _effect_subject_authorised( arguments.get(subject_argument), scope, ): - return index, ( + return index, contained( _error_payload( "effect_subject_not_authorised", "The effect subject is not scoped to the authenticated " "actor or organisation.", - ), - None, - arguments, - canonical_name, - is_effect, + ) ) + + effect_request_signature = ( + _effect_request_signature(canonical_name, arguments) + if is_effect + else None + ) + prior_terminal_failure = ( + terminal_failed_effect_requests.get(effect_request_signature) + if effect_request_signature is not None + else None + ) + if ( + prior_terminal_failure is not None + and prior_terminal_failure[0] + == successful_effect_mutation_generation + ): + return index, contained( + { + **_error_payload( + "effect_request_unchanged_after_terminal_failure", + ( + "This exact effect request was not repeated because " + "it already failed terminally and no successful " + "intervening effect changed the turn state." + ), + ), + "status": "not_started", + "mutation_outcome": "not_started", + "outcome_finality": "terminal_for_turn", + "prior_error_code": prior_terminal_failure[1], + "changed": False, + "recovery_affordances": [ + { + "action_type": "change_arguments_or_use_typed_recovery" + }, + { + "action_type": "inspect_canonical_state_before_retry" + }, + ], + } + ) + effect_identifier = ( _effect_id( turn_id=turn_id, @@ -3014,7 +3397,7 @@ def invoke_and_contain( }, ) if not _effect_phase_acknowledged(dispatch_outcome): - return index, ( + return index, contained( { **_error_payload( "effect_observation_unavailable", @@ -3039,11 +3422,7 @@ def invoke_and_contain( ) } ], - }, - None, - arguments, - canonical_name, - is_effect, + } ) if isinstance(effect_window_denial, Mapping): denial_payload = dict(effect_window_denial) @@ -3077,20 +3456,14 @@ def invoke_and_contain( ), } ) - return index, ( - denial_payload, - None, - arguments, - canonical_name, - is_effect, - ) + return index, contained(denial_payload) try: with override_current_actor( scope.user_concept_id, scope.organisation_concept_id, ): transport_result = gateway.invoke( - canonical_name, + execution_method_name, arguments, deadline_monotonic=deadline_monotonic, late_completion_observer=( @@ -3136,6 +3509,52 @@ def invoke_and_contain( if is_effect: raw_payload["mutation_outcome"] = "unknown" transport_result = None + + if item.capability_kind == "represented_workflow": + from src.backend.services.workflow_turn_capability_service import ( + normalise_workflow_effect_receipt, + ) + + workflow_capability = workflow_capabilities_by_name.get( + canonical_name.lower() + ) + if workflow_capability is not None: + raw_payload = normalise_workflow_effect_receipt( + raw_payload, + capability=workflow_capability, + ) + + terminal_effect_status = ( + _effect_status( + raw_payload, + transport_result=transport_result, + ) + if is_effect + else None + ) + if ( + is_effect + and effect_request_signature is not None + and terminal_effect_status in {"failed", "not_started"} + and isinstance(raw_payload, Mapping) + and raw_payload.get("retryable") is not True + ): + terminal_failed_effect_requests[effect_request_signature] = ( + successful_effect_mutation_generation, + ( + str(raw_payload.get("error_code")).strip() + if raw_payload.get("error_code") + else None + ), + ) + elif ( + is_effect + and terminal_effect_status in {"succeeded", "partial"} + and isinstance(raw_payload, Mapping) + and raw_payload.get("changed") is True + ): + successful_effect_mutation_generation += 1 + if effect_identifier is not None: transport_metadata_fn = getattr( transport_result, @@ -3147,10 +3566,7 @@ def invoke_and_contain( if callable(transport_metadata_fn) else {} ) - terminal_effect_status = _effect_status( - raw_payload, - transport_result=transport_result, - ) + assert terminal_effect_status is not None changed = ( raw_payload.get("changed") if isinstance(raw_payload, Mapping) @@ -3192,15 +3608,11 @@ def invoke_and_contain( ), } ) - return index, ( - raw_payload, - transport_result, - arguments, - canonical_name, - is_effect, - ) + return index, contained(raw_payload, transport_result) - if actual_capabilities and any(item[4] for item in actual_capabilities): + if actual_capabilities and any( + item.is_effect for item in actual_capabilities + ): # Preserve model-call order whenever the batch contains an effect. # Each effect receives an independent admission decision in model # order. One invalid or oversized call therefore cannot deny an @@ -3209,14 +3621,8 @@ def invoke_and_contain( # remain available for that inspection. prior_indeterminate_effect_id: str | None = None for item in actual_capabilities: - ( - _item_index, - _call, - canonical_name, - _arguments, - is_effect, - _minimum_effect_window, - ) = item + canonical_name = item.capability_name + is_effect = item.is_effect effect_window_denial = None if is_effect and prior_indeterminate_effect_id is not None: effect_window_denial = { @@ -3249,19 +3655,18 @@ def invoke_and_contain( effect_window_denial=effect_window_denial, ) raw_capability_results[result_index] = contained - raw_payload, transport_result, *_rest = contained if ( is_effect and effect_window_denial is None and _effect_status( - raw_payload, - transport_result=transport_result, + contained.raw_payload, + transport_result=contained.transport_result, ) == "indeterminate" ): prior_indeterminate_effect_id = _effect_id( turn_id=turn_id, - call_id=_call.call_id, + call_id=item.call.call_id, capability_name=canonical_name, ) elif actual_capabilities: @@ -3294,15 +3699,13 @@ def invoke_and_contain( # handler therefore cannot mutate the terminal transcript. for index, call in enumerate(calls): if index in raw_capability_results: - ( - raw_payload, - transport_result, - arguments, - canonical_name, - is_effect, - ) = ( - raw_capability_results[index] - ) + contained_result = raw_capability_results[index] + raw_payload = contained_result.raw_payload + transport_result = contained_result.transport_result + arguments = contained_result.arguments + canonical_name = contained_result.capability_name + is_effect = contained_result.is_effect + execution_method_name = contained_result.execution_method_name effect_identifier = ( _effect_id( turn_id=turn_id, @@ -3348,6 +3751,17 @@ def invoke_and_contain( "namespace": scope.namespace, "user_concept_id": scope.user_concept_id, "organisation_concept_id": scope.organisation_concept_id, + "capability_kind": contained_result.capability_kind, + "execution_method": execution_method_name, + **( + { + "represented_workflow_id": ( + contained_result.represented_workflow_id + ) + } + if contained_result.represented_workflow_id + else {} + ), "transport": transport_metadata, **( { @@ -3417,6 +3831,8 @@ def invoke_and_contain( invocation: dict[str, Any] = { "tool": canonical_name, "via": call.tool_name, + "capability_kind": contained_result.capability_kind, + "execution_method": execution_method_name, "call_id": call.call_id, "payload": dict(call.payload), "effective_arguments": arguments, @@ -3428,6 +3844,14 @@ def invoke_and_contain( else {} ), } + if contained_result.represented_workflow_id: + invocation["represented_workflow_id"] = ( + contained_result.represented_workflow_id + ) + if contained_result.binding_diagnostics: + invocation["binding_diagnostics"] = dict( + contained_result.binding_diagnostics + ) if is_effect: invocation.update( { @@ -3455,6 +3879,7 @@ def invoke_and_contain( "stage": "adaptive_research", "phase": "adaptive_research", "tool": canonical_name, + "execution_method": execution_method_name, "call_id": call.call_id, "success": status == "ok", "result_summary": ( diff --git a/src/backend/services/workflow_turn_capability_service.py b/src/backend/services/workflow_turn_capability_service.py new file mode 100644 index 000000000..c38534efd --- /dev/null +++ b/src/backend/services/workflow_turn_capability_service.py @@ -0,0 +1,667 @@ +"""Turn-scoped affordances for represented workflow discovery and execution. + +This service adapts the existing actor-filtered workflow capability index and +verified durable submission surface to the thin ordinary-turn capability +interface. It does not select a workflow or interpret request semantics. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field +from typing import Any, Mapping, Sequence + +from src.backend.security.access_control import override_current_actor + + +WORKFLOW_TURN_CAPABILITY_SCHEMA_VERSION = "workflow_turn_capability.v1" +WORKFLOW_TURN_DISCOVERY_SCHEMA_VERSION = "workflow_turn_capability_discovery.v1" +WORKFLOW_TURN_RECEIPT_SCHEMA_VERSION = "workflow_turn_effect_receipt.v1" + +_SERVER_PROVIDED_WORKFLOW_INPUT_KEYS = frozenset( + { + "actor_concept_id", + "actor_user_concept_id", + "augmented_context", + "namespace", + "org_concept_id", + "organisation_concept_id", + "prompt", + "workflow_id", + "user_id", + "org_id", + "user_concept_id", + "user_namespace", + "user_prompt", + "workflow_launch_inputs", + } +) +_MODEL_WORKFLOW_CONTROL_ARGUMENTS = frozenset( + { + "await_terminal", + "include_step_result_envelopes", + "include_trace", + "inputs", + "max_retries", + "poll_interval_seconds", + "timeout_seconds", + } +) +_TERMINAL_SUCCESS_STATUSES = frozenset({"completed", "succeeded", "success"}) +_TERMINAL_FAILURE_STATUSES = frozenset( + {"cancelled", "canceled", "failed", "rejected", "terminated"} +) + + +def _normalise_non_empty_text(value: Any) -> str | None: + if not isinstance(value, str): + return None + text = value.strip() + return text or None + + +def _coerce_bool(value: Any, *, default: bool) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + normalised = value.strip().lower() + if normalised in {"1", "true", "yes", "on"}: + return True + if normalised in {"0", "false", "no", "off", ""}: + return False + return default + + +def _bounded_mapping_sequence( + value: Any, + *, + limit: int, +) -> list[dict[str, Any]]: + if not isinstance(value, Sequence) or isinstance( + value, + (str, bytes, bytearray), + ): + return [] + return [ + dict(item) + for item in list(value)[:limit] + if isinstance(item, Mapping) + ] + + +def _turn_capability_name(*, turn_id: str, workflow_id: str) -> str: + digest = hashlib.sha256( + f"{turn_id}\0{workflow_id}".encode("utf-8") + ).hexdigest()[:20] + return f"represented_workflow_{digest}" + + +def _mapping_source_input_key(source_expression: Any) -> str | None: + expression = _normalise_non_empty_text(source_expression) + if expression is None or not expression.startswith("inputs."): + return None + input_path = expression.removeprefix("inputs.").strip() + if not input_path: + return None + return input_path.split(".", 1)[0].strip() or None + + +def _workflow_model_input_schema( + contract: Mapping[str, Any] | None, +) -> dict[str, Any]: + mappings = _bounded_mapping_sequence( + contract.get("input_mappings") if isinstance(contract, Mapping) else None, + limit=40, + ) + represented_mappings: list[dict[str, Any]] = [] + model_properties: dict[str, Any] = {} + required_model_inputs: list[str] = [] + explicit_required_inputs = { + str(item).strip() + for item in ( + contract.get("required_inputs", []) + if isinstance(contract, Mapping) + else [] + ) + if isinstance(item, str) and item.strip() + } + + for mapping in mappings: + target_key = _normalise_non_empty_text( + mapping.get("target_context_key") + or mapping.get("workflow_context_key") + or mapping.get("context_key") + or mapping.get("target_key") + ) + source_expression = _normalise_non_empty_text( + mapping.get("source_expression") + or mapping.get("source") + or mapping.get("source_context_key") + or mapping.get("source_path") + ) + if target_key is None or source_expression is None: + continue + source_input_key = _mapping_source_input_key(source_expression) + required = bool(mapping.get("required")) or target_key in explicit_required_inputs + represented_mapping = { + "target_context_key": target_key, + "source_expression": source_expression, + "extractor": ( + _normalise_non_empty_text(mapping.get("extractor")) or "identity" + ), + "required": required, + } + description = _normalise_non_empty_text(mapping.get("description")) + if description is not None: + represented_mapping["description"] = description[:500] + represented_mappings.append(represented_mapping) + + if ( + source_input_key is None + or source_input_key in _SERVER_PROVIDED_WORKFLOW_INPUT_KEYS + ): + continue + property_schema: dict[str, Any] = {} + if description is not None: + property_schema["description"] = description[:500] + model_properties.setdefault(source_input_key, property_schema) + if required and source_input_key not in required_model_inputs: + required_model_inputs.append(source_input_key) + + inputs_schema: dict[str, Any] = { + "type": "object", + "properties": model_properties, + "additionalProperties": True, + "description": ( + "Workflow launch inputs not already supplied by the server from the " + "turn prompt, authenticated actor, context, or explicit request inputs." + ), + } + if required_model_inputs: + inputs_schema["required"] = sorted(required_model_inputs) + + return { + "type": "object", + "properties": { + "inputs": inputs_schema, + "await_terminal": { + "type": "boolean", + "description": ( + "Wait for a terminal workflow state before returning. " + "Defaults to true." + ), + }, + "timeout_seconds": { + "type": "number", + "minimum": 0.1, + "description": ( + "Maximum bounded terminal wait, further capped by the turn." + ), + }, + "poll_interval_seconds": { + "type": "number", + "minimum": 0.0, + }, + "include_trace": {"type": "boolean"}, + "include_step_result_envelopes": {"type": "boolean"}, + "max_retries": {"type": "integer", "minimum": 0, "maximum": 50}, + }, + "additionalProperties": False, + "x-von-represented-launch-input-mappings": represented_mappings, + "x-von-server-provided-inputs": sorted( + _SERVER_PROVIDED_WORKFLOW_INPUT_KEYS + ), + } + + +@dataclass(frozen=True) +class WorkflowTurnCapability: + """One actor-visible workflow exposed as a turn-bound capability.""" + + name: str + workflow_id: str + display_name: str + description: str + relevance_score: float + input_schema: Mapping[str, Any] + launch_input_contract_source: str | None = None + discovery_metadata: Mapping[str, Any] = field(default_factory=dict) + + def to_catalogue_entry(self) -> dict[str, Any]: + return { + "schema_version": WORKFLOW_TURN_CAPABILITY_SCHEMA_VERSION, + "name": self.name, + "kind": "represented_workflow", + "workflow_id": self.workflow_id, + "display_name": self.display_name, + "description": self.description, + "input_schema": dict(self.input_schema), + "query_match": True, + "semantic_effect": True, + "minimum_effect_window_seconds": 5.0, + "server_bound_arguments": [ + "workflow_id", + "user_id", + "org_id", + "namespace", + ], + "terminal_readback": True, + "relevance_score": round(float(self.relevance_score), 4), + "launch_input_contract_source": self.launch_input_contract_source, + "discovery_metadata": dict(self.discovery_metadata), + } + + +def discover_turn_workflow_capabilities( + query: str, + *, + namespace: str | None, + user_concept_id: str | None, + organisation_concept_id: str | None, + turn_id: str, + max_results: int = 5, + timeout_seconds: float = 5.0, +) -> tuple[list[WorkflowTurnCapability], dict[str, Any]]: + """Return bounded actor-accessible workflow affordances for a model query.""" + + query_text = _normalise_non_empty_text(query) + if query_text is None: + return [], { + "schema_version": WORKFLOW_TURN_DISCOVERY_SCHEMA_VERSION, + "status": "not_requested", + "match_count": 0, + } + if not _normalise_non_empty_text(user_concept_id) or not _normalise_non_empty_text( + namespace + ): + return [], { + "schema_version": WORKFLOW_TURN_DISCOVERY_SCHEMA_VERSION, + "status": "authenticated_actor_required", + "match_count": 0, + } + + from src.backend.services.workflow_discovery_service import ( + discover_workflows_for_turn, + ) + from src.backend.workflows.vontology_loader import ( + resolve_workflow_launch_input_contract, + ) + + bounded_max_results = max(1, min(10, int(max_results))) + bounded_timeout_seconds = max(0.1, min(10.0, float(timeout_seconds))) + with override_current_actor(user_concept_id, organisation_concept_id): + payload = discover_workflows_for_turn( + query_text, + namespace=namespace, + max_results=bounded_max_results, + timeout_seconds=bounded_timeout_seconds, + allow_non_executable=False, + ) + + matches = ( + payload.get("matches") + if isinstance(payload, Mapping) + and isinstance(payload.get("matches"), list) + else [] + ) + capabilities: list[WorkflowTurnCapability] = [] + seen_workflow_ids: set[str] = set() + contract_errors: list[dict[str, Any]] = [] + for raw_match in matches[:bounded_max_results]: + if not isinstance(raw_match, Mapping): + continue + workflow_id = _normalise_non_empty_text(raw_match.get("concept_id")) + if workflow_id is None or workflow_id in seen_workflow_ids: + continue + if raw_match.get("is_executable") is not True: + continue + if raw_match.get("routing_eligible") is not True: + continue + seen_workflow_ids.add(workflow_id) + try: + launch_contract, launch_contract_source = ( + resolve_workflow_launch_input_contract(workflow_id) + ) + except Exception as exc: # noqa: BLE001 + launch_contract = None + launch_contract_source = None + contract_errors.append( + { + "workflow_id": workflow_id, + "stage": "launch_input_contract_resolution", + "error_type": type(exc).__name__, + "error": str(exc)[:500], + } + ) + routing_profile = raw_match.get("routing_profile") + routing_index_metadata = raw_match.get("routing_index_metadata") + capabilities.append( + WorkflowTurnCapability( + name=_turn_capability_name( + turn_id=turn_id, + workflow_id=workflow_id, + ), + workflow_id=workflow_id, + display_name=( + _normalise_non_empty_text(raw_match.get("name")) + or workflow_id + ), + description=( + _normalise_non_empty_text(raw_match.get("description")) + or f"Execute represented workflow {workflow_id}." + ), + relevance_score=_coerce_relevance_score( + raw_match.get("relevance_score") + ), + input_schema=_workflow_model_input_schema(launch_contract), + launch_input_contract_source=( + _normalise_non_empty_text(launch_contract_source) + ), + discovery_metadata={ + "match_source": raw_match.get("match_source"), + "executability_reason": raw_match.get( + "executability_reason" + ), + "routing_readiness_status": raw_match.get( + "routing_readiness_status" + ), + "routing_profile": ( + dict(routing_profile) + if isinstance(routing_profile, Mapping) + else None + ), + "routing_index_metadata": ( + { + key: routing_index_metadata.get(key) + for key in ( + "authority_source", + "description_source", + "publication_lifecycle", + "routing_profile_source", + ) + if key in routing_index_metadata + } + if isinstance(routing_index_metadata, Mapping) + else None + ), + }, + ) + ) + + diagnostic = { + "schema_version": WORKFLOW_TURN_DISCOVERY_SCHEMA_VERSION, + "status": "completed", + "query": query_text, + "match_count": len(capabilities), + "candidate_count": ( + payload.get("candidate_count") if isinstance(payload, Mapping) else None + ), + "search_time_ms": ( + payload.get("search_time_ms") if isinstance(payload, Mapping) else None + ), + "budget_exhausted": bool( + payload.get("budget_exhausted") + if isinstance(payload, Mapping) + else False + ), + "match_absence_reason": ( + payload.get("match_absence_reason") + if isinstance(payload, Mapping) + else None + ), + "errors": ( + [*list(payload.get("errors") or [])[:5], *contract_errors[:5]] + if isinstance(payload, Mapping) + else contract_errors[:5] + ), + } + return capabilities, diagnostic + + +def _bounded_context_projection( + context: Sequence[Mapping[str, Any]] | None, +) -> list[dict[str, Any]]: + projected: list[dict[str, Any]] = [] + remaining_chars = 24_000 + for item in list(context or ())[-20:]: + if not isinstance(item, Mapping) or remaining_chars <= 0: + continue + role = _normalise_non_empty_text(item.get("role")) or "context" + content = item.get("content") + if isinstance(content, str): + bounded_content = content[: min(4_000, remaining_chars)] + projected.append({"role": role, "content": bounded_content}) + remaining_chars -= len(bounded_content) + return projected + + +def _coerce_relevance_score(value: Any) -> float: + try: + return max(0.0, min(1.0, float(value or 0.0))) + except (TypeError, ValueError): + return 0.0 + + +def build_workflow_execution_arguments( + capability: WorkflowTurnCapability, + model_arguments: Mapping[str, Any], + *, + prompt: str, + context: Sequence[Mapping[str, Any]] | None, + request_workflow_launch_inputs: Mapping[str, Any] | None, + user_concept_id: str, + organisation_concept_id: str | None, + namespace: str, + maximum_wait_seconds: float, +) -> tuple[dict[str, Any], dict[str, Any]]: + """Bind workflow identity/actor and build verified-submission arguments.""" + + unknown_arguments = sorted( + str(key) + for key in model_arguments + if str(key) not in _MODEL_WORKFLOW_CONTROL_ARGUMENTS + ) + raw_model_inputs = model_arguments.get("inputs") + if raw_model_inputs is None: + model_inputs: dict[str, Any] = {} + elif isinstance(raw_model_inputs, Mapping): + model_inputs = { + str(key): value + for key, value in raw_model_inputs.items() + if isinstance(key, str) + and key.strip() + and key not in _SERVER_PROVIDED_WORKFLOW_INPUT_KEYS + and not key.startswith("__") + } + else: + raise ValueError("workflow capability inputs must be an object") + + trusted_request_inputs = ( + { + str(key): value + for key, value in request_workflow_launch_inputs.items() + if isinstance(key, str) + and key.strip() + and key not in _SERVER_PROVIDED_WORKFLOW_INPUT_KEYS + and not key.startswith("__") + } + if isinstance(request_workflow_launch_inputs, Mapping) + else {} + ) + launch_inputs = dict(model_inputs) + for key, value in trusted_request_inputs.items(): + if isinstance(key, str) and key.strip() and not key.startswith("__"): + launch_inputs[key] = value + if trusted_request_inputs: + launch_inputs["workflow_launch_inputs"] = dict(trusted_request_inputs) + launch_inputs.update( + { + "actor_concept_id": user_concept_id, + "actor_user_concept_id": user_concept_id, + "user_concept_id": user_concept_id, + "namespace": namespace, + "user_namespace": namespace, + } + ) + if organisation_concept_id is not None: + launch_inputs.update( + { + "org_concept_id": organisation_concept_id, + "organisation_concept_id": organisation_concept_id, + } + ) + clean_prompt = str(prompt or "").strip() + if clean_prompt: + launch_inputs["prompt"] = clean_prompt + launch_inputs["user_prompt"] = clean_prompt + context_projection = _bounded_context_projection(context) + if context_projection: + launch_inputs["augmented_context"] = context_projection + + try: + requested_wait = float(model_arguments.get("timeout_seconds", 60.0)) + except (TypeError, ValueError): + requested_wait = 60.0 + effective_wait = max( + 0.1, + min( + requested_wait if requested_wait > 0.0 else 0.1, + max(0.1, float(maximum_wait_seconds)), + 90.0, + ), + ) + try: + poll_interval = float(model_arguments.get("poll_interval_seconds", 0.5)) + except (TypeError, ValueError): + poll_interval = 0.5 + try: + max_retries = int(model_arguments.get("max_retries", 3)) + except (TypeError, ValueError): + max_retries = 3 + + effective_arguments = { + "workflow_id": capability.workflow_id, + "user_id": user_concept_id, + "namespace": namespace, + "inputs": launch_inputs, + "max_retries": max(0, min(max_retries, 50)), + "await_terminal": _coerce_bool( + model_arguments.get("await_terminal"), + default=True, + ), + "timeout_seconds": effective_wait, + "poll_interval_seconds": max(0.0, poll_interval), + "include_step_result_envelopes": _coerce_bool( + model_arguments.get("include_step_result_envelopes"), + default=False, + ), + "include_trace": _coerce_bool( + model_arguments.get("include_trace"), + default=False, + ), + } + if organisation_concept_id is not None: + effective_arguments["org_id"] = organisation_concept_id + binding_diagnostics = { + "schema_version": "workflow_turn_capability_binding.v1", + "capability_name": capability.name, + "workflow_id": capability.workflow_id, + "server_bound_arguments": [ + "workflow_id", + "user_id", + "org_id", + "namespace", + ], + "ignored_model_arguments": unknown_arguments, + "trusted_request_input_keys": sorted(trusted_request_inputs), + "model_input_keys": sorted(model_inputs), + "effective_wait_seconds": effective_wait, + } + return effective_arguments, binding_diagnostics + + +def normalise_workflow_effect_receipt( + payload: Any, + *, + capability: WorkflowTurnCapability, +) -> Any: + """Project durable workflow launch/terminal state into effect semantics.""" + + if not isinstance(payload, Mapping): + return payload + receipt = dict(payload) + instance_id = _normalise_non_empty_text(receipt.get("instance_id")) + final_status = _normalise_non_empty_text(receipt.get("final_status")) + workflow_instance = receipt.get("workflow_instance") + if final_status is None and isinstance(workflow_instance, Mapping): + final_status = _normalise_non_empty_text(workflow_instance.get("status")) + status_key = (final_status or "").lower() + timed_out = bool(receipt.get("timed_out")) + created_new = receipt.get("created_new") + changed = bool(instance_id and created_new is not False) + + if receipt.get("success") is False and instance_id is None: + effect_status = "failed" + changed = False + elif status_key in _TERMINAL_SUCCESS_STATUSES and not timed_out: + effect_status = "succeeded" + elif status_key in _TERMINAL_FAILURE_STATUSES: + effect_status = "failed" + elif ( + timed_out + or status_key in {"pending", "queued", "running", "paused"} + or not status_key + ): + effect_status = "partial" if instance_id is not None else "failed" + else: + effect_status = "partial" if instance_id is not None else "failed" + + receipt.update( + { + "workflow_turn_receipt_schema_version": ( + WORKFLOW_TURN_RECEIPT_SCHEMA_VERSION + ), + "capability_kind": "represented_workflow", + "capability_name": capability.name, + "workflow_id": capability.workflow_id, + "effect_status": effect_status, + "changed": changed, + "mutation_outcome": ( + "completed" + if effect_status == "succeeded" + else ("partial" if changed else "not_started") + ), + "outcome_finality": "terminal_for_turn", + } + ) + if effect_status != "succeeded": + recovery_affordances = list(receipt.get("recovery_affordances") or []) + if instance_id is not None: + recovery_affordances.append( + { + "action_type": "inspect_workflow_instance", + "capability": "workflow_get_instance", + "arguments": {"instance_id": instance_id}, + } + ) + else: + recovery_affordances.append( + { + "action_type": "inspect_launch_failure_before_retry", + "workflow_id": capability.workflow_id, + } + ) + receipt["recovery_affordances"] = recovery_affordances + return receipt + + +__all__ = [ + "WorkflowTurnCapability", + "build_workflow_execution_arguments", + "discover_turn_workflow_capabilities", + "normalise_workflow_effect_receipt", +] diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py index 7afe88302..69e7f4ead 100644 --- a/tests/backend/test_adaptive_turn_service.py +++ b/tests/backend/test_adaptive_turn_service.py @@ -320,6 +320,53 @@ def _effect_gateway( ) +def _workflow_gateway(handler: Any) -> InternalMCPGateway: + catalogue = MethodCatalogue() + catalogue.register( + MethodDefinition( + name="general_read", + handler=lambda **_kwargs: {"success": True}, + input_schema=Schema(allow_unknown=True), + category="read", + ordinary_turn_public=True, + ) + ) + catalogue.register( + MethodDefinition( + name="workflow_execute", + handler=handler, + input_schema=Schema( + required={"workflow_id": str}, + optional={ + "user_id": str, + "org_id": (str, type(None)), + "namespace": str, + "inputs": dict, + "max_retries": int, + "await_terminal": bool, + "timeout_seconds": (int, float), + "poll_interval_seconds": (int, float), + "include_step_result_envelopes": bool, + "include_trace": bool, + }, + allow_unknown=False, + ), + output_schema=Schema(required={"success": bool}, allow_unknown=True), + category="write", + timeout_sec=1.0, + effect_admission_window_sec=0.01, + ) + ) + return InternalMCPGateway( + catalogue=catalogue, + transport=InternalMCPTransport( + read_timeout_sec=1.0, + write_timeout_sec=1.0, + ), + enabled=True, + ) + + class _ManualClock: def __init__(self, now: float = 0.0) -> None: self.now = now @@ -4244,3 +4291,230 @@ def handler(**kwargs: Any) -> dict[str, Any]: result.tool_invocations[0]["effective_payload"]["error_code"] == "capability_not_delegated" ) + + +def test_represented_workflow_is_discovered_and_invoked_as_bound_capability( + monkeypatch, +) -> None: + from src.backend.services.workflow_turn_capability_service import ( + WorkflowTurnCapability, + ) + + seen_arguments: dict[str, Any] = {} + workflow_capability = WorkflowTurnCapability( + name="represented_workflow_turn_test", + workflow_id="#V#represented_test_workflow", + display_name="Represented test workflow", + description="Produce the represented test work product.", + relevance_score=0.94, + input_schema={ + "type": "object", + "properties": { + "inputs": { + "type": "object", + "properties": {"record_id": {"type": "string"}}, + } + }, + "additionalProperties": False, + }, + ) + monkeypatch.setattr( + "src.backend.services.workflow_turn_capability_service." + "discover_turn_workflow_capabilities", + lambda *_args, **_kwargs: ( + [workflow_capability], + { + "schema_version": "workflow_turn_capability_discovery.v1", + "status": "completed", + "match_count": 1, + }, + ), + ) + + def _execute_workflow(**kwargs: Any) -> dict[str, Any]: + seen_arguments.update(kwargs) + return { + "success": True, + "instance_id": "workflow-instance-1", + "created_new": True, + "final_status": "completed", + } + + gateway = _workflow_gateway(_execute_workflow) + assert "workflow_execute" not in ordinary_turn_capability_delegation( + gateway, + user_concept_id="#V#real_user", + ) + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_capabilities", + call_id="discover-workflow", + payload={"query": "produce the represented work product"}, + ) + ], + continuation=LLMContinuation( + provider="test", + api_surface="responses", + model="test-model", + response_id="workflow-discovery-response", + ), + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="invoke-workflow", + payload={ + "name": workflow_capability.name, + "arguments": { + "workflow_id": "#V#spoofed_workflow", + "user_id": "#V#spoofed_user", + "inputs": { + "record_id": "#V#record", + "user_concept_id": "#V#spoofed_user", + }, + }, + }, + ) + ], + continuation=LLMContinuation( + provider="test", + api_surface="responses", + model="test-model", + response_id="workflow-invocation-response", + ), + ), + LLMResponse(text_response="The represented work product was completed."), + ) + + result = execute_adaptive_turn( + gateway=gateway, + prompt="Produce the represented work product.", + context=[], + llm_client=client, + model="test-model", + user_namespace="#V#real_user@real_org", + user_concept_id="#V#real_user", + org_concept_id="#V#real_org", + workflow_launch_inputs={"authorised_record": "#V#request_record"}, + turn_id="turn-represented-workflow", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + catalogue_result = client.calls[1]["tool_results"][0].output + assert catalogue_result["represented_workflow_total"] == 1 + assert catalogue_result["capabilities"][0]["kind"] == ( + "represented_workflow" + ) + assert seen_arguments["workflow_id"] == "#V#represented_test_workflow" + assert seen_arguments["user_id"] == "#V#real_user" + assert seen_arguments["org_id"] == "#V#real_org" + assert seen_arguments["namespace"] == "#V#real_user@real_org" + assert seen_arguments["inputs"]["user_concept_id"] == "#V#real_user" + assert seen_arguments["inputs"]["record_id"] == "#V#record" + assert seen_arguments["inputs"]["authorised_record"] == "#V#request_record" + invocation = next( + item + for item in result.tool_invocations + if item.get("tool") == workflow_capability.name + ) + assert invocation["tool"] == workflow_capability.name + assert invocation["execution_method"] == "workflow_execute" + assert invocation["capability_kind"] == "represented_workflow" + assert invocation["represented_workflow_id"] == ( + "#V#represented_test_workflow" + ) + assert invocation["effect_status"] == "succeeded" + assert invocation["changed"] is True + assert result.response_text == "The represented work product was completed." + + +def test_unchanged_terminally_failed_effect_is_not_dispatched_twice( + monkeypatch, +) -> None: + handler_calls: list[dict[str, Any]] = [] + + monkeypatch.setattr( + "src.backend.services.adaptive_turn_service." + "_effect_subject_authorised", + lambda *_args, **_kwargs: True, + ) + + def handler(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + assert name == "add_relationship" + handler_calls.append(arguments) + return { + "success": False, + "effect_status": "failed", + "changed": False, + "error_code": "invalid_predicate_format", + "retryable": False, + } + + effect_arguments = { + "source_id": "#V#source", + "predicate": "unresolved predicate", + "target": "#V#target", + } + client = _SequenceClient( + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="invalid-effect-1", + payload={ + "name": "add_relationship", + "arguments": effect_arguments, + }, + ) + ], + ), + LLMResponse( + text_response="", + tool_calls=[ + ToolCall( + tool_name="turn_invoke_capability", + call_id="invalid-effect-2", + payload={ + "name": "add_relationship", + "arguments": effect_arguments, + }, + ) + ], + ), + LLMResponse( + text_response=( + "The relationship was not changed; a different typed predicate " + "reference is required." + ) + ), + ) + + result = execute_adaptive_turn( + gateway=_effect_gateway(handler, effect_admission_window_sec=0.01), + prompt="Add this relationship.", + context=[], + llm_client=client, + model="test-model", + user_concept_id="#V#user", + turn_id="turn-repeat-terminal-failure", + turn_budget_seconds=20, + final_synthesis_reserve_seconds=2, + ) + + assert handler_calls == [effect_arguments] + assert len(result.tool_invocations) == 2 + assert result.tool_invocations[0]["error_code"] == ( + "invalid_predicate_format" + ) + assert result.tool_invocations[1]["error_code"] == ( + "effect_request_unchanged_after_terminal_failure" + ) + assert result.tool_invocations[1]["effect_status"] == "not_started" + assert result.tool_invocations[1]["changed"] is False diff --git a/tests/backend/test_internal_mcp_add_relationship_predicate_validation.py b/tests/backend/test_internal_mcp_add_relationship_predicate_validation.py index 7eb699f48..83a6947ac 100644 --- a/tests/backend/test_internal_mcp_add_relationship_predicate_validation.py +++ b/tests/backend/test_internal_mcp_add_relationship_predicate_validation.py @@ -1,7 +1,9 @@ import pytest -def test_add_relationship_rejects_non_vontology_predicate_keys(monkeypatch): +def test_add_relationship_returns_typed_recovery_for_unknown_predicate_name( + monkeypatch, +): from src.backend.integrations.internal_mcp import catalogue calls = {"find_one": [], "update_one": []} @@ -35,10 +37,216 @@ def _fake_update_one(filter_doc, update_doc, upsert=False): ) assert result["success"] is False - assert result.get("error_code") == "invalid_predicate_format" - assert "invalid_predicate_format" in (result.get("error") or "") - assert "mutation_outcome" not in result - assert result.get("effect_status") != "indeterminate" + assert result.get("error_code") == "predicate_reference_not_found" + assert result["mutation_outcome"] == "not_started" + assert result["changed"] is False + assert result["predicate_resolution"]["status"] == "not_found" + assert result["recovery_affordances"][0]["action_type"] == ( + "retry_with_predicate_concept_id" + ) + assert calls["update_one"] == [] + + +def test_add_relationship_resolves_accessible_predicate_name_before_write( + monkeypatch, +): + from src.backend.integrations.internal_mcp import catalogue + + captured: dict[str, object] = {} + + monkeypatch.setattr( + "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one", + lambda filter_doc, projection=None: { + "concept_id": filter_doc.get("concept_id"), + "relationships": ( + {"is_an_instance_of": ["#V#predicate"]} + if filter_doc.get("concept_id") == "#V#depends_on" + else {} + ), + }, + ) + monkeypatch.setattr( + "src.backend.services.concept_resolution_service.resolve_concept_by_name", + lambda **kwargs: { + "success": True, + "status": "resolved", + "resolved_concept_id": "#V#depends_on", + "match": {"stage": "exact_name"}, + "candidates": [], + "audit": [], + }, + ) + + def _add_edge(**kwargs): + captured.update(kwargs) + return { + "success": True, + "predicate": kwargs["predicate"], + "target_id": kwargs["target"], + "modified": True, + } + + monkeypatch.setattr( + "src.backend.services.relationship_write_service.add_relationship", + _add_edge, + ) + + result = catalogue._add_relationship( + source_id="#V#source", + predicate="Depends on", + target="#V#target", + ) + + assert result["success"] is True + assert captured["predicate"] == "#V#depends_on" + assert result["predicate_resolution"]["status"] == "resolved" + assert result["predicate_resolution"]["resolved_concept_id"] == ( + "#V#depends_on" + ) + + +def test_add_relationship_returns_candidates_for_ambiguous_predicate_name( + monkeypatch, +): + from src.backend.integrations.internal_mcp import catalogue + + monkeypatch.setattr( + "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one", + lambda filter_doc, projection=None: ( + {"concept_id": "#V#source", "relationships": {}} + if filter_doc.get("concept_id") == "#V#source" + else None + ), + ) + monkeypatch.setattr( + "src.backend.services.concept_resolution_service.resolve_concept_by_name", + lambda **kwargs: { + "success": True, + "status": "ambiguous", + "resolved_concept_id": None, + "candidates": [ + {"concept_id": "#V#about"}, + {"concept_id": "#V#is_about"}, + ], + "audit": [], + }, + ) + monkeypatch.setattr( + "src.backend.services.relationship_write_service.add_relationship", + lambda **_kwargs: (_ for _ in ()).throw( + AssertionError("an ambiguous predicate must not be written") + ), + ) + + result = catalogue._add_relationship( + source_id="#V#source", + predicate="about", + target="#V#target", + ) + + assert result["success"] is False + assert result["error_code"] == "predicate_reference_ambiguous" + assert result["mutation_outcome"] == "not_started" + assert [ + item["predicate_ref"]["concept_id"] + for item in result["recovery_affordances"] + ] == ["#V#about", "#V#is_about"] + + +def test_add_relationship_typed_reference_can_create_text_predicate( + monkeypatch, +): + from src.backend.integrations.internal_mcp import catalogue + + events: list[str] = [] + validation_results = iter( + [ + ( + False, + "predicate_concept_not_found", + {"predicate": "#V#has_summary"}, + ), + (True, None, None), + ] + ) + + def _find_one(filter_doc, projection=None): + concept_id = filter_doc.get("concept_id") + if concept_id == "#V#source": + return {"concept_id": concept_id, "relationships": {}} + if concept_id == "#V#has_summary": + return { + "concept_id": concept_id, + "relationships": { + "is_an_instance_of": ["#V#binary_text_predicate"] + }, + } + return None + + monkeypatch.setattr( + "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one", + _find_one, + ) + monkeypatch.setattr( + "src.backend.services.concept_resolution_service.resolve_concept_by_name", + lambda **kwargs: { + "success": True, + "status": "not_found", + "resolved_concept_id": None, + "candidates": [], + "audit": [], + }, + ) + monkeypatch.setattr( + "src.backend.services.relationship_write_service.validate_predicate_concept", + lambda predicate, repo=None: next(validation_results), + ) + + def _create_predicate(**kwargs): + events.append("create") + assert kwargs["parent_id"] == "#V#binary_text_predicate" + assert kwargs["concepts"][0]["name"] == "Has summary" + return { + "success": True, + "effect_status": "succeeded", + "changed": True, + "created_concept_ids": ["#V#has_summary"], + } + + monkeypatch.setattr(catalogue, "_create_concepts", _create_predicate) + + def _upsert_text(**kwargs): + events.append("text") + assert kwargs["predicate"] == "#V#has_summary" + assert kwargs["text"] == "A concise summary" + return { + "text_value_id": "text-1", + "relation_id": "relation-1", + "relation_created": True, + "context_updated": False, + } + + monkeypatch.setattr( + "src.backend.services.text_value_service.upsert_text_for_concept", + _upsert_text, + ) + + result = catalogue._add_relationship( + source_id="#V#source", + target="A concise summary", + predicate_ref={ + "name": "Has summary", + "on_missing": "create_typed_predicate", + "value_kind": "text", + }, + ) + + assert events == ["create", "text"] + assert result["success"] is True + assert result["relationship_type"] == "text_relation" + assert result["changed"] is True + assert result["predicate"] == "#V#has_summary" + assert result["predicate_dependency"]["status"] == "created" def test_add_relationship_rejects_missing_dynamic_predicate_concepts(monkeypatch): diff --git a/tests/backend/test_von_generate_workflow_instances.py b/tests/backend/test_von_generate_workflow_instances.py index 7eda05203..dd3e18eb1 100644 --- a/tests/backend/test_von_generate_workflow_instances.py +++ b/tests/backend/test_von_generate_workflow_instances.py @@ -256,6 +256,28 @@ def test_generate_canonicalises_actor_scope_for_the_adaptive_turn( assert adaptive_call["user_namespace"] == "#V#michael_witbrock@sail_lab" +def test_generate_passes_authorised_workflow_inputs_to_adaptive_capabilities( + app: Flask, +) -> None: + response = app.test_client().post( + "/von/generate", + json={ + "prompt": "Use the represented capability with these inputs.", + "workflow_inputs": { + "record_id": "#V#record", + "maximum_results": 3, + }, + }, + ) + + assert response.status_code == 200 + adaptive_call = app.config["_ADAPTIVE_TURN_CALLS"][-1] + assert adaptive_call["workflow_launch_inputs"] == { + "record_id": "#V#record", + "maximum_results": 3, + } + + def test_body_identity_cannot_override_actor_scope_passed_to_adaptive_turn( app: Flask, ) -> None: diff --git a/tests/backend/test_workflow_turn_capability_service.py b/tests/backend/test_workflow_turn_capability_service.py new file mode 100644 index 000000000..f8c7d1b82 --- /dev/null +++ b/tests/backend/test_workflow_turn_capability_service.py @@ -0,0 +1,243 @@ +from __future__ import annotations + + +def _capability(): + from src.backend.services.workflow_turn_capability_service import ( + WorkflowTurnCapability, + ) + + return WorkflowTurnCapability( + name="represented_workflow_test", + workflow_id="#V#test_workflow", + display_name="Test workflow", + description="Produce a represented work product.", + relevance_score=0.91, + input_schema={"type": "object", "properties": {}}, + launch_input_contract_source="text_relation:test", + ) + + +def test_discovery_exposes_only_executable_routing_eligible_workflows( + monkeypatch, +): + from src.backend.services.workflow_turn_capability_service import ( + discover_turn_workflow_capabilities, + ) + + monkeypatch.setattr( + "src.backend.services.workflow_discovery_service." + "discover_workflows_for_turn", + lambda *_args, **_kwargs: { + "matches": [ + { + "concept_id": "#V#usable_workflow", + "name": "Usable workflow", + "description": "Produce the requested reusable work product.", + "relevance_score": 0.87, + "is_executable": True, + "routing_eligible": True, + "match_source": "semantic", + }, + { + "concept_id": "#V#draft_workflow", + "name": "Draft workflow", + "relevance_score": 0.99, + "is_executable": True, + "routing_eligible": False, + }, + { + "concept_id": "#V#broken_workflow", + "name": "Broken workflow", + "relevance_score": 0.98, + "is_executable": False, + "routing_eligible": True, + }, + ], + "candidate_count": 3, + "search_time_ms": 12.5, + "errors": [], + }, + ) + monkeypatch.setattr( + "src.backend.workflows.vontology_loader." + "resolve_workflow_launch_input_contract", + lambda workflow_id: ( + { + "required_inputs": ["record_id", "prompt"], + "input_mappings": [ + { + "target_context_key": "prompt", + "source_expression": "inputs.prompt", + "required": True, + }, + { + "target_context_key": "record_id", + "source_expression": "inputs.record_id", + "required": True, + "description": "The represented record to process.", + }, + ], + }, + "text_relation:launch_contract", + ), + ) + + capabilities, diagnostic = discover_turn_workflow_capabilities( + "produce the reusable work product", + namespace="#V#user@org", + user_concept_id="#V#user", + organisation_concept_id="#V#org", + turn_id="turn-123", + ) + + assert [item.workflow_id for item in capabilities] == [ + "#V#usable_workflow" + ] + capability = capabilities[0] + assert capability.name.startswith("represented_workflow_") + inputs_schema = capability.input_schema["properties"]["inputs"] + assert list(inputs_schema["properties"]) == ["record_id"] + assert inputs_schema["required"] == ["record_id"] + assert "prompt" in capability.input_schema["x-von-server-provided-inputs"] + assert diagnostic["status"] == "completed" + assert diagnostic["match_count"] == 1 + assert diagnostic["candidate_count"] == 3 + + +def test_discovery_requires_authenticated_actor_without_querying_index( + monkeypatch, +): + from src.backend.services.workflow_turn_capability_service import ( + discover_turn_workflow_capabilities, + ) + + monkeypatch.setattr( + "src.backend.services.workflow_discovery_service." + "discover_workflows_for_turn", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("actorless discovery must not query the index") + ), + ) + + capabilities, diagnostic = discover_turn_workflow_capabilities( + "produce a represented work product", + namespace="#V#user@org", + user_concept_id=None, + organisation_concept_id="#V#org", + turn_id="turn-123", + ) + + assert capabilities == [] + assert diagnostic["status"] == "authenticated_actor_required" + + +def test_execution_arguments_bind_workflow_and_actor_server_side(): + from src.backend.services.workflow_turn_capability_service import ( + build_workflow_execution_arguments, + ) + + effective, diagnostic = build_workflow_execution_arguments( + _capability(), + { + "workflow_id": "#V#spoofed_workflow", + "user_id": "#V#spoofed_user", + "org_id": "#V#spoofed_org", + "namespace": "#V#spoofed@org", + "inputs": { + "record_id": "#V#record", + "user_concept_id": "#V#spoofed_user", + }, + "await_terminal": "false", + "timeout_seconds": 500, + }, + prompt="Process this record.", + context=[{"role": "user", "content": "Earlier context"}], + request_workflow_launch_inputs={ + "file_copy_concept_id": "#V#authorised_file", + "organisation_concept_id": "#V#spoofed_org", + }, + user_concept_id="#V#real_user", + organisation_concept_id="#V#real_org", + namespace="#V#real_user@real_org", + maximum_wait_seconds=20, + ) + + assert effective["workflow_id"] == "#V#test_workflow" + assert effective["user_id"] == "#V#real_user" + assert effective["org_id"] == "#V#real_org" + assert effective["namespace"] == "#V#real_user@real_org" + assert effective["await_terminal"] is False + assert effective["timeout_seconds"] == 20 + assert effective["inputs"]["record_id"] == "#V#record" + assert effective["inputs"]["file_copy_concept_id"] == "#V#authorised_file" + assert effective["inputs"]["user_concept_id"] == "#V#real_user" + assert effective["inputs"]["organisation_concept_id"] == "#V#real_org" + assert effective["inputs"]["prompt"] == "Process this record." + assert effective["inputs"]["augmented_context"] == [ + {"role": "user", "content": "Earlier context"} + ] + assert diagnostic["ignored_model_arguments"] == [ + "namespace", + "org_id", + "user_id", + "workflow_id", + ] + + +def test_workflow_receipt_distinguishes_completion_partial_failure_and_no_start(): + from src.backend.services.workflow_turn_capability_service import ( + normalise_workflow_effect_receipt, + ) + + capability = _capability() + completed = normalise_workflow_effect_receipt( + { + "success": True, + "instance_id": "instance-1", + "created_new": True, + "final_status": "completed", + }, + capability=capability, + ) + partial = normalise_workflow_effect_receipt( + { + "success": True, + "instance_id": "instance-2", + "created_new": True, + "final_status": "running", + "timed_out": True, + }, + capability=capability, + ) + terminal_failure = normalise_workflow_effect_receipt( + { + "success": True, + "instance_id": "instance-3", + "created_new": True, + "final_status": "failed", + }, + capability=capability, + ) + no_start = normalise_workflow_effect_receipt( + { + "success": False, + "error_code": "workflow_submission_failed", + }, + capability=capability, + ) + + assert completed["effect_status"] == "succeeded" + assert completed["changed"] is True + assert partial["effect_status"] == "partial" + assert partial["recovery_affordances"][0]["arguments"] == { + "instance_id": "instance-2" + } + assert terminal_failure["effect_status"] == "failed" + assert terminal_failure["changed"] is True + assert terminal_failure["mutation_outcome"] == "partial" + assert terminal_failure["recovery_affordances"][0]["arguments"] == { + "instance_id": "instance-3" + } + assert no_start["effect_status"] == "failed" + assert no_start["changed"] is False + assert no_start["mutation_outcome"] == "not_started"