feat: advance TWM simulator validation roadmap

Author: zhouning

data_agent/api/territory_world_model_routes.py

@@ -525,6 +525,26 @@ async def twm_geofm_ablation_gate(request: Request):
         return JSONResponse({"error": str(exc)}, status_code=400)
 
 
+async def twm_geofm_downstream_experiment_report(request: Request):
+    user = _get_user_from_request(request)
+    if not user:
+        return JSONResponse({"error": "Unauthorized"}, status_code=401)
+    _set_user_context(user)
+    svc = get_territory_world_model_service()
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    if not isinstance(body, dict):
+        return JSONResponse({"error": "JSON body must be an object"}, status_code=400)
+    try:
+        return JSONResponse(svc.geofm_downstream_experiment_report(request.path_params["id"], body))
+    except LookupError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=404)
+    except Exception as exc:
+        return JSONResponse({"error": str(exc)}, status_code=400)
+
+
 async def twm_causal_calibration_report(request: Request):
     user = _get_user_from_request(request)
     if not user:
@@ -573,6 +593,7 @@ def get_territory_world_model_routes() -> list[Route]:
         Route("/api/twm/states/{id}/fit-dynamics-candidate", endpoint=twm_fit_dynamics_candidate, methods=["POST"]),
         Route("/api/twm/states/{id}/train-dynamics-candidate", endpoint=twm_train_dynamics_candidate, methods=["POST"]),
         Route("/api/twm/states/{id}/geofm-ablation-gate", endpoint=twm_geofm_ablation_gate, methods=["POST"]),
+        Route("/api/twm/states/{id}/geofm-downstream-experiment-report", endpoint=twm_geofm_downstream_experiment_report, methods=["POST"]),
         Route("/api/twm/states/{id}/causal-calibration-report", endpoint=twm_causal_calibration_report, methods=["POST"]),
         Route("/api/twm/scenarios", endpoint=twm_scenarios, methods=["GET", "POST"]),
         Route("/api/twm/scenarios/{id}/compare", endpoint=twm_scenario_compare, methods=["GET", "POST"]),

data_agent/fusion/twm_state_input.py

@@ -154,6 +154,8 @@ def validate_twm_state_input(payload: dict) -> dict:
         errors.append("object_role_registry must be a list")
     if not isinstance(payload.get("semantic_relation_registry"), list):
         errors.append("semantic_relation_registry must be a list")
+    if payload.get("canonical_object_type_registry") is not None and not isinstance(payload.get("canonical_object_type_registry"), list):
+        errors.append("canonical_object_type_registry must be a list")
 
     relation_summary = payload.get("semantic_relation_summary") or {}
     relation_registry = payload.get("semantic_relation_registry") or []
@@ -165,9 +167,86 @@ def validate_twm_state_input(payload: dict) -> dict:
     components = payload.get("state_components") or {}
     if not isinstance(components, dict):
         errors.append("state_components must be an object")
+        components = {}
+    if isinstance(payload.get("object_role_registry"), list):
+        errors.extend(_validate_role_type_closure(payload))
+    if isinstance(relation_registry, list):
+        errors.extend(_validate_component_reference_closure(payload, components, relation_registry))
     return {"valid": not errors, "errors": errors}
 
 
