feat(knowledge): add temporal validity windows and as-of filtering

Author: ibacalu

README.md

@@ -858,15 +858,15 @@ When you configure workflows-mcp, your AI assistant gets these tools:
 ### Knowledge Tools
 
 - **search_knowledge** - Hybrid search (vector + full-text + RRF fusion)
-  - `query`, `source`, `categories`, `min_confidence`, `limit`
+  - `query`, `source`, `categories`, `as_of`, `min_confidence`, `limit`
 
 - **store_knowledge** - Persist a new fact with auto-computed embedding
-  - `content`, `source`, `confidence` (default 0.8), `categories`
+  - `content`, `source`, `path`, `valid_from`, `valid_to`, `confidence` (default 0.8), `categories`
   - `authority`: `AGENT` (default), `EXTRACTED`, `COMMUNITY_SUMMARY`, or `USER_VALIDATED`
   - `lifecycle_state`: `ACTIVE` (default) or `QUARANTINED`
 
 - **recall_knowledge** - Filter-based retrieval (no semantic search)
-  - `source`, `categories`, `lifecycle_state`, `min_confidence`, `limit`, `order`
+  - `source`, `categories`, `as_of`, `lifecycle_state`, `min_confidence`, `limit`, `order`
 
 - **forget_knowledge** - Archive propositions (transition to ARCHIVED state)
   - By ID: `proposition_ids` (list of UUIDs)
@@ -887,7 +887,7 @@ When you configure workflows-mcp, your AI assistant gets these tools:
   - Returns `invalidated_count`
 
 - **knowledge_context** - Token-budgeted context assembly for LLM prompts
-  - `query`, `source`, `categories`, `max_tokens`, `diversity`
+  - `query`, `source`, `categories`, `as_of`, `max_tokens`, `diversity`
 
 ---
 

src/workflows_mcp/engine/executors_knowledge.py

@@ -13,6 +13,7 @@
 import logging
 import os
 import uuid
+from datetime import datetime
 from typing import Any, ClassVar, Literal
 
 from pydantic import Field, model_validator
@@ -84,6 +85,17 @@ def _get_auth_method(context: Execution) -> str:
     return "SYSTEM"
 
 
+def _coerce_iso_datetime(value: str | None, field_name: str) -> datetime | None:
+    """Parse ISO datetime strings (including trailing Z) to datetime for DB bindings."""
+    if value is None:
+        return None
+    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
+    try:
+        return datetime.fromisoformat(normalized)
+    except ValueError as e:
+        raise ValueError(f"'{field_name}' must be a valid ISO datetime") from e
+
+
 # ===========================================================================
 # Input / Output Models
 # ===========================================================================
@@ -156,6 +168,10 @@ class KnowledgeInput(BlockInput):
         default=None,
         description="Filter by category UUIDs",
     )
+    as_of: str | None = Field(
+        default=None,
+        description="Point-in-time filter (ISO datetime) against proposition validity window",
+    )
     min_confidence: float | None = Field(
         default=None,
         description="Minimum confidence threshold",
@@ -222,6 +238,14 @@ class KnowledgeInput(BlockInput):
             "Required for document-derived propositions; omit for agent observations."
         ),
     )
+    valid_from: str | None = Field(
+        default=None,
+        description="World-truth start datetime (ISO). NULL means open start",
+    )
+    valid_to: str | None = Field(
+        default=None,
+        description="World-truth end datetime (ISO). NULL means open end",
+    )
 
     # --- Recall fields ---
     where: dict[str, Any] | None = Field(
@@ -510,6 +534,7 @@ async def _op_search(
         resolved_categories = None
         if inputs.categories:
             resolved_categories = await self._resolve_categories(inputs.categories, backend)
+        as_of = _coerce_iso_datetime(inputs.as_of, "as_of")
 
         # Compute query embedding
         embedding, _, _, _ = await compute_embedding(
@@ -523,6 +548,7 @@ async def _op_search(
             query_embedding=embedding,
             source=inputs.source,
             categories=resolved_categories,
+            as_of=as_of,
             min_confidence=inputs.min_confidence
             if inputs.min_confidence is not None
             else DEFAULT_MIN_CONFIDENCE,
@@ -537,6 +563,7 @@ async def _op_search(
             query_text=inputs.query,
             source=inputs.source,
             categories=resolved_categories,
+            as_of=as_of,
             min_confidence=inputs.min_confidence
             if inputs.min_confidence is not None
             else DEFAULT_MIN_CONFIDENCE,
@@ -674,6 +701,8 @@ async def _op_store(
         created_by = _get_audit_user_id(context)
         auth_method = _get_auth_method(context)
         user_string = _get_user_string_id(context)
+        valid_from = _coerce_iso_datetime(inputs.valid_from, "valid_from")
+        valid_to = _coerce_iso_datetime(inputs.valid_to, "valid_to")
 
         # Build metadata JSON string with auth info
         if auth_method and isinstance(auth_method, str):
@@ -689,13 +718,15 @@ async def _op_store(
                 (id, item_id, content, embedding, search_vector,
                  authority, lifecycle_state, confidence,
                  embedding_model, embedding_dimensions, metadata,
+                 valid_from, valid_to,
                  created_by, auth_method, source_name, source_type)
             VALUES
                 ($1::uuid, $2::uuid, $3, $4::vector,
                  to_tsvector('english', $3),
                  $5, $6, $7,
                  $8, $9, $10::jsonb,
-                 $11::uuid, $12, $13, $14)
+                 $11::timestamptz, $12::timestamptz,
+                 $13::uuid, $14, $15, $16)
             """,
             (
                 prop_id,
@@ -708,6 +739,8 @@ async def _op_store(
                 model_name,
                 dimensions,
                 metadata_json,
+                valid_from,
+                valid_to,
                 str(created_by),
                 auth_method,
                 source_name,
@@ -881,6 +914,15 @@ def next_param(value: Any) -> str:
                 f"kp.created_at <= {next_param(inputs.created_before)}::timestamptz"
             )
 
+        if inputs.as_of:
+            as_of_param = next_param(_coerce_iso_datetime(inputs.as_of, "as_of"))
+            where_clauses.append(
+                f"(kp.valid_from IS NULL OR kp.valid_from <= {as_of_param}::timestamptz)"
+            )
+            where_clauses.append(
+                f"(kp.valid_to IS NULL OR kp.valid_to >= {as_of_param}::timestamptz)"
+            )
+
         where_sql = " AND ".join(where_clauses) if where_clauses else "TRUE"
         return where_sql, params
 
@@ -1259,6 +1301,7 @@ async def _search_with_embeddings(
         resolved_categories = None
         if inputs.categories:
             resolved_categories = await self._resolve_categories(inputs.categories, backend)
+        as_of = _coerce_iso_datetime(inputs.as_of, "as_of")
 
         query_embedding, _, _, _ = await compute_embedding(
             text=inputs.query,
@@ -1271,6 +1314,7 @@ async def _search_with_embeddings(
             query_embedding=query_embedding,
             source=inputs.source,
             categories=resolved_categories,
+            as_of=as_of,
             min_confidence=inputs.min_confidence
             if inputs.min_confidence is not None
             else DEFAULT_MIN_CONFIDENCE,
@@ -1286,6 +1330,7 @@ async def _search_with_embeddings(
             query_text=inputs.query,
             source=inputs.source,
             categories=resolved_categories,
+            as_of=as_of,
             min_confidence=inputs.min_confidence
             if inputs.min_confidence is not None
             else DEFAULT_MIN_CONFIDENCE,

src/workflows_mcp/engine/knowledge/schema.py

@@ -479,6 +479,28 @@
         DELETE FROM knowledge_entities WHERE entity_type = 'category';
         """,
     ),
+    (
+        11,
+        "Add temporal validity window columns to knowledge_propositions",
+        """
+        DO $$
+        BEGIN
+            IF NOT EXISTS (
+                SELECT 1 FROM information_schema.columns
+                WHERE table_name = 'knowledge_propositions' AND column_name = 'valid_from'
+            ) THEN
+                ALTER TABLE knowledge_propositions ADD COLUMN valid_from TIMESTAMPTZ NULL;
+            END IF;
+
+            IF NOT EXISTS (
+                SELECT 1 FROM information_schema.columns
+                WHERE table_name = 'knowledge_propositions' AND column_name = 'valid_to'
+            ) THEN
+                ALTER TABLE knowledge_propositions ADD COLUMN valid_to TIMESTAMPTZ NULL;
+            END IF;
+        END $$;
+        """,
+    ),
 ]
 
 SCHEMA_VERSION = MIGRATIONS[-1][0] if MIGRATIONS else 0

src/workflows_mcp/engine/knowledge/search.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import logging
+from datetime import datetime
 from typing import Any
 
 from .constants import (
@@ -27,6 +28,7 @@ def build_vector_search_query(
     *,
     source: str | None = None,
     categories: list[str] | None = None,
+    as_of: datetime | None = None,
     min_confidence: float = DEFAULT_MIN_CONFIDENCE,
     lifecycle_state: str = LifecycleState.ACTIVE,
     limit: int = DEFAULT_LIMIT,
@@ -89,6 +91,13 @@ def next_param(value: Any) -> str:
             f")"
         )
 
+    if as_of:
+        as_of_param = next_param(as_of)
+        where_clauses.append(
+            f"(kp.valid_from IS NULL OR kp.valid_from <= {as_of_param}::timestamptz)"
+        )
+        where_clauses.append(f"(kp.valid_to IS NULL OR kp.valid_to >= {as_of_param}::timestamptz)")
+
     where_clause = " AND ".join(where_clauses)
 
     embedding_col = ", kp.embedding" if include_embeddings else ""
@@ -114,6 +123,7 @@ def build_fts_search_query(
     *,
     source: str | None = None,
     categories: list[str] | None = None,
+    as_of: datetime | None = None,
     min_confidence: float = DEFAULT_MIN_CONFIDENCE,
     lifecycle_state: str = LifecycleState.ACTIVE,
     limit: int = DEFAULT_LIMIT,
@@ -167,6 +177,13 @@ def next_param(value: Any) -> str:
             f")"
         )
 
+    if as_of:
+        as_of_param = next_param(as_of)
+        where_clauses.append(
+            f"(kp.valid_from IS NULL OR kp.valid_from <= {as_of_param}::timestamptz)"
+        )
+        where_clauses.append(f"(kp.valid_to IS NULL OR kp.valid_to >= {as_of_param}::timestamptz)")
+
     where_clause = " AND ".join(where_clauses)
 
     sql = f"""