requirements.txt
jsonschema>=4.20
fastapi>=0.115
uvicorn>=0.30
qwp_reference_v1.py
from __future__ import annotations
import argparse
import copy
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Tuple
try:
from jsonschema import Draft202012Validator
except ImportError:
Draft202012Validator = None
# ============================================================
# QWP VERSION
# ============================================================
QWP_VERSION = "1.0"
ENGINE_VERSION = "1.0.0-rc2"
# ============================================================
# CONTROLLED VALUES
# ============================================================
REQUIREMENT_TYPES = {
"MANDATORY",
"COMPENSABLE",
"INTEGRATIVE",
}
REQUIREMENT_STATUSES = {
"SUPPORTED",
"NOT_SUPPORTED",
"INSUFFICIENT_EVIDENCE",
"CONTESTED",
"HUMAN_REVIEW",
}
DECISIONS = {
"PASS",
"FAIL",
"INDETERMINATE",
}
PROOF_STATUSES = {
"DRAFT",
"READY_FOR_EVALUATION",
"VALIDATED",
"EVALUATED",
"SUPERSEDED",
"REVOKED",
"INVALID",
}
AUTHORITY_STATUSES = {
"ACTIVE",
"EXPIRED",
"SUSPENDED",
"REVOKED",
"UNVERIFIED",
}
VALIDITY_STATUSES = {
"SUPPORTED",
"PARTIALLY_SUPPORTED",
"NOT_SUPPORTED",
"INSUFFICIENT_BASIS",
"CONTESTED",
}
# ============================================================
# ASSURANCE LEVELS
# ============================================================
IA_LEVELS = {
"IA-0": 0,
"IA-1": 1,
"IA-2": 2,
"IA-3": 3,
"IA-4": 4,
}
PA_LEVELS = {
"PA-0": 0,
"PA-1": 1,
"PA-2": 2,
"PA-3": 3,
"PA-4": 4,
}
# ============================================================
# HUMAN JUDGMENT
# ============================================================
HUMAN_JUDGMENT_MAP = {
"SUPPORTED": "SUPPORTED",
"NOT_SUPPORTED": "NOT_SUPPORTED",
"CONTESTED": "CONTESTED",
"DEFERRED": "HUMAN_REVIEW",
"REASSESSMENT_REQUIRED": "HUMAN_REVIEW",
}
# ============================================================
# QWP PROFILES
# ============================================================
PROFILES = {
"QWP-CORE": {
"min_identity": "IA-0",
"min_process": "PA-0",
"authority_required": False,
"human_review_for_integrative": False,
},
"QWP-RPL": {
"min_identity": "IA-2",
"min_process": "PA-2",
"authority_required": True,
"human_review_for_integrative": True,
},
"QWP-HIGH-STAKES": {
"min_identity": "IA-3",
"min_process": "PA-3",
"authority_required": True,
"human_review_for_integrative": True,
},
}
# ============================================================
# JSON SCHEMAS
# ============================================================
SPEC_SCHEMA = {
"$schema":
"https://json-schema.org/draft/2020-12/schema",
"type":
"object",
"required": [
"id",
"type",
"version",
"title",
"requirements",
"decisionRuleSet",
],
"properties": {
"id": {
"type": "string",
"minLength": 1,
},
"type": {
"const": "QualificationSpecification",
},
"version": {
"type": "string",
"minLength": 1,
},
"title": {
"type": "string",
"minLength": 1,
},
"requirements": {
"type": "array",
"minItems": 1,
},
"decisionRuleSet": {
"type": "object",
},
},
}
PROOF_SCHEMA = {
"$schema":
"https://json-schema.org/draft/2020-12/schema",
"type":
"object",
"required": [
"id",
"type",
"version",
"qwpVersion",
"subject",
"qualificationSpecification",
"proofStatus",
"evidence",
],
"properties": {
"id": {
"type": "string",
"minLength": 1,
},
"type": {
"const": "QualificationProof",
},
"version": {
"type": "string",
"minLength": 1,
},
"qwpVersion": {
"type": "string",
"minLength": 1,
},
"subject": {
"type": "string",
"minLength": 1,
},
"qualificationSpecification": {
"type": "string",
"minLength": 1,
},
"proofStatus": {
"enum": sorted(PROOF_STATUSES),
},
"evidence": {
"type": "array",
},
},
}
# ============================================================
# BASIC UTILITIES
# ============================================================
def utcnow() -> str:
return datetime.now(
timezone.utc
).isoformat()
def canonical_json(value: Any) -> str:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
def sha256_of(value: Any) -> str:
return hashlib.sha256(
canonical_json(value)
.encode("utf-8")
).hexdigest()
def parse_major(version: str) -> str:
return str(version).split(
".",
1
)[0]
# ============================================================
# OBJECT WALKING / IDENTIFIERS
# ============================================================
def walk_dicts(value: Any):
if isinstance(value, dict):
yield value
for child in value.values():
yield from walk_dicts(child)
elif isinstance(value, list):
for child in value:
yield from walk_dicts(child)
def collect_ids(
*documents: Dict[str, Any]
) -> Tuple[
Dict[str, Dict[str, Any]],
List[str]
]:
index = {}
duplicates = []
for document in documents:
for obj in walk_dicts(document):
object_id = obj.get("id")
if not isinstance(
object_id,
str
):
continue
if (
object_id in index
and index[object_id]
is not obj
):
duplicates.append(
object_id
)
else:
index[object_id] = obj
return (
index,
sorted(set(duplicates))
)
# ============================================================
# JSON SCHEMA VALIDATION
# ============================================================
def schema_errors(
instance: Dict[str, Any],
schema: Dict[str, Any],
) -> List[Dict[str, Any]]:
if Draft202012Validator is None:
return [{
"code":
"QWP_JSONSCHEMA_UNAVAILABLE",
"message":
(
"Install 'jsonschema' "
"for Draft 2020-12 "
"runtime validation."
),
"severity":
"warning",
}]
validator = Draft202012Validator(
schema
)
output = []
for error in sorted(
validator.iter_errors(
instance
),
key=lambda e: list(e.path),
):
output.append({
"code":
"QWP_SCHEMA_VALIDATION_FAILED",
"path":
"/".join(
str(x)
for x
in error.absolute_path
),
"message":
error.message,
"severity":
"error",
})
return output
# ============================================================
# SPECIFICATION INDEXES
# ============================================================
def build_spec_indexes(
specification: Dict[str, Any]
):
requirements = {}
claims = {}
for requirement in specification.get(
"requirements",
[]
):
if (
isinstance(requirement, dict)
and isinstance(
requirement.get("id"),
str
)
):
requirements[
requirement["id"]
] = requirement
for claim in requirement.get(
"claimObjects",
[]
):
if (
isinstance(claim, dict)
and isinstance(
claim.get("id"),
str
)
):
claims[
claim["id"]
] = claim
for claim in specification.get(
"claims",
[]
):
if (
isinstance(claim, dict)
and isinstance(
claim.get("id"),
str
)
):
claims[
claim["id"]
] = claim
return requirements, claims
# ============================================================
# ASSURANCE HELPERS
# ============================================================
def find_identity_level(
proof: Dict[str, Any],
subject: str,
) -> int:
level = 0
for record in proof.get(
"identityAssurance",
[]
):
if record.get(
"subject"
) == subject:
level = max(
level,
IA_LEVELS.get(
record.get(
"level",
"IA-0"
),
0
)
)
return level
def find_process_level(
proof: Dict[str, Any],
assessment_id: str,
) -> int:
level = 0
for record in proof.get(
"processAssurance",
[]
):
if record.get(
"assessmentEvent"
) == assessment_id:
level = max(
level,
PA_LEVELS.get(
record.get(
"level",
"PA-0"
),
0
)
)
return level
# ============================================================
# AUTHORITY
# ============================================================
def active_authority_for(
proof: Dict[str, Any],
assessor: str,
scopes: List[str],
evaluation_time: str,
) -> bool:
date = evaluation_time[:10]
for record in proof.get(
"authorityRecords",
[]
):
if record.get(
"holder"
) != assessor:
continue
if record.get(
"status"
) != "ACTIVE":
continue
valid_from = record.get(
"validFrom"
)
valid_until = record.get(
"validUntil"
)
if (
valid_from
and date < valid_from[:10]
):
continue
if (
valid_until
and date > valid_until[:10]
):
continue
declared_scopes = set(
record.get(
"scope",
[]
)
)
if (
not scopes
or declared_scopes
.intersection(scopes)
):
return True
return False
# ============================================================
# VALIDATE
# ============================================================
def validate(
specification: Dict[str, Any],
proof: Dict[str, Any],
profile: str = "QWP-CORE",
) -> Dict[str, Any]:
errors = []
warnings = []
if profile not in PROFILES:
errors.append({
"code":
"QWP_UNKNOWN_PROFILE",
"profile":
profile,
})
# ----------------------------------------
# JSON Schema
# ----------------------------------------
for error in schema_errors(
specification,
SPEC_SCHEMA
):
if (
error.get("severity")
== "warning"
):
warnings.append(error)
else:
errors.append(error)
for error in schema_errors(
proof,
PROOF_SCHEMA
):
if (
error.get("severity")
== "warning"
):
warnings.append(error)
else:
errors.append(error)
# ----------------------------------------
# Protocol version
# ----------------------------------------
if (
parse_major(
proof.get(
"qwpVersion",
"0"
)
)
!=
parse_major(
QWP_VERSION
)
):
errors.append({
"code":
"QWP_UNSUPPORTED_MAJOR_VERSION",
"documentVersion":
proof.get(
"qwpVersion"
),
"engineVersion":
QWP_VERSION,
})
# ----------------------------------------
# Specification link
# ----------------------------------------
if (
proof.get(
"qualificationSpecification"
)
!=
specification.get("id")
):
errors.append({
"code":
"QWP_SPECIFICATION_MISMATCH",
"proofValue":
proof.get(
"qualificationSpecification"
),
"expected":
specification.get("id"),
})
# ----------------------------------------
# Duplicate identifiers
# ----------------------------------------
_, duplicates = collect_ids(
specification,
proof
)
if duplicates:
errors.append({
"code":
"QWP_DUPLICATE_IDENTIFIER",
"identifiers":
duplicates,
})
# ----------------------------------------
# Specification references
# ----------------------------------------
requirements, claims = (
build_spec_indexes(
specification
)
)
for requirement_id, requirement \
in requirements.items():
if requirement.get(
"requirementType"
) not in REQUIREMENT_TYPES:
errors.append({
"code":
"QWP_INVALID_REQUIREMENT_TYPE",
"requirement":
requirement_id,
"value":
requirement.get(
"requirementType"
),
})
for claim_id in requirement.get(
"claims",
[]
):
if claim_id not in claims:
errors.append({
"code":
"QWP_DANGLING_CLAIM_REFERENCE",
"requirement":
requirement_id,
"claim":
claim_id,
})
# ----------------------------------------
# Evidence references
# ----------------------------------------
for evidence in proof.get(
"evidence",
[]
):
if not isinstance(
evidence,
dict
):
errors.append({
"code":
"QWP_INVALID_EVIDENCE_OBJECT"
})
continue
for claim_id in evidence.get(
"supports",
[]
):
if claim_id not in claims:
errors.append({
"code":
"QWP_DANGLING_CLAIM_REFERENCE",
"evidence":
evidence.get("id"),
"claim":
claim_id,
})
# ----------------------------------------
# Validity arguments
# ----------------------------------------
for argument in proof.get(
"validityArguments",
[]
):
target = argument.get(
"targetClaim"
)
if (
target
and target not in claims
):
errors.append({
"code":
"QWP_DANGLING_VALIDITY_TARGET",
"validityArgument":
argument.get("id"),
"claim":
target,
})
if argument.get(
"status"
) not in VALIDITY_STATUSES:
errors.append({
"code":
"QWP_INVALID_VALIDITY_STATUS",
"validityArgument":
argument.get("id"),
"status":
argument.get("status"),
})
# ----------------------------------------
# Human judgments
# ----------------------------------------
for judgment in proof.get(
"humanJudgments",
[]
):
if judgment.get(
"target"
) not in requirements:
errors.append({
"code":
"QWP_DANGLING_HUMAN_JUDGMENT_TARGET",
"humanJudgment":
judgment.get("id"),
"target":
judgment.get("target"),
})
# ----------------------------------------
# Extensions
# ----------------------------------------
for namespace, extension \
in proof.get(
"extensions",
{}
).items():
if (
isinstance(
extension,
dict
)
and
extension.get(
"critical"
) is True
):
warnings.append({
"code":
"QWP_UNKNOWN_CRITICAL_EXTENSION",
"namespace":
namespace,
})
# ----------------------------------------
# Proof status
# ----------------------------------------
if proof.get(
"proofStatus"
) in {
"REVOKED",
"INVALID",
"SUPERSEDED",
}:
warnings.append({
"code":
"QWP_PROOF_STATUS_NOT_ACTIVE",
"status":
proof.get(
"proofStatus"
),
})
valid = not errors
return {
"valid":
valid,
"technicalStatus":
(
"VALID"
if valid
else "INVALID_PROOF"
),
"validationLevel":
"V3",
"profile":
profile,
"errors":
errors,
"warnings":
warnings,
"documentHash":
sha256_of(proof),
}
# ============================================================
# RUNTIME INDEXES
# ============================================================
def build_runtime_indexes(
specification: Dict[str, Any],
proof: Dict[str, Any],
):
requirements, claims = (
build_spec_indexes(
specification
)
)
evidence_by_claim = {}
validity_by_claim = {}
judgments_by_requirement = {}
assessments = {
assessment.get("id"):
assessment
for assessment
in proof.get(
"assessmentEvents",
[]
)
if isinstance(
assessment,
dict
)
}
for evidence in proof.get(
"evidence",
[]
):
if evidence.get(
"status",
"ACTIVE"
) in {
"INVALID",
"REVOKED",
}:
continue
for claim_id in evidence.get(
"supports",
[]
):
evidence_by_claim.setdefault(
claim_id,
[]
).append(evidence)
for argument in proof.get(
"validityArguments",
[]
):
validity_by_claim.setdefault(
argument.get(
"targetClaim"
),
[]
).append(argument)
for judgment in proof.get(
"humanJudgments",
[]
):
judgments_by_requirement.setdefault(
judgment.get(
"target"
),
[]
).append(judgment)
return (
requirements,
claims,
evidence_by_claim,
validity_by_claim,
judgments_by_requirement,
assessments,
)
# ============================================================
# STATUS AGGREGATION
# ============================================================
def aggregate_statuses(
statuses: List[str]
) -> str:
if not statuses:
return "INSUFFICIENT_EVIDENCE"
if "CONTESTED" in statuses:
return "CONTESTED"
if "NOT_SUPPORTED" in statuses:
return "NOT_SUPPORTED"
if "HUMAN_REVIEW" in statuses:
return "HUMAN_REVIEW"
if "INSUFFICIENT_EVIDENCE" \
in statuses:
return "INSUFFICIENT_EVIDENCE"
if all(
status == "SUPPORTED"
for status in statuses
):
return "SUPPORTED"
return "INSUFFICIENT_EVIDENCE"
# ============================================================
# CLAIM EVALUATION
# ============================================================
def claim_status(
claim_id: str,
proof: Dict[str, Any],
evidence_by_claim,
validity_by_claim,
assessments,
profile_name: str,
evaluation_time: str,
requirement_scope: List[str],
):
evidence = evidence_by_claim.get(
claim_id,
[]
)
if not evidence:
return (
"INSUFFICIENT_EVIDENCE",
{
"reason":
"NO_EVIDENCE"
}
)
validity_arguments = (
validity_by_claim.get(
claim_id,
[]
)
)
statuses = {
argument.get("status")
for argument
in validity_arguments
}
if "CONTESTED" in statuses:
return (
"CONTESTED",
{
"reason":
"VALIDITY_CONTESTED"
}
)
if "NOT_SUPPORTED" in statuses:
return (
"NOT_SUPPORTED",
{
"reason":
"VALIDITY_NOT_SUPPORTED"
}
)
if (
"INSUFFICIENT_BASIS"
in statuses
or not validity_arguments
):
return (
"INSUFFICIENT_EVIDENCE",
{
"reason":
"VALIDITY_INSUFFICIENT"
}
)
if "SUPPORTED" not in statuses:
return (
"INSUFFICIENT_EVIDENCE",
{
"reason":
"VALIDITY_NOT_FULLY_SUPPORTED"
}
)
profile = PROFILES[
profile_name
]
subject = proof["subject"]
# ----------------------------------------
# Identity assurance
# ----------------------------------------
actual_identity = (
find_identity_level(
proof,
subject
)
)
required_identity = (
IA_LEVELS[
profile["min_identity"]
]
)
if (
actual_identity
<
required_identity
):
return (
"HUMAN_REVIEW",
{
"reason":
"IDENTITY_ASSURANCE_INSUFFICIENT"
}
)
# ----------------------------------------
# Evidence-linked assessments
# ----------------------------------------
cited_assessments = []
for evidence_object in evidence:
for assessment_id \
in evidence_object.get(
"assessmentEvents",
[]
):
if assessment_id in assessments:
cited_assessments.append(
assessments[
assessment_id
]
)
# ----------------------------------------
# Process assurance / authority
# ----------------------------------------
for assessment in cited_assessments:
assessment_id = (
assessment.get("id")
)
actual_process = (
find_process_level(
proof,
assessment_id
)
)
required_process = (
PA_LEVELS[
profile["min_process"]
]
)
if (
actual_process
<
required_process
):
return (
"HUMAN_REVIEW",
{
"reason":
"PROCESS_ASSURANCE_INSUFFICIENT",
"assessment":
assessment_id,
}
)
if profile[
"authority_required"
]:
assessors = assessment.get(
"assessors",
[]
)
if not assessors:
return (
"HUMAN_REVIEW",
{
"reason":
"ASSESSOR_MISSING",
"assessment":
assessment_id,
}
)
authority_valid = all(
active_authority_for(
proof,
assessor,
requirement_scope,
evaluation_time,
)
for assessor
in assessors
)
if not authority_valid:
return (
"HUMAN_REVIEW",
{
"reason":
"ASSESSOR_AUTHORITY_UNVERIFIED",
"assessment":
assessment_id,
}
)
return (
"SUPPORTED",
{
"reason":
"EVIDENCE_AND_VALIDITY_SUPPORTED",
"evidence": [
item.get("id")
for item in evidence
],
}
)
# ============================================================
# REQUIREMENT EVALUATION
# ============================================================
def requirement_result(
requirement,
proof,
evidence_by_claim,
validity_by_claim,
judgments_by_requirement,
assessments,
profile_name,
evaluation_time,
):
requirement_id = (
requirement["id"]
)
judgments = (
judgments_by_requirement.get(
requirement_id,
[]
)
)
# ----------------------------------------
# Integrative human review
# ----------------------------------------
if (
requirement.get(
"requirementType"
)
== "INTEGRATIVE"
and
PROFILES[
profile_name
][
"human_review_for_integrative"
]
and
not judgments
):
return {
"requirement":
requirement_id,
"requirementType":
"INTEGRATIVE",
"status":
"HUMAN_REVIEW",
"basis":
"PROFILE_REQUIRES_HUMAN_REVIEW",
"reason":
"INTEGRATIVE_HUMAN_REVIEW_REQUIRED",
}
# ----------------------------------------
# Explicit human judgment
# ----------------------------------------
if judgments:
mapped = [
HUMAN_JUDGMENT_MAP.get(
judgment.get(
"decision"
),
"HUMAN_REVIEW"
)
for judgment
in judgments
]
return {
"requirement":
requirement_id,
"requirementType":
requirement.get(
"requirementType",
"MANDATORY"
),
"status":
aggregate_statuses(
mapped
),
"basis":
"HUMAN_JUDGMENT",
"judgments": [
judgment.get("id")
for judgment
in judgments
],
}
# ----------------------------------------
# Claim evaluation
# ----------------------------------------
claim_results = []
for claim_id in requirement.get(
"claims",
[]
):
status, detail = claim_status(
claim_id,
proof,
evidence_by_claim,
validity_by_claim,
assessments,
profile_name,
evaluation_time,
requirement.get(
"authorityScopes",
[requirement_id]
),
)
claim_results.append({
"claim":
claim_id,
"status":
status,
**detail,
})
return {
"requirement":
requirement_id,
"requirementType":
requirement.get(
"requirementType",
"MANDATORY"
),
"status":
aggregate_statuses(
[
result["status"]
for result
in claim_results
]
),
"basis":
"CLAIM_EVIDENCE_CHAIN",
"claims":
claim_results,
}
# ============================================================
# FINAL DECISION
# ============================================================
def final_decision(
specification,
requirement_results,
):
hard_requirements = [
result
for result
in requirement_results
if result["requirementType"]
in {
"MANDATORY",
"INTEGRATIVE",
}
]
compensable_requirements = [
result
for result
in requirement_results
if result["requirementType"]
== "COMPENSABLE"
]
# ----------------------------------------
# Hard failure
# ----------------------------------------
if any(
result["status"]
== "NOT_SUPPORTED"
for result
in hard_requirements
):
return "FAIL"
# ----------------------------------------
# Hard uncertainty
# ----------------------------------------
if any(
result["status"]
in {
"INSUFFICIENT_EVIDENCE",
"CONTESTED",
"HUMAN_REVIEW",
}
for result
in hard_requirements
):
return "INDETERMINATE"
# ----------------------------------------
# Compensable requirements
# ----------------------------------------
if compensable_requirements:
rules = (
specification
.get(
"decisionRuleSet",
{}
)
.get(
"rules",
{}
)
)
minimum_ratio = float(
rules.get(
"compensableMinimumSupportedRatio",
1.0
)
)
maximum_failed = int(
rules.get(
"compensableMaxNotSupported",
0
)
)
supported_count = sum(
result["status"]
== "SUPPORTED"
for result
in compensable_requirements
)
failed_count = sum(
result["status"]
== "NOT_SUPPORTED"
for result
in compensable_requirements
)
unresolved = any(
result["status"]
in {
"INSUFFICIENT_EVIDENCE",
"CONTESTED",
"HUMAN_REVIEW",
}
for result
in compensable_requirements
)
if (
failed_count
>
maximum_failed
):
return "FAIL"
ratio = (
supported_count
/
len(
compensable_requirements
)
)
if ratio >= minimum_ratio:
return "PASS"
if unresolved:
return "INDETERMINATE"
return "FAIL"
return "PASS"
# ============================================================
# EVALUATE
# ============================================================
def evaluate(
specification,
proof,
profile="QWP-CORE",
evaluation_time=None,
):
evaluation_time = (
evaluation_time
or utcnow()
)
validation = validate(
specification,
proof,
profile
)
if not validation["valid"]:
return {
"technicalStatus":
"INVALID_PROOF",
"decision":
None,
"validation":
validation,
}
# ----------------------------------------
# Unknown critical extension
# ----------------------------------------
if any(
warning["code"]
==
"QWP_UNKNOWN_CRITICAL_EXTENSION"
for warning
in validation["warnings"]
):
return {
"technicalStatus":
"VALID",
"decision":
"INDETERMINATE",
"reason":
"UNKNOWN_CRITICAL_EXTENSION",
"validation":
validation,
"limitations": [
(
"Unknown critical extension "
"prevents authoritative PASS."
)
],
}
(
requirements,
claims,
evidence_by_claim,
validity_by_claim,
judgments_by_requirement,
assessments,
) = build_runtime_indexes(
specification,
proof
)
requirement_results = [
requirement_result(
requirement,
proof,
evidence_by_claim,
validity_by_claim,
judgments_by_requirement,
assessments,
profile,
evaluation_time,
)
for requirement
in requirements.values()
]
decision = final_decision(
specification,
requirement_results,
)
decision_seed = {
"proof":
proof["id"],
"specification":
specification["id"],
"evaluationTime":
evaluation_time,
}
decision_id = (
"urn:qwp:decision:"
+
sha256_of(
decision_seed
)[:24]
)
decision_object = {
"id":
decision_id,
"type":
"ConformanceDecision",
"qwpVersion":
QWP_VERSION,
"engineVersion":
ENGINE_VERSION,
"profile":
profile,
"proof":
proof["id"],
"proofVersion":
proof["version"],
"qualificationSpecification":
specification["id"],
"specificationVersion":
specification["version"],
"decisionRulesVersion":
specification
.get(
"decisionRuleSet",
{}
)
.get("version"),
"evaluationTime":
evaluation_time,
"decision":
decision,
"requirementResults":
requirement_results,
"decisionTrace": {
"type":
"DecisionTrace",
"steps":
copy.deepcopy(
requirement_results
),
},
"limitations": [
"Academic conformance only.",
(
"No legal award, "
"recognition, or "
"professional authorization "
"is inferred."
),
],
}
decision_object[
"decisionHash"
] = sha256_of(
decision_object
)
return {
"technicalStatus":
"VALID",
**decision_object,
}
# ============================================================
# EXPLAIN
# ============================================================
def explain(
specification,
proof,
profile="QWP-CORE",
):
result = evaluate(
specification,
proof,
profile
)
if (
result["technicalStatus"]
==
"INVALID_PROOF"
):
return {
"technicalStatus":
"INVALID_PROOF",
"summary":
(
"Доказовий пакет "
"некоректний і не може "
"бути змістовно оцінений."
),
"errors":
result[
"validation"
][
"errors"
],
}
if (
result.get("reason")
==
"UNKNOWN_CRITICAL_EXTENSION"
):
return {
"technicalStatus":
"VALID",
"decision":
"INDETERMINATE",
"summary":
(
"Оцінювання не завершено: "
"реалізація не розуміє "
"критичне розширення."
),
"limitations":
result.get(
"limitations",
[]
),
}
requirement_labels = {
"SUPPORTED":
"підтримано доказами",
"NOT_SUPPORTED":
"не підтримано доказами",
"INSUFFICIENT_EVIDENCE":
"недостатньо доказів",
"CONTESTED":
"стан оспорюється",
"HUMAN_REVIEW":
"потрібен експертний перегляд",
}
decision_labels = {
"PASS":
(
"академічна відповідність "
"підтверджена"
),
"FAIL":
(
"академічна відповідність "
"не підтверджена"
),
"INDETERMINATE":
(
"остаточний висновок "
"поки неможливий"
),
}
lines = [
(
f"{result['decision']}: "
f"{decision_labels[result['decision']]}."
)
]
for requirement in result[
"requirementResults"
]:
status = (
requirement["status"]
)
lines.append(
f"{requirement['requirement']}: "
f"{status} — "
f"{requirement_labels[status]}."
)
return {
"technicalStatus":
"VALID",
"decision":
result["decision"],
"summary":
" ".join(lines),
"decisionId":
result["id"],
"decisionTrace":
result["decisionTrace"],
"limitations":
result["limitations"],
}
# ============================================================
# GAPS
# ============================================================
def gaps(
specification,
proof,
profile="QWP-CORE",
):
result = evaluate(
specification,
proof,
profile
)
if (
result["technicalStatus"]
==
"INVALID_PROOF"
):
return {
"technicalStatus":
"INVALID_PROOF",
"gaps":
[],
"errors":
result[
"validation"
][
"errors"
],
}
if "requirementResults" \
not in result:
return {
"technicalStatus":
"VALID",
"currentDecision":
"INDETERMINATE",
"gaps": [{
"gapType":
"UNKNOWN_CRITICAL_EXTENSION"
}],
}
next_steps = {
"NOT_SUPPORTED": (
"REQUIREMENT_NOT_SUPPORTED",
(
"Надати нові докази "
"або пройти повторне "
"допустиме оцінювання."
),
),
"INSUFFICIENT_EVIDENCE": (
"MISSING_OR_INSUFFICIENT_EVIDENCE",
(
"Додати допустимі докази "
"та, за потреби, "
"незалежне оцінювання."
),
),
"CONTESTED": (
"CONTEST_UNRESOLVED",
(
"Провести процедуру "
"вирішення розбіжностей "
"або додаткову "
"незалежну оцінку."
),
),
"HUMAN_REVIEW": (
"HUMAN_REVIEW_PENDING",
(
"Завершити експертний "
"людський перегляд."
),
),
}
output = []
for requirement in result[
"requirementResults"
]:
status = (
requirement["status"]
)
if status == "SUPPORTED":
continue
gap_type, next_step = (
next_steps[status]
)
output.append({
"requirement":
requirement[
"requirement"
],
"status":
status,
"gapType":
gap_type,
"recommendedNextStep":
next_step,
})
return {
"technicalStatus":
"VALID",
"currentDecision":
result["decision"],
"proof":
proof["id"],
"qualificationSpecification":
specification["id"],
"gaps":
output,
}
# ============================================================
# AUDIT TRAIL
# ============================================================
def audit_event(
operation,
specification,
proof,
result,
profile,
):
event = {
"type":
"QWPAuditEvent",
"timestamp":
utcnow(),
"operation":
operation,
"qwpVersion":
QWP_VERSION,
"engineVersion":
ENGINE_VERSION,
"profile":
profile,
"specification":
specification.get("id"),
"specificationVersion":
specification.get(
"version"
),
"proof":
proof.get("id"),
"proofVersion":
proof.get("version"),
"proofHash":
sha256_of(proof),
"resultTechnicalStatus":
result.get(
"technicalStatus"
),
"resultDecision":
result.get(
"decision"
),
}
event[
"auditHash"
] = sha256_of(event)
return event
# ============================================================
# UNIFIED OPERATION
# ============================================================
def perform(
operation,
specification,
proof,
profile="QWP-CORE",
):
operations = {
"validate":
validate,
"evaluate":
evaluate,
"explain":
explain,
"gaps":
gaps,
}
if operation not in operations:
raise ValueError(
f"Unsupported operation: "
f"{operation}"
)
result = operations[
operation
](
specification,
proof,
profile
)
return {
"result":
result,
"audit":
audit_event(
operation,
specification,
proof,
result,
profile,
),
}
# ============================================================
# FILE LOADING
# ============================================================
def load_json(
path: str
) -> Dict[str, Any]:
return json.loads(
Path(path)
.read_text(
encoding="utf-8"
)
)
# ============================================================
# CLI
# ============================================================
def cli():
parser = argparse.ArgumentParser(
description=(
"QWP v1.0 Candidate "
"Reference Implementation"
)
)
parser.add_argument(
"operation",
choices=[
"validate",
"evaluate",
"explain",
"gaps",
]
)
parser.add_argument(
"specification"
)
parser.add_argument(
"proof"
)
parser.add_argument(
"--profile",
default="QWP-CORE",
choices=sorted(
PROFILES
)
)
args = parser.parse_args()
specification = load_json(
args.specification
)
proof = load_json(
args.proof
)
result = perform(
args.operation,
specification,
proof,
args.profile,
)
print(
json.dumps(
result,
ensure_ascii=False,
indent=2,
)
)
if __name__ == "__main__":
cli()
qwp_api.py
Цей файл створює REST API поверх того
самого QWP engine. Тобто сайт, зовнішня програма або AI-агент не повинні реалізовувати логіку QWP самостійно.
from typing import Any, Dict
from fastapi import (
FastAPI,
HTTPException,
)
from pydantic import BaseModel
from qwp_reference_v1 import (
QWP_VERSION,
ENGINE_VERSION,
PROFILES,
perform,
)
app = FastAPI(
title=(
"Qualification Web Protocol "
"Reference API"
),
version=ENGINE_VERSION,
description=(
"Reference REST API "
"for QWP v1.0 Candidate."
),
)
class EvaluationRequest(
BaseModel
):
specification: Dict[str, Any]
proof: Dict[str, Any]
profile: str = "QWP-CORE"
def run(
operation: str,
body: EvaluationRequest,
):
if body.profile not in PROFILES:
raise HTTPException(
status_code=400,
detail=(
"Unknown QWP profile."
),
)
return perform(
operation,
body.specification,
body.proof,
body.profile,
)
@app.get(
"/health"
)
def health():
return {
"status":
"ok",
"qwpVersion":
QWP_VERSION,
"engineVersion":
ENGINE_VERSION,
}
@app.get(
"/v1/profiles"
)
def profiles():
return {
"profiles":
PROFILES
}
@app.post(
"/v1/proof/validate"
)
def validate_endpoint(
body: EvaluationRequest
):
return run(
"validate",
body
)
@app.post(
"/v1/proof/evaluate"
)
def evaluate_endpoint(
body: EvaluationRequest
):
return run(
"evaluate",
body
)
@app.post(
"/v1/proof/explain"
)
def explain_endpoint(
body: EvaluationRequest
):
return run(
"explain",
body
)
@app.post(
"/v1/proof/gaps"
)
def gaps_endpoint(
body: EvaluationRequest
):
return run(
"gaps",
body
)
Запуск
pip install -r requirements.txt
CLI:
python qwp_reference_v1.py evaluate specification.json proof.json
Для RPL:
python qwp_reference_v1.py evaluate specification.json proof.json --profile QWP-RPL
Для high-stakes:
python qwp_reference_v1.py evaluate specification.json proof.json --profile QWP-HIGH-STAKES
REST API:
uvicorn qwp_api:app --reload
Після запуску:
GET /health
GET /v1/profiles
POST /v1/proof/validate
POST /v1/proof/evaluate
POST /v1/proof/explain
POST /v1/proof/gaps
Наприклад запит до /v1/proof/evaluate:
{
"profile": "QWP-RPL",
"specification": {
"id": "qualification:example-master",
"type": "QualificationSpecification",
"version": "1.0",
"title": "Example Qualification",
"requirements": [],
"decisionRuleSet": {
"id": "rules:1",
"version": "1.0"
}
},
"proof": {
"id": "proof:001",
"type": "QualificationProof",
"qwpVersion": "1.0",
"version": "1.0",
"subject": "candidate:001",
"qualificationSpecification":
"qualification:example-master",
"proofStatus":
"READY_FOR_EVALUATION",
"evidence": []
}
}
Що тепер принципово змінило
Reference Implementation тепер має три режими строгості:
QWP-CORE
— мінімальна загальна реалізація.
QWP-RPL
— для визнання попереднього навчання; вже вимагає сильнішого підтвердження особи, процесу та повноважень оцінювача.
QWP-HIGH-STAKES
— для рішень із високими наслідками; пороги ще суворіші, а інтегративні вимоги можуть вимагати людського експертного рішення.
Також з'явився важливий механізм:
proof
→ SHA-256 hash
decision
→ SHA-256 hash
operation
→ audit event
Це означає, що надалі можна встановити який саме доказовий пакет був оцінений, якою версією engine, за яким профілем і який результат було отримано.
При цьому базовий інваріант не змінено:
PASS
!=
Legal Award
!=
Recognition
!=
Professional Authorization
Синтаксис основного Python-файлу та REST API я перевірив; повне систематичне тестування поведінки спеціально стає наступним окремим артефактом