+def _validate_role_type_closure(payload: dict) -> list[str]:
+    errors: list[str] = []
+    role_registry = [row for row in payload.get("object_role_registry") or [] if isinstance(row, dict)]
+    canonical_registry = [row for row in payload.get("canonical_object_type_registry") or [] if isinstance(row, dict)]
+    canonical_types = {str(row.get("object_type")) for row in canonical_registry if row.get("object_type")}
+    seen_roles: set[str] = set()
+    for idx, row in enumerate(role_registry):
+        role = str(row.get("role") or "")
+        standard_role = str(row.get("standard_role") or "")
+        object_type = str(row.get("object_type") or "")
+        if not role:
+            errors.append(f"object_role_registry[{idx}].role is required")
+        elif role in seen_roles:
+            errors.append(f"object_role_registry role is duplicated: {role}")
+        seen_roles.add(role)
+        if not standard_role:
+            errors.append(f"object_role_registry[{idx}].standard_role is required")
+        if not object_type:
+            errors.append(f"object_role_registry[{idx}].object_type is required")
+        elif canonical_types and object_type not in canonical_types:
+            errors.append(f"object_role_registry[{idx}].object_type is not in canonical_object_type_registry: {object_type}")
+    for idx, row in enumerate(canonical_registry):
+        object_type = row.get("object_type")
+        if not object_type:
+            errors.append(f"canonical_object_type_registry[{idx}].object_type is required")
+    return errors
+
+
+def _validate_component_reference_closure(payload: dict, components: dict, relation_registry: list[dict]) -> list[str]:
+    errors: list[str] = []
+    relation_rule_ids = {
+        str(rule_id)
+        for row in relation_registry
+        if isinstance(row, dict)
+        for rule_id in row.get("rule_ids") or []
+        if rule_id
+    }
+    relation_objective_ids = {
+        str(objective_id)
+        for row in relation_registry
+        if isinstance(row, dict)
+        for objective_id in row.get("objective_ids") or []
+        if objective_id
+    }
+    optimization_interface = payload.get("optimization_interface") or {}
+    objective_bindings = optimization_interface.get("objective_bindings") or []
+    bound_objective_ids = {
+        str(row.get("objective_id"))
+        for row in objective_bindings
+        if isinstance(row, dict) and row.get("objective_id")
+    }
+    known_objective_ids = relation_objective_ids | bound_objective_ids
+
+    for component_name, component in components.items():
+        if not isinstance(component, dict):
+            errors.append(f"state_components.{component_name} must be an object")
+            continue
+        for rule_id in component.get("rule_ids") or []:
+            if rule_id and str(rule_id) not in relation_rule_ids:
+                errors.append(f"state_components.{component_name}.rule_ids references unknown semantic relation rule_id: {rule_id}")
+        for objective_id in component.get("objective_ids") or []:
+            if objective_id and str(objective_id) not in known_objective_ids:
+                errors.append(f"state_components.{component_name}.objective_ids references unknown objective_id: {objective_id}")
+
+    hard_constraints = components.get("hard_constraints") or {}
+    if isinstance(hard_constraints, dict):
+        for objective_id in hard_constraints.get("objective_ids") or []:
+            if objective_id and str(objective_id) not in bound_objective_ids:
+                errors.append(f"state_components.hard_constraints.objective_ids is not bound in optimization_interface: {objective_id}")
+    return errors
+
+
 def write_twm_state_input(payload: dict, out_path: str | Path) -> str:
     """Write a TWM state-input artifact and return its path."""
     path = Path(out_path)

data_agent/territory_world_model/__init__.py

@@ -17,6 +17,7 @@
     TwmDynamicsTrainingExample,
     TwmAuditReport,
     TwmEvidenceItem,
+    TwmGeoFMDownstreamExperimentReport,
     TwmGeoFMGateReport,
     TwmGeoFMGateVariant,
     TwmLayerBinding,
@@ -51,6 +52,7 @@
 from .rule_dsl import normalize_rule_body, validate_rule_body
 from .rule_evaluator import RuleEvaluator, evaluate_rules
 from .service import TerritoryWorldModelService, get_territory_world_model_service, reset_territory_world_model_service
+from .spatial_causal_estimator import SpatialCausalEstimatorAdapter, estimate_spatial_treatment_effect
 from .state_builder import StateBuilder, build_state_from_bundle, load_state_source
 
 __all__ = [
@@ -70,6 +72,7 @@
     "TwmDynamicsTrainingDataset",
     "TwmDynamicsTrainingExample",
     "TwmEvidenceItem",
+    "TwmGeoFMDownstreamExperimentReport",
     "TwmGeoFMGateReport",
     "TwmGeoFMGateVariant",
     "TwmLayerBinding",
@@ -99,11 +102,13 @@
     "TwmWorldModelProfile",
     "RuleEvaluator",
     "StateBuilder",
+    "SpatialCausalEstimatorAdapter",
     "TerritoryWorldModelService",
     "build_evidence_chain",
     "build_state_from_bundle",
     "evidence_checksum",
     "evaluate_rules",
+    "estimate_spatial_treatment_effect",
     "get_territory_world_model_service",
     "get_twm_repository",
     "jsonable",

data_agent/territory_world_model/causal_calibration.py

@@ -3,6 +3,7 @@
 import math
 from typing import Any
 
+from .spatial_causal_estimator import build_neighbor_edges, estimate_spatial_treatment_effect
 from .utils import safe_float, truthy
 
 
@@ -30,7 +31,15 @@ def estimate_observational_treatment_effect(records: list[dict[str, Any]], *, th
     aipw_effect, aipw_se, influence = _augmented_ipw_ate(usable)
     balance = _covariate_balance(usable)
     overlap = _overlap_diagnostics(usable, thresholds)
-    spatial = _spatial_interference_diagnostics(usable, influence, thresholds)
+    neighbor_edges = build_neighbor_edges(usable, thresholds)
+    spatial = _spatial_interference_diagnostics(usable, influence, thresholds, neighbor_edges=neighbor_edges)
+    spatial_estimator = estimate_spatial_treatment_effect(
+        usable,
+        thresholds=thresholds,
+        neighbor_edges=neighbor_edges,
+        observational_effect=aipw_effect,
+        observational_standard_error=aipw_se,
+    )
 
     primary_name, primary_effect, primary_se = _primary_estimator(
         naive_effect=naive_effect,
@@ -45,6 +54,7 @@ def estimate_observational_treatment_effect(records: list[dict[str, Any]], *, th
         control_count=len(control),
         thresholds=thresholds,
         overlap=overlap,
+        spatial_estimator=spatial_estimator,
     )
     model_effects = [safe_float(row.get("model_effect"), None) for row in usable]
     model_effects = [float(item) for item in model_effects if item is not None]
@@ -85,6 +95,7 @@ def estimate_observational_treatment_effect(records: list[dict[str, Any]], *, th
         "overlap": overlap,
         "balance": balance,
         "spatial": spatial,
+        "spatial_estimator": spatial_estimator,
     }
 
 
@@ -249,7 +260,21 @@ def _primary_estimator(
     control_count: int,
     thresholds: dict[str, Any],
     overlap: dict[str, Any],
+    spatial_estimator: dict[str, Any] | None = None,
 ) -> tuple[str, float, float]:
+    if (
+        spatial_estimator
+        and spatial_estimator.get("status") == "pass"
+        and usable_count >= int(thresholds.get("min_records", 8))
+        and treated_count >= int(thresholds.get("min_treated", 3))
+        and control_count >= int(thresholds.get("min_control", 3))
+        and overlap.get("status") == "pass"
+    ):
+        return (
+            "spatial_fixed_effect_neighbor_adapter",
+            float(spatial_estimator.get("effect") or 0.0),
+            float(spatial_estimator.get("standard_error") or 0.0),
+        )
     if (
         usable_count >= int(thresholds.get("min_records", 8))
         and treated_count >= int(thresholds.get("min_treated", 3))
@@ -312,7 +337,13 @@ def _overlap_diagnostics(usable: list[dict[str, Any]], thresholds: dict[str, Any
     }
 
 
-def _spatial_interference_diagnostics(usable: list[dict[str, Any]], influence: list[float], thresholds: dict[str, Any]) -> dict[str, Any]:
+def _spatial_interference_diagnostics(
+    usable: list[dict[str, Any]],
+    influence: list[float],
+    thresholds: dict[str, Any],
+    *,
+    neighbor_edges: list[tuple[int, int, float]] | None = None,
+) -> dict[str, Any]:
     spatial_rows = [row for row in usable if row.get("spatial")]
     if not spatial_rows:
         return {
@@ -323,7 +354,7 @@ def _spatial_interference_diagnostics(usable: list[dict[str, Any]], influence: l
             "note": "no spatial coordinates, cluster ids or neighbor links supplied",
         }
 
-    neighbor_edges = _neighbor_edges(usable, thresholds)
+    neighbor_edges = list(neighbor_edges or [])
     cluster_summary = _spatial_cluster_summary(usable)
     exposure = _neighborhood_exposure(usable, neighbor_edges)
     moran = _moran_like_residual_correlation(usable, influence, neighbor_edges)
@@ -366,34 +397,18 @@ def _spatial_attributes(row: dict[str, Any]) -> dict[str, Any]:
     cluster = row.get("spatial_cluster") or row.get("cluster") or row.get("block_id") or row.get("township_id")
     if cluster is not None:
         spatial["cluster"] = str(cluster)
-    neighbors = row.get("neighbors") or row.get("neighbor_unit_ids") or []
-    if isinstance(neighbors, (list, tuple, set)) and neighbors:
-        spatial["neighbors"] = [str(item) for item in neighbors]
+    neighbors = _neighbor_ids(row.get("neighbors") or row.get("neighbor_unit_ids") or [])
+    if neighbors:
+        spatial["neighbors"] = neighbors
     return spatial
 
 
-def _neighbor_edges(usable: list[dict[str, Any]], thresholds: dict[str, Any]) -> list[tuple[int, int, float]]:
-    index_by_id = {str(row.get("unit_id") or idx): idx for idx, row in enumerate(usable)}
-    edges: dict[tuple[int, int], float] = {}
-    for idx, row in enumerate(usable):
-        spatial = dict(row.get("spatial") or {})
-        for neighbor_id in spatial.get("neighbors") or []:
-            other = index_by_id.get(str(neighbor_id))
-            if other is None or other == idx:
-                continue
-            pair = tuple(sorted((idx, other)))
-            edges[pair] = 1.0
-
-    distance_threshold = safe_float(thresholds.get("spatial_neighbor_distance"), None)
-    coordinate_rows = [(idx, dict(row.get("spatial") or {})) for idx, row in enumerate(usable) if "x" in dict(row.get("spatial") or {}) and "y" in dict(row.get("spatial") or {})]
-    if distance_threshold is not None and distance_threshold > 0 and len(coordinate_rows) > 1:
-        for pos, (idx, left) in enumerate(coordinate_rows):
-            for other, right in coordinate_rows[pos + 1 :]:
-                distance = math.sqrt((float(left["x"]) - float(right["x"])) ** 2 + (float(left["y"]) - float(right["y"])) ** 2)
-                if distance <= float(distance_threshold):
-                    pair = tuple(sorted((idx, other)))
-                    edges[pair] = 1.0 / max(distance, 1e-9)
-    return [(left, right, weight) for (left, right), weight in sorted(edges.items())]
+def _neighbor_ids(value: Any) -> list[str]:
+    if isinstance(value, (list, tuple, set)):
+        return [str(item).strip() for item in value if str(item).strip()]
+    if isinstance(value, str):
+        return [item.strip() for item in value.replace(";", ",").split(",") if item.strip()]
+    return []
 
 
 def _spatial_cluster_summary(usable: list[dict[str, Any]]) -> dict[str, Any]:

data_agent/territory_world_model/models.py

@@ -585,6 +585,7 @@ class TwmBeamPlanReport:
     schema: str = "territory_world_model.beam_plan_report.v1"
     scenario: str = "baseline"
     status: str = "review"
+    ranking_policy: dict[str, Any] = field(default_factory=dict)
     candidates: list[dict[str, Any]] = field(default_factory=list)
     ranking: list[dict[str, Any]] = field(default_factory=list)
     selected: dict[str, Any] = field(default_factory=dict)
@@ -628,6 +629,23 @@ def to_dict(self) -> dict[str, Any]:
         return jsonable(self)
 
 
+@dataclass
+class TwmGeoFMDownstreamExperimentReport:
+    state_version_id: str = ""
+    project_id: str = ""
+    schema: str = "territory_world_model.geofm_downstream_experiment_report.v1"
+    status: str = "review"
+    experiment: dict[str, Any] = field(default_factory=dict)
+    variants: dict[str, Any] = field(default_factory=dict)
+    evidence: dict[str, Any] = field(default_factory=dict)
+    gate_report: dict[str, Any] = field(default_factory=dict)
+    recommendations: list[str] = field(default_factory=list)
+    created_at: str = field(default_factory=now_utc_iso)
+
+    def to_dict(self) -> dict[str, Any]:
+        return jsonable(self)
+
+
 @dataclass
 class TwmCausalCalibrationReport:
     state_version_id: str = ""