feat(telemetry): add ProductionDebtVectorGate and TechnicalDueDiligenceLedger #1355
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,196 @@ | ||
| from __future__ import annotations | ||
|
|
||
| import hashlib | ||
| import json | ||
| import logging | ||
| import os | ||
| from dataclasses import dataclass | ||
| from datetime import datetime, timezone | ||
| from pathlib import Path | ||
| from typing import Any, Dict, List, Optional | ||
|
|
||
| log: logging.Logger = logging.getLogger(__name__) | ||
|
|
||
| GENESIS_HASH: str = ( | ||
| "0000000000000000000000000000000000000000000000000000000000000000" | ||
| ) | ||
|
|
||
|
|
||
| @dataclass | ||
| class VectorDebtReport: | ||
| collection_name: str | ||
| vdi_score: float # Vector Debt Index (target <= 12.0) | ||
| memory_multiplier: float # Target <= 1.10x | ||
| search_latency_ms: float # Target <= 25.0ms | ||
| mutation_safety_score: float # Target 100.0 | ||
| production_readiness_index: float # Scale 0 - 100 | ||
| is_production_ready: bool | ||
| critical_smells: List[str] | ||
| receipt_hash: str | ||
|
|
||
|
|
||
| class TechnicalDueDiligenceLedger: | ||
| """ | ||
| Cryptographic SHA-256 hash-chained Action Ledger for Qdrant enterprise vector collections. | ||
| """ | ||
|
|
||
| def __init__(self) -> None: | ||
| self._entries: List[Dict[str, Any]] = [] | ||
| self._last_hash: str = GENESIS_HASH | ||
|
|
||
| def record_collection_event( | ||
| self, | ||
| collection_name: str, | ||
| event_type: str, | ||
| readiness_index: float, | ||
| critical_smells: List[str], | ||
| metadata: Dict[str, Any], | ||
| ) -> Dict[str, Any]: | ||
| timestamp = datetime.now(timezone.utc).isoformat() | ||
| index = len(self._entries) | ||
|
|
||
| meta_bytes = json.dumps(metadata, sort_keys=True).encode("utf-8") | ||
| canonical_content = f"{index}|{self._last_hash}|{collection_name}|{event_type}|{readiness_index}|{timestamp}|{hashlib.sha256(meta_bytes).hexdigest()}" | ||
| curr_hash = hashlib.sha256(canonical_content.encode("utf-8")).hexdigest() | ||
|
|
||
| entry = { | ||
| "index": index, | ||
| "timestamp": timestamp, | ||
| "collection_name": collection_name, | ||
| "event_type": event_type, | ||
| "readiness_index": readiness_index, | ||
| "critical_smells": critical_smells, | ||
| "prev_hash": self._last_hash, | ||
| "curr_hash": curr_hash, | ||
| "metadata": metadata, | ||
| } | ||
|
|
||
| self._entries.append(entry) | ||
| self._last_hash = curr_hash | ||
| return entry | ||
|
|
||
| def get_ledger_entries(self) -> List[Dict[str, Any]]: | ||
| return list(self._entries) | ||
|
|
||
| def verify_ledger_integrity(self) -> bool: | ||
| prev = GENESIS_HASH | ||
| for entry in self._entries: | ||
| if entry["prev_hash"] != prev: | ||
| return False | ||
| prev = entry["curr_hash"] | ||
| return True | ||
|
|
||
|
|
||
| class ProductionDebtVectorGate: | ||
| """ | ||
| A2Z SOC Production Debt & Technical Due Diligence Gate for Qdrant Vector Search. | ||
|
|
||
| Quantifies vector collection memory and search latency against 4 Enterprise Forward Deployed Engineering KPIs: | ||
| 1. Vector Debt Index (VDI <= 12.0) | ||
| 2. HNSW Index Memory Multiplier (IMM <= 1.10x) | ||
| 3. P99 Vector Search Latency Ceiling (<= 25ms) | ||
| 4. Deterministic Mutation Boundaries (never_equate_intent_to_approval) | ||
| """ | ||
|
|
||
| def __init__( | ||
| self, | ||
| never_equate_intent_to_approval: bool = True, | ||
| max_acceptable_vdi: float = 12.0, | ||
| ) -> None: | ||
| self.never_equate_intent_to_approval = never_equate_intent_to_approval | ||
| self.max_acceptable_vdi = max_acceptable_vdi | ||
| self.ledger = TechnicalDueDiligenceLedger() | ||
|
|
||
| def check_kill_switch(self) -> bool: | ||
| if os.environ.get("AAG_KILL_SWITCH", "").lower() in ("true", "1", "yes"): | ||
| return True | ||
| for path_str in ("artifacts/KILL", "/tmp/KILL"): | ||
| if Path(path_str).exists(): | ||
| return True | ||
|
Comment on lines
+107
to
+109
There was a problem hiding this comment. 🔒 Security & Privacy | 🟠 Major | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
rg -n -C 3 'AAG_KILL_SWITCH|/tmp/KILL|artifacts/KILL' .Repository: qdrant/qdrant-client Length of output: 1516 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- production_debt.py ---'
sed -n '1,180p' qdrant_client/production_debt.py
printf '%s\n' '--- kill-switch usages and tests ---'
rg -n -C 4 'check_kill_switch|evaluate_collection|KILL|production_debt' tests qdrant_client --glob '*.py'Repository: qdrant/qdrant-client Length of output: 12791 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path("qdrant_client/production_debt.py")
text = p.read_text()
for needle in ("def check_kill_switch", "def evaluate_collection"):
start = text.index(needle)
end = text.find("\n def ", start + 5)
if end == -1:
end = len(text)
print(f"--- {needle} ---")
print("\n".join(f"{i+1:4}: {line}" for i, line in enumerate(text[start:end].splitlines(), start=text[:start].count("\n")+1)))
PY
printf '%s\n' '--- all call sites ---'
rg -n -C 3 'check_kill_switch\(' .Repository: qdrant/qdrant-client Length of output: 5764 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
for name in ("tests/test_production_debt.py",):
p = Path(name)
if p.exists():
print(f"--- {name} ---")
print(p.read_text())
PY
printf '%s\n' '--- repository guidance and packaging context ---'
rg -n -C 3 'kill switch|kill-switch|AAG_KILL_SWITCH|artifacts/KILL|/tmp' README.md docs pyproject.toml setup.cfg setup.py qdrant_client tests 2>/dev/null || trueRepository: qdrant/qdrant-client Length of output: 6038 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import ast
from pathlib import Path
source_path = Path("qdrant_client/production_debt.py")
tree = ast.parse(source_path.read_text())
functions = {
node.name: node
for node in ast.walk(tree)
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
check = functions["check_kill_switch"]
kill_paths = [
node.value.value
for node in ast.walk(check)
if isinstance(node, ast.Constant) and isinstance(node.value, str)
]
exists_calls = [
node
for node in ast.walk(check)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "exists"
]
evaluate = functions["evaluate_collection"]
calls_check = any(
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "check_kill_switch"
for node in ast.walk(evaluate)
)
raises_permission_error = any(
isinstance(node, ast.Raise)
and isinstance(node.exc, ast.Call)
and isinstance(node.exc.func, ast.Name)
and node.exc.func.id == "PermissionError"
for node in ast.walk(evaluate)
)
print({"kill_switch_literals": kill_paths,
"exists_call_count": len(exists_calls),
"evaluate_calls_check_kill_switch": calls_check,
"evaluate_raises_permission_error": raises_permission_error})
PY
stat -c 'tmp_mode=%a tmp_owner=%U tmp_group=%G' /tmpRepository: qdrant/qdrant-client Length of output: 327 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import ast
from pathlib import Path
tree = ast.parse(Path("qdrant_client/production_debt.py").read_text())
functions = {
node.name: node
for node in ast.walk(tree)
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
check = functions["check_kill_switch"]
kill_paths = [
node.value
for node in ast.walk(check)
if isinstance(node, ast.Constant) and isinstance(node.value, str)
]
exists_calls = [
node
for node in ast.walk(check)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "exists"
]
evaluate = functions["evaluate_collection"]
calls_check = any(
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "check_kill_switch"
for node in ast.walk(evaluate)
)
raises_permission_error = any(
isinstance(node, ast.Raise)
and isinstance(node.exc, ast.Call)
and isinstance(node.exc.func, ast.Name)
and node.exc.func.id == "PermissionError"
for node in ast.walk(evaluate)
)
print({
"kill_switch_literals": kill_paths,
"exists_call_count": len(exists_calls),
"evaluate_calls_check_kill_switch": calls_check,
"evaluate_raises_permission_error": raises_permission_error,
})
PY
stat -c 'tmp_mode=%a tmp_owner=%U tmp_group=%G' /tmpRepository: qdrant/qdrant-client Length of output: 410 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import os
import stat
st = os.stat("/tmp")
print({
"tmp_mode": oct(stat.S_IMODE(st.st_mode)),
"tmp_world_writable": bool(st.st_mode & stat.S_IWOTH),
"tmp_sticky_bit": bool(st.st_mode & stat.S_ISVTX),
"current_user_can_write_tmp": os.access("/tmp", os.W_OK),
})
PYRepository: qdrant/qdrant-client Length of output: 269 Do not use Any local process can create this world-writable marker and make every 🧰 Tools🪛 Ruff (0.16.1)[error] 107-107: Probable insecure usage of temporary file or directory: "/tmp/KILL" (S108) 🤖 Prompt for AI AgentsSource: Linters/SAST tools |
||
| return False | ||
|
|
||
| def evaluate_collection( | ||
| self, | ||
| collection_name: str, | ||
| raw_vector_bytes: int = 1000000000, | ||
| hnsw_index_bytes: int = 1050000000, | ||
| search_latency_ms: float = 18.5, | ||
| payload_fragmentation_count: int = 0, | ||
| un_gated_mutations: int = 0, | ||
| ) -> VectorDebtReport: | ||
| # 1. Evaluate emergency kill switch | ||
| if self.check_kill_switch(): | ||
| self.ledger.record_collection_event( | ||
| collection_name=collection_name, | ||
| event_type="collection_halted_kill_switch", | ||
| readiness_index=0.0, | ||
| critical_smells=["EMERGENCY_KILL_SWITCH_ENGAGED"], | ||
| metadata={"reason": "AAG_KILL_SWITCH is set"}, | ||
| ) | ||
| raise PermissionError( | ||
| "A2Z SOC ActionGate: Emergency kill switch is engaged. Vector collection operations halted." | ||
| ) | ||
|
|
||
| critical_smells: List[str] = [] | ||
|
|
||
| # KPI 2: Memory Multiplier | ||
| memory_ratio = hnsw_index_bytes / max(1, raw_vector_bytes) | ||
| if memory_ratio > 2.0: | ||
| critical_smells.append(f"HIGH_HNSW_MEMORY_SPRAWL_{memory_ratio:.2f}X") | ||
|
|
||
| # KPI 3: Latency Ceiling | ||
| if search_latency_ms > 80.0: | ||
| critical_smells.append(f"HIGH_SEARCH_LATENCY_{search_latency_ms:.1f}MS") | ||
|
|
||
| # Payload fragmentation | ||
| if payload_fragmentation_count > 2: | ||
| critical_smells.append(f"DETECTED_{payload_fragmentation_count}_FRAGMENTED_PAYLOAD_INDEXES") | ||
|
|
||
| # KPI 4: Mutation Safety | ||
| if un_gated_mutations > 0: | ||
| critical_smells.append(f"DETECTED_{un_gated_mutations}_UNGATED_COLLECTION_MUTATIONS") | ||
|
|
||
| # KPI 1: Vector Debt Index (0 = Clean, 100 = Catastrophic) | ||
| vdi = ( | ||
| max(0.0, (memory_ratio - 1.0) * 20.0) | ||
| + max(0.0, (search_latency_ms - 25.0) * 0.5) | ||
| + (payload_fragmentation_count * 12.0) | ||
| + (un_gated_mutations * 30.0) | ||
| ) | ||
| vdi_score = round(min(100.0, vdi), 2) | ||
|
|
||
| # Production Readiness Index (0 - 100) | ||
| readiness = max(0.0, 100.0 - vdi_score) | ||
| is_production_ready = ( | ||
| vdi_score <= self.max_acceptable_vdi and len(critical_smells) == 0 | ||
| ) | ||
|
Comment on lines
+137
to
+166
There was a problem hiding this comment. 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win Enforce the documented memory and latency limits before authorization. The gate authorizes collections above the stated KPI targets. For example, a Make 🤖 Prompt for AI Agents |
||
|
|
||
| # Cryptographic Ledger Entry | ||
| entry = self.ledger.record_collection_event( | ||
| collection_name=collection_name, | ||
| event_type="collection_authorized" if is_production_ready else "collection_flagged_debt", | ||
| readiness_index=readiness, | ||
| critical_smells=critical_smells, | ||
| metadata={ | ||
| "vdi_score": vdi_score, | ||
| "memory_ratio": memory_ratio, | ||
| "search_latency_ms": search_latency_ms, | ||
| "payload_fragmentation_count": payload_fragmentation_count, | ||
| "un_gated_mutations": un_gated_mutations, | ||
| "never_equate_intent_to_approval": self.never_equate_intent_to_approval, | ||
| }, | ||
| ) | ||
|
|
||
| return VectorDebtReport( | ||
| collection_name=collection_name, | ||
| vdi_score=vdi_score, | ||
| memory_multiplier=round(memory_ratio, 2), | ||
| search_latency_ms=round(search_latency_ms, 2), | ||
| mutation_safety_score=( | ||
| 100.0 if un_gated_mutations == 0 else max(0.0, 100.0 - un_gated_mutations * 30.0) | ||
| ), | ||
| production_readiness_index=readiness, | ||
| is_production_ready=is_production_ready, | ||
| critical_smells=critical_smells, | ||
| receipt_hash=entry["curr_hash"], | ||
| ) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| import importlib.util | ||
| import os | ||
| import sys | ||
| import unittest | ||
|
|
||
| # Load module directly | ||
| file_path = os.path.join( | ||
| os.path.dirname(__file__), | ||
| "../qdrant_client/production_debt.py", | ||
| ) | ||
| spec = importlib.util.spec_from_file_location("qdrant_production_debt", file_path) | ||
| production_debt_mod = importlib.util.module_from_spec(spec) | ||
| sys.modules["qdrant_production_debt"] = production_debt_mod | ||
| spec.loader.exec_module(production_debt_mod) | ||
|
|
||
| ProductionDebtVectorGate = production_debt_mod.ProductionDebtVectorGate | ||
| TechnicalDueDiligenceLedger = production_debt_mod.TechnicalDueDiligenceLedger | ||
| GENESIS_HASH = production_debt_mod.GENESIS_HASH | ||
|
|
||
|
|
||
| class TestProductionDebtVectorGate(unittest.TestCase): | ||
| def setUp(self) -> None: | ||
| self.gate = ProductionDebtVectorGate( | ||
| never_equate_intent_to_approval=True, | ||
| max_acceptable_vdi=12.0, | ||
| ) | ||
|
|
||
| def test_clean_collection_passes_readiness(self) -> None: | ||
| report = self.gate.evaluate_collection( | ||
| collection_name="enterprise_knowledge_base", | ||
| raw_vector_bytes=1000000000, | ||
| hnsw_index_bytes=1040000000, | ||
| search_latency_ms=18.5, | ||
| payload_fragmentation_count=0, | ||
| un_gated_mutations=0, | ||
| ) | ||
| self.assertTrue(report.is_production_ready) | ||
| self.assertLessEqual(report.vdi_score, 12.0) | ||
| self.assertEqual(len(report.critical_smells), 0) | ||
| self.assertTrue(bool(report.receipt_hash)) | ||
|
|
||
| def test_degraded_collection_fails_debt(self) -> None: | ||
| report = self.gate.evaluate_collection( | ||
| collection_name="unoptimized_payload_dump", | ||
| raw_vector_bytes=1000000000, | ||
| hnsw_index_bytes=3800000000, # High memory sprawl (3.8x) | ||
| search_latency_ms=120.0, # High latency | ||
| payload_fragmentation_count=4, # 4 fragmented indexes | ||
| un_gated_mutations=2, # 2 un-gated mutations | ||
| ) | ||
| self.assertFalse(report.is_production_ready) | ||
| self.assertGreater(report.vdi_score, 50.0) | ||
| self.assertIn("HIGH_HNSW_MEMORY_SPRAWL_3.80X", report.critical_smells) | ||
| self.assertIn("HIGH_SEARCH_LATENCY_120.0MS", report.critical_smells) | ||
| self.assertIn("DETECTED_4_FRAGMENTED_PAYLOAD_INDEXES", report.critical_smells) | ||
| self.assertIn("DETECTED_2_UNGATED_COLLECTION_MUTATIONS", report.critical_smells) | ||
|
|
||
| def test_cryptographic_ledger_integrity(self) -> None: | ||
| self.gate.evaluate_collection("coll-1") | ||
| self.gate.evaluate_collection("coll-2") | ||
| self.gate.evaluate_collection("coll-3") | ||
|
|
||
| entries = self.gate.ledger.get_ledger_entries() | ||
| self.assertEqual(len(entries), 3) | ||
| self.assertEqual(entries[0]["prev_hash"], GENESIS_HASH) | ||
| self.assertEqual(entries[1]["prev_hash"], entries[0]["curr_hash"]) | ||
| self.assertEqual(entries[2]["prev_hash"], entries[1]["curr_hash"]) | ||
| self.assertTrue(self.gate.ledger.verify_ledger_integrity()) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Make ledger integrity validate event content.
verify_ledger_integrityonly checksprev_hash. It does not recomputecurr_hash. Also,critical_smellsis not part of the hashed content, andget_ledger_entries()exposes mutable nested dictionaries and lists. A caller can modify an entry, including throughreport.critical_smells, and verification still returnsTrue.qdrant_client/production_debt.py#L52-L81: hash all protected entry fields, recompute each hash during verification, and return immutable or deep-copied entry snapshots.tests/test_production_debt.py#L63-L68: mutatemetadataandcritical_smellsin a retrieved entry, then assert thatverify_ledger_integrity()returnsFalse.📍 Affects 2 files
qdrant_client/production_debt.py#L52-L81(this comment)tests/test_production_debt.py#L63-L68🤖 Prompt for AI Agents