diff --git a/.gitignore b/.gitignore
index 1c46983c5..3ce7f1130 100644
--- a/.gitignore
+++ b/.gitignore
@@ -131,3 +131,6 @@ geaflow-console/app/bootstrap/src/main/resources/public
# vs code
.vscode
+
+# Python bytecode generated by temporal benchmark scripts
+geaflow-ai/bin/__pycache__/
diff --git a/geaflow-ai/bin/requirements-cross-encoder.txt b/geaflow-ai/bin/requirements-cross-encoder.txt
new file mode 100644
index 000000000..e8ead4f3c
--- /dev/null
+++ b/geaflow-ai/bin/requirements-cross-encoder.txt
@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# setup-temporal-cross-encoder.ps1 installs torch 2.8.0+cpu or 2.8.0+cu128 from its matching wheel index.
+transformers==4.57.1
+numpy==2.3.5
diff --git a/geaflow-ai/bin/score-temporal-cross-encoder.py b/geaflow-ai/bin/score-temporal-cross-encoder.py
new file mode 100644
index 000000000..0477fe4ed
--- /dev/null
+++ b/geaflow-ai/bin/score-temporal-cross-encoder.py
@@ -0,0 +1,412 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Score exported query/evidence pairs with the pinned MiniLM cross-encoder."""
+
+import argparse
+import hashlib
+import importlib.metadata
+import json
+import math
+import os
+from pathlib import Path
+import platform
+import re
+import sys
+import time
+
+
+MODEL_ID = "cross-encoder/ms-marco-MiniLM-L6-v2"
+MODEL_REVISION = "233902d25c440f23af6f7d6e94d2946bac0bee0a"
+DEPENDENCIES = {"torch": "2.8.0", "transformers": "4.57.1", "numpy": "2.3.5"}
+QUERY_FIELDS = {"schemaVersion", "queryId", "queryText", "poolSha256", "candidates"}
+CANDIDATE_FIELDS = {"evidenceId", "text"}
+
+
+def _object_without_duplicates(pairs):
+ result = {}
+ for name, value in pairs:
+ if name in result:
+ raise ValueError("Duplicate JSON field: " + name)
+ result[name] = value
+ return result
+
+
+def _reject_constant(value):
+ raise ValueError("Non-finite JSON number: " + value)
+
+
+def _require_fields(value, expected, name):
+ if not isinstance(value, dict) or set(value) != expected:
+ raise ValueError(name + " must contain exactly: " + ", ".join(sorted(expected)))
+
+
+def _require_text(value, name):
+ if not isinstance(value, str) or not value.strip():
+ raise ValueError(name + " must be a nonempty string")
+
+
+def file_sha256(path):
+ digest = hashlib.sha256()
+ with Path(path).open("rb") as source:
+ for block in iter(lambda: source.read(1024 * 1024), b""):
+ digest.update(block)
+ return digest.hexdigest()
+
+
+def iter_queries(path, max_candidates=100):
+ """Validate one query at a time without retaining all evidence texts."""
+ if type(max_candidates) is not int or not 100 <= max_candidates <= 1000:
+ raise ValueError("max-candidates must be between 100 and 1000")
+ query_ids = set()
+ with Path(path).open("r", encoding="utf-8", newline="") as source:
+ for line_number, line in enumerate(source, 1):
+ try:
+ row = json.loads(line, object_pairs_hook=_object_without_duplicates,
+ parse_constant=_reject_constant)
+ _require_fields(row, QUERY_FIELDS, "Query")
+ if type(row["schemaVersion"]) is not int or row["schemaVersion"] != 1:
+ raise ValueError("schemaVersion must be integer 1")
+ _require_text(row["queryId"], "queryId")
+ _require_text(row["queryText"], "queryText")
+ if row["queryId"] in query_ids:
+ raise ValueError("Duplicate queryId: " + row["queryId"])
+ fingerprint = row["poolSha256"]
+ if not isinstance(fingerprint, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", fingerprint):
+ raise ValueError("poolSha256 must contain 64 hexadecimal characters")
+ candidates = row["candidates"]
+ if not isinstance(candidates, list) or len(candidates) > max_candidates:
+ raise ValueError("candidates must be a list containing at most "
+ + str(max_candidates) + " entries")
+ evidence_ids = set()
+ for candidate in candidates:
+ _require_fields(candidate, CANDIDATE_FIELDS, "Candidate")
+ _require_text(candidate["evidenceId"], "evidenceId")
+ _require_text(candidate["text"], "candidate text")
+ if candidate["evidenceId"] in evidence_ids:
+ raise ValueError("Duplicate evidenceId: " + candidate["evidenceId"])
+ evidence_ids.add(candidate["evidenceId"])
+ query_ids.add(row["queryId"])
+ yield row
+ except (ValueError, TypeError) as error:
+ raise ValueError("Input line " + str(line_number) + ": " + str(error)) from error
+ if not query_ids:
+ raise ValueError("Input must contain at least one query")
+
+
+def scan_queries(path, max_candidates=100):
+ query_count = pair_count = 0
+ for row in iter_queries(path, max_candidates):
+ query_count += 1
+ pair_count += len(row["candidates"])
+ return query_count, pair_count, file_sha256(path)
+
+
+def read_queries(path, max_candidates=100):
+ """Compatibility helper; the scoring CLI uses the streaming iterator instead."""
+ return list(iter_queries(path, max_candidates)), file_sha256(path)
+
+
+def peak_rss_bytes():
+ """OS process-lifetime high-water mark, including model load and Python allocations."""
+ if os.name == "nt":
+ import ctypes
+ from ctypes import wintypes
+
+ class MemoryCounters(ctypes.Structure):
+ _fields_ = [("cb", wintypes.DWORD), ("page_faults", wintypes.DWORD)] + [
+ (name, ctypes.c_size_t) for name in ("peak_working_set", "working_set",
+ "peak_paged", "paged", "peak_nonpaged",
+ "nonpaged", "pagefile", "peak_pagefile")]
+
+ counters = MemoryCounters()
+ counters.cb = ctypes.sizeof(counters)
+ kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+ kernel.GetCurrentProcess.restype = wintypes.HANDLE
+ psapi = ctypes.WinDLL("psapi", use_last_error=True)
+ psapi.GetProcessMemoryInfo.argtypes = [wintypes.HANDLE, ctypes.POINTER(MemoryCounters),
+ wintypes.DWORD]
+ if not psapi.GetProcessMemoryInfo(kernel.GetCurrentProcess(), ctypes.byref(counters), counters.cb):
+ raise ctypes.WinError(ctypes.get_last_error())
+ return counters.peak_working_set
+ import resource
+
+ value = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+ return value if sys.platform == "darwin" else value * 1024
+
+
+def check_environment(device="cpu"):
+ if device not in ("cpu", "cuda"):
+ raise ValueError("device must be cpu or cuda")
+ versions = {}
+ failures = []
+ if not (3, 11) <= sys.version_info[:2] <= (3, 13):
+ failures.append("Python 3.11 through 3.13 is required")
+ for package, expected in DEPENDENCIES.items():
+ try:
+ actual = importlib.metadata.version(package)
+ except importlib.metadata.PackageNotFoundError:
+ failures.append(package + "==" + expected + " is missing")
+ continue
+ versions[package] = actual
+ if actual.split("+", 1)[0] != expected:
+ failures.append(package + " must be " + expected + "; found " + actual)
+ if failures:
+ instructions = ("Run geaflow-ai/bin/setup-temporal-cross-encoder.ps1 -Device " + device)
+ raise RuntimeError("; ".join(failures) + "\n" + instructions)
+ if device == "cuda":
+ os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
+ import torch
+
+ if not torch.cuda.is_available() or torch.cuda.device_count() < 1 or not torch.version.cuda:
+ raise RuntimeError("CUDA is unavailable. Check the NVIDIA driver and run "
+ "geaflow-ai/bin/setup-temporal-cross-encoder.ps1 -Device cuda; "
+ "CPU fallback is disabled.")
+ properties = torch.cuda.get_device_properties(0)
+ versions.update({"cudaVersion": torch.version.cuda, "deviceName": properties.name,
+ "deviceMemoryBytes": properties.total_memory})
+ return versions
+
+
+class TransformerScorer:
+ def __init__(self, cache_dir, batch_size, max_length, threads, local_files_only, device="cpu"):
+ if device not in ("cpu", "cuda"):
+ raise ValueError("device must be cpu or cuda")
+ if device == "cuda":
+ os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
+ # Keep imports lazy so protocol validation and unit tests need no model dependencies.
+ import torch
+ from transformers import AutoModelForSequenceClassification, AutoTokenizer
+
+ torch.manual_seed(0)
+ torch.set_num_threads(threads)
+ torch.set_num_interop_threads(1)
+ torch.use_deterministic_algorithms(True)
+ if device == "cuda":
+ torch.backends.cuda.matmul.allow_tf32 = False
+ torch.backends.cudnn.allow_tf32 = False
+ torch.backends.cudnn.benchmark = False
+ options = {"revision": MODEL_REVISION, "cache_dir": str(cache_dir),
+ "local_files_only": local_files_only, "trust_remote_code": False}
+ self.tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, **options)
+ self.model = AutoModelForSequenceClassification.from_pretrained(
+ MODEL_ID, **options, use_safetensors=True, attn_implementation="eager",
+ dtype=torch.float32)
+ if self.model.config.num_labels != 1:
+ raise ValueError("The pinned model must have exactly one output label")
+ self.model.to(device=device, dtype=torch.float32)
+ self.model.eval()
+ self.torch = torch
+ self.batch_size = batch_size
+ self.max_length = max_length
+ self.device = device
+
+ def synchronize(self):
+ if self.device == "cuda":
+ self.torch.cuda.synchronize()
+
+ def memory_metrics(self):
+ if self.device == "cuda":
+ return {"cudaPeakAllocatedBytes": self.torch.cuda.max_memory_allocated(),
+ "cudaPeakReservedBytes": self.torch.cuda.max_memory_reserved(),
+ "cudaPeakScope": "process_lifetime_including_model_load"}
+ return {}
+
+ def __call__(self, query_text, evidence_texts):
+ scores = []
+ with self.torch.inference_mode():
+ for start in range(0, len(evidence_texts), self.batch_size):
+ texts = evidence_texts[start:start + self.batch_size]
+ tokens = self.tokenizer([query_text] * len(texts), texts, padding=True,
+ truncation=True, max_length=self.max_length,
+ return_tensors="pt")
+ tokens = {name: tensor.to(device=self.device) for name, tensor in tokens.items()}
+ logits = self.model(**tokens).logits
+ if tuple(logits.shape) != (len(texts), 1):
+ raise ValueError("Unexpected model output shape: " + str(tuple(logits.shape)))
+ scores.extend(float(value) for value in logits[:, 0].cpu().tolist())
+ return scores
+
+
+def validate_output_paths(output, metadata):
+ if output.resolve() == metadata.resolve():
+ raise ValueError("Scores and metadata must have different paths")
+ for path in (output, metadata):
+ if path.exists():
+ raise FileExistsError("Refusing to overwrite: " + str(path))
+
+
+def write_scores(rows, input_sha256, output, metadata, scorer, batch_size=16,
+ max_length=512, threads=1, load_nanos=0, versions=None, progress=None, device="cpu",
+ query_count=None, max_candidates=100, input_path=None):
+ if device not in ("cpu", "cuda"):
+ raise ValueError("device must be cpu or cuda")
+ if type(max_candidates) is not int or not 100 <= max_candidates <= 1000:
+ raise ValueError("max-candidates must be between 100 and 1000")
+ versions = versions or {}
+ cuda_details = {}
+ if device == "cuda":
+ for key in ("cudaVersion", "deviceName"):
+ _require_text(versions.get(key), key)
+ cuda_details[key] = versions[key]
+ memory = versions.get("deviceMemoryBytes")
+ if type(memory) is not int or memory <= 0:
+ raise ValueError("deviceMemoryBytes must be a positive integer")
+ cuda_details["deviceMemoryBytes"] = memory
+ output, metadata = Path(output), Path(metadata)
+ validate_output_paths(output, metadata)
+ output.parent.mkdir(parents=True, exist_ok=True)
+ metadata.parent.mkdir(parents=True, exist_ok=True)
+ if query_count is None:
+ try:
+ query_count = len(rows)
+ except TypeError:
+ pass
+ inference_nanos = 0
+ pair_count = 0
+ processed_queries = 0
+ scores_digest = hashlib.sha256()
+ wall_started = time.monotonic_ns()
+ with output.open("xb") as scores_file:
+ for index, row in enumerate(rows, 1):
+ candidates = row["candidates"]
+ if len(candidates) > max_candidates:
+ raise ValueError("Candidate pool exceeds max-candidates")
+ if candidates and device == "cuda":
+ scorer.synchronize()
+ started = time.perf_counter_ns() if candidates else 0
+ values = list(scorer(row["queryText"], [item["text"] for item in candidates])) if candidates else []
+ if candidates and device == "cuda":
+ scorer.synchronize()
+ elapsed = time.perf_counter_ns() - started if candidates else 0
+ if len(values) != len(candidates):
+ raise ValueError("Scorer returned the wrong number of scores for " + row["queryId"])
+ if any(type(value) not in (int, float) or not math.isfinite(value) for value in values):
+ raise ValueError("Scorer returned a non-finite or nonnumeric score for " + row["queryId"])
+ result = {"schemaVersion": 1, "queryId": row["queryId"],
+ "poolSha256": row["poolSha256"], "elapsedNanos": elapsed,
+ "scores": [{"evidenceId": candidate["evidenceId"], "logit": float(value)}
+ for candidate, value in zip(candidates, values)]}
+ line = (json.dumps(result, ensure_ascii=False, separators=(",", ":"),
+ allow_nan=False) + "\n").encode("utf-8")
+ scores_file.write(line)
+ scores_digest.update(line)
+ inference_nanos += elapsed
+ pair_count += len(candidates)
+ processed_queries = index
+ if progress is not None and (index == 1 or index % 10 == 0 or index == query_count):
+ print("Scored " + str(index) + "/" + str(query_count or "?") + " queries; "
+ + str(pair_count) + " pairs",
+ file=progress, flush=True)
+ if not processed_queries or (query_count is not None and processed_queries != query_count):
+ raise ValueError("Input query count changed while scoring")
+ if input_path is not None and file_sha256(input_path) != input_sha256:
+ raise ValueError("Input changed after validation")
+ memory = {}
+ try:
+ memory["peakRssBytes"] = peak_rss_bytes()
+ except (OSError, ImportError) as error:
+ memory.update({"peakRssBytes": None, "peakRssError": str(error)})
+ memory["peakRssScope"] = "process_lifetime_including_model_load"
+ memory["peakRssMethod"] = "GetProcessMemoryInfo.PeakWorkingSetSize" if os.name == "nt" else "getrusage.ru_maxrss"
+ if hasattr(scorer, "memory_metrics"):
+ measured = scorer.memory_metrics()
+ if isinstance(measured, dict):
+ memory.update(measured)
+ result = {"schemaVersion": 1, "modelId": MODEL_ID, "modelRevision": MODEL_REVISION,
+ "device": device, "batchSize": batch_size, "maxLength": max_length,
+ "threads": threads, "inputSha256": input_sha256,
+ "scoresSha256": scores_digest.hexdigest(),
+ "queryCount": processed_queries, "pairCount": pair_count, "loadNanos": load_nanos,
+ "inferenceNanos": inference_nanos, "pythonVersion": platform.python_version(),
+ "torchVersion": versions.get("torch", "unknown"),
+ "transformersVersion": versions.get("transformers", "unknown"),
+ "scoreKind": "raw_logit", "maxCandidates": max_candidates,
+ "timingScope": "complete_pool",
+ "queryElapsedNanosScope": "scores.jsonl elapsedNanos measures every candidate: tokenizer+forward+device synchronization",
+ "loadNanosScope": "tokenizer and model load plus selected-device transfer; excludes input validation",
+ "scoringWallNanos": time.monotonic_ns() - wall_started,
+ "scoringWallScope": "stream reading, inference, score writing, final input hash; excludes model load and preflight"}
+ result.update(cuda_details)
+ result.update(memory)
+ # Metadata is the completion marker: leave it absent if scoring fails.
+ with metadata.open("x", encoding="utf-8", newline="\n") as metadata_file:
+ metadata_file.write(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + "\n")
+ return result
+
+
+def _positive_integer(text):
+ value = int(text)
+ if value <= 0:
+ raise argparse.ArgumentTypeError("Value must be positive")
+ return value
+
+
+def _candidate_limit(text):
+ value = int(text)
+ if not 100 <= value <= 1000:
+ raise argparse.ArgumentTypeError("max-candidates must be between 100 and 1000")
+ return value
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--input", type=Path)
+ parser.add_argument("--output", type=Path)
+ parser.add_argument("--metadata", type=Path)
+ parser.add_argument("--cache-dir", type=Path)
+ parser.add_argument("--batch-size", type=_positive_integer, default=16)
+ parser.add_argument("--max-length", type=_positive_integer, default=512)
+ parser.add_argument("--max-candidates", type=_candidate_limit, default=100)
+ parser.add_argument("--threads", type=_positive_integer, default=1)
+ parser.add_argument("--device", choices=("cpu", "cuda"), default="cpu")
+ parser.add_argument("--local-files-only", action="store_true")
+ parser.add_argument("--check-environment", action="store_true")
+ args = parser.parse_args(argv)
+ try:
+ if args.check_environment:
+ print(json.dumps(check_environment(args.device), sort_keys=True))
+ return 0
+ if any(value is None for value in (args.input, args.output, args.metadata, args.cache_dir)):
+ parser.error("--input, --output, --metadata and --cache-dir are required for scoring")
+ if args.max_length > 512:
+ raise ValueError("The pinned MiniLM model supports max-length at most 512")
+ validate_output_paths(args.output, args.metadata)
+ query_count, _, input_sha256 = scan_queries(args.input, args.max_candidates)
+ versions = check_environment(args.device)
+ print("Loading pinned " + MODEL_ID + " on " + args.device.upper() + "...",
+ file=sys.stderr, flush=True)
+ started = time.perf_counter_ns()
+ scorer = TransformerScorer(args.cache_dir, args.batch_size, args.max_length,
+ args.threads, args.local_files_only, device=args.device)
+ if args.device == "cuda":
+ scorer.synchronize()
+ load_nanos = time.perf_counter_ns() - started
+ write_scores(iter_queries(args.input, args.max_candidates), input_sha256,
+ args.output, args.metadata, scorer,
+ args.batch_size, args.max_length, args.threads, load_nanos,
+ versions, sys.stderr, device=args.device, query_count=query_count,
+ max_candidates=args.max_candidates, input_path=args.input)
+ print("Completed local scoring: " + str(args.metadata), file=sys.stderr, flush=True)
+ return 0
+ except Exception as error:
+ print("[ERROR] " + str(error), file=sys.stderr, flush=True)
+ return 2
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/geaflow-ai/bin/test_temporal_cross_encoder.py b/geaflow-ai/bin/test_temporal_cross_encoder.py
new file mode 100644
index 000000000..fd492ad37
--- /dev/null
+++ b/geaflow-ai/bin/test_temporal_cross_encoder.py
@@ -0,0 +1,437 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Protocol tests use only the standard library and never load model weights."""
+
+import hashlib
+from contextlib import nullcontext
+import importlib.util
+import io
+import json
+from pathlib import Path
+import sys
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+
+MODULE_PATH = Path(__file__).with_name("score-temporal-cross-encoder.py")
+SPEC = importlib.util.spec_from_file_location("temporal_cross_encoder", MODULE_PATH)
+SCORER = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(SCORER)
+
+
+class CrossEncoderProtocolTest(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary.cleanup)
+ self.root = Path(self.temporary.name)
+ self.input = self.root / "input.jsonl"
+ self.output = self.root / "scores.jsonl"
+ self.metadata = self.root / "metadata.json"
+
+ @staticmethod
+ def query(query_id="q1"):
+ return {"schemaVersion": 1, "queryId": query_id, "queryText": "地球 is warming?",
+ "poolSha256": "a" * 64,
+ "candidates": [{"evidenceId": "e2", "text": "Second evidence"},
+ {"evidenceId": "e1", "text": "First evidence"}]}
+
+ def write_input(self, rows):
+ self.input.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
+ encoding="utf-8")
+
+ def test_rejects_unexpected_fields_types_and_candidate_limits(self):
+ invalid = []
+ row = self.query()
+ row["gold"] = ["e1"]
+ invalid.append(row)
+ row = self.query()
+ row["candidates"][0]["label"] = "SUPPORTS"
+ invalid.append(row)
+ row = self.query()
+ row["schemaVersion"] = True
+ invalid.append(row)
+ row = self.query()
+ row["poolSha256"] = "not-a-fingerprint"
+ invalid.append(row)
+ row = self.query()
+ row["queryText"] = " "
+ invalid.append(row)
+ row = self.query()
+ row["candidates"] = [{"evidenceId": str(index), "text": "text"} for index in range(101)]
+ invalid.append(row)
+ for row in invalid:
+ with self.subTest(row=row):
+ self.write_input([row])
+ with self.assertRaises(ValueError):
+ SCORER.read_queries(self.input)
+
+ def test_rejects_duplicate_query_evidence_and_json_fields(self):
+ self.write_input([self.query(), self.query()])
+ with self.assertRaisesRegex(ValueError, "Duplicate queryId"):
+ SCORER.read_queries(self.input)
+ row = self.query()
+ row["candidates"][1]["evidenceId"] = "e2"
+ self.write_input([row])
+ with self.assertRaisesRegex(ValueError, "Duplicate evidenceId"):
+ SCORER.read_queries(self.input)
+ self.input.write_text('{"schemaVersion":1,"schemaVersion":1}\n', encoding="utf-8")
+ with self.assertRaisesRegex(ValueError, "Duplicate JSON field"):
+ SCORER.read_queries(self.input)
+
+ def test_scores_only_text_preserves_order_hashes_and_progress(self):
+ self.write_input([self.query("q" + str(index)) for index in range(11)])
+ rows, fingerprint = SCORER.read_queries(self.input)
+ calls = []
+
+ def fake(query_text, evidence_texts):
+ calls.append((query_text, evidence_texts))
+ return [-2.5, 3.25]
+
+ progress = io.StringIO()
+ result = SCORER.write_scores(rows, fingerprint, self.output, self.metadata, fake,
+ versions={"torch": "fake", "transformers": "fake"},
+ progress=progress)
+ output = [json.loads(line) for line in self.output.read_text(encoding="utf-8").splitlines()]
+ self.assertEqual(calls[0], ("地球 is warming?", ["Second evidence", "First evidence"]))
+ self.assertEqual(output[0]["scores"], [{"evidenceId": "e2", "logit": -2.5},
+ {"evidenceId": "e1", "logit": 3.25}])
+ self.assertEqual(output[0]["poolSha256"], "a" * 64)
+ self.assertGreaterEqual(output[0]["elapsedNanos"], 0)
+ self.assertEqual(result["inputSha256"], hashlib.sha256(self.input.read_bytes()).hexdigest())
+ self.assertEqual(result["scoresSha256"], hashlib.sha256(self.output.read_bytes()).hexdigest())
+ self.assertEqual(result["pairCount"], 22)
+ self.assertEqual(result["queryCount"], 11)
+ self.assertEqual(result["modelId"], SCORER.MODEL_ID)
+ self.assertEqual(result["modelRevision"], SCORER.MODEL_REVISION)
+ self.assertEqual(json.loads(self.metadata.read_text(encoding="utf-8")), result)
+ self.assertIn("Scored 1/11", progress.getvalue())
+ self.assertIn("Scored 10/11", progress.getvalue())
+ self.assertIn("Scored 11/11", progress.getvalue())
+
+ def test_empty_candidate_query_is_preserved_without_inference(self):
+ row = self.query()
+ row["candidates"] = []
+ self.write_input([row])
+ rows, fingerprint = SCORER.read_queries(self.input)
+
+ def forbidden(*unused):
+ self.fail("An empty candidate pool must not call the scorer")
+
+ result = SCORER.write_scores(rows, fingerprint, self.output, self.metadata, forbidden)
+ self.assertEqual(result["pairCount"], 0)
+ output = json.loads(self.output.read_text(encoding="utf-8"))
+ self.assertEqual(output["scores"], [])
+ self.assertEqual(output["elapsedNanos"], 0)
+
+ def test_default_limit_is_100_and_larger_pools_require_explicit_bound(self):
+ row = self.query()
+ row["candidates"] = [{"evidenceId": "e" + str(index), "text": "evidence " + str(index)}
+ for index in range(500)]
+ self.write_input([row])
+ with self.assertRaisesRegex(ValueError, "at most 100 entries"):
+ SCORER.read_queries(self.input)
+ count, pairs, digest = SCORER.scan_queries(self.input, 500)
+ self.assertEqual((count, pairs), (1, 500))
+ self.assertEqual(digest, SCORER.file_sha256(self.input))
+ for limit in (99, 1001, True):
+ with self.subTest(limit=limit), self.assertRaisesRegex(ValueError, "100 and 1000"):
+ list(SCORER.iter_queries(self.input, limit))
+ row["candidates"].append({"evidenceId": "e500", "text": "one too many"})
+ self.write_input([row])
+ with self.assertRaisesRegex(ValueError, "at most 500 entries"):
+ list(SCORER.iter_queries(self.input, 500))
+
+ def test_streaming_iterator_does_not_preload_later_queries(self):
+ self.write_input([self.query()])
+ with self.input.open("a", encoding="utf-8") as source:
+ source.write('{"not":"a query"}\n')
+ rows = SCORER.iter_queries(self.input)
+ self.assertEqual(next(rows)["queryId"], "q1")
+ with self.assertRaisesRegex(ValueError, "Input line 2"):
+ next(rows)
+
+ def test_main_streams_500_pools_and_hashes_without_whole_file_read(self):
+ first, second = self.query("q1"), self.query("q2")
+ for row in (first, second):
+ row["candidates"] = [{"evidenceId": "e" + str(index), "text": "证据 " + str(index)}
+ for index in range(500)]
+ self.write_input([first, second])
+ calls = []
+
+ def fake(query, texts):
+ calls.append(len(texts))
+ return [index / 10 for index in range(len(texts))]
+
+ with patch.object(SCORER, "check_environment", return_value={}), \
+ patch.object(SCORER, "TransformerScorer", return_value=fake), \
+ patch.object(SCORER, "read_queries", side_effect=AssertionError("preloaded input")), \
+ patch.object(Path, "read_bytes", side_effect=AssertionError("whole-file read")), \
+ patch("sys.stderr", new=io.StringIO()) as progress:
+ result = SCORER.main(["--input", str(self.input), "--output", str(self.output),
+ "--metadata", str(self.metadata), "--cache-dir", str(self.root),
+ "--max-candidates", "500"])
+ self.assertEqual(result, 0)
+ self.assertEqual(calls, [500, 500])
+ self.assertIn("Scored 2/2 queries", progress.getvalue())
+ metadata = json.loads(self.metadata.read_text(encoding="utf-8"))
+ self.assertEqual(metadata["maxCandidates"], 500)
+ self.assertEqual(metadata["queryCount"], 2)
+ self.assertEqual(metadata["pairCount"], 1000)
+ self.assertEqual(metadata["inputSha256"], SCORER.file_sha256(self.input))
+ self.assertEqual(metadata["scoresSha256"], SCORER.file_sha256(self.output))
+ self.assertEqual(metadata["timingScope"], "complete_pool")
+ self.assertEqual(metadata["peakRssScope"], "process_lifetime_including_model_load")
+ self.assertGreater(metadata["peakRssBytes"], 0)
+ self.assertGreaterEqual(metadata["scoringWallNanos"], metadata["inferenceNanos"])
+ self.assertIn("excludes model load", metadata["scoringWallScope"])
+
+ def test_generator_without_length_and_count_mismatch_completion(self):
+ rows = (self.query("q" + str(index)) for index in range(2))
+ result = SCORER.write_scores(rows, "a" * 64, self.output, self.metadata,
+ lambda query, texts: [-1, 1])
+ self.assertEqual(result["queryCount"], 2)
+ self.assertEqual(result["pairCount"], 4)
+ self.output.unlink()
+ self.metadata.unlink()
+ with self.assertRaisesRegex(ValueError, "query count changed"):
+ SCORER.write_scores(iter([self.query()]), "a" * 64, self.output, self.metadata,
+ lambda query, texts: [-1, 1], query_count=2)
+ self.assertFalse(self.metadata.exists())
+
+ def test_input_change_after_validation_does_not_publish_completion(self):
+ row = self.query()
+ self.write_input([row])
+ digest = SCORER.file_sha256(self.input)
+
+ def tamper(query, texts):
+ changed = self.query()
+ changed["queryText"] = "changed while scoring"
+ self.write_input([changed])
+ return [-1, 1]
+
+ with self.assertRaisesRegex(ValueError, "Input changed after validation"):
+ SCORER.write_scores(iter([row]), digest, self.output, self.metadata, tamper,
+ input_path=self.input)
+ self.assertTrue(self.output.exists())
+ self.assertFalse(self.metadata.exists())
+
+ @unittest.skipUnless(SCORER.os.name == "nt", "Windows memory API")
+ def test_windows_peak_rss_uses_pointer_sized_process_handle(self):
+ import ctypes
+ from ctypes import wintypes
+
+ wide_handle = (1 << 48) + 17
+ kernel, psapi = Mock(), Mock()
+ kernel.GetCurrentProcess.return_value = wide_handle
+
+ def memory(handle, counters, size):
+ self.assertEqual(handle, wide_handle)
+ self.assertEqual(size, ctypes.sizeof(counters._obj))
+ counters._obj.peak_working_set = 123456
+ return 1
+
+ psapi.GetProcessMemoryInfo.side_effect = memory
+ with patch.object(ctypes, "WinDLL", side_effect=[kernel, psapi]):
+ self.assertEqual(SCORER.peak_rss_bytes(), 123456)
+ self.assertIs(kernel.GetCurrentProcess.restype, wintypes.HANDLE)
+ self.assertIs(psapi.GetProcessMemoryInfo.argtypes[0], wintypes.HANDLE)
+
+ def test_invalid_scores_leave_no_completion_metadata(self):
+ for values in ([float("nan"), 0], [float("inf"), 0], [True, 0], ["1", 0], [1]):
+ with self.subTest(values=values):
+ self.output.unlink(missing_ok=True)
+ with self.assertRaises(ValueError):
+ SCORER.write_scores([self.query()], "a" * 64, self.output, self.metadata,
+ lambda query, texts: values)
+ self.assertFalse(self.metadata.exists())
+
+ def test_model_failure_leaves_no_completion_metadata(self):
+ def fail(query, texts):
+ raise RuntimeError("fake inference failure")
+
+ with self.assertRaisesRegex(RuntimeError, "fake inference failure"):
+ SCORER.write_scores([self.query()], "a" * 64, self.output, self.metadata, fail)
+ self.assertFalse(self.metadata.exists())
+
+ def test_refuses_overwriting_either_output(self):
+ for existing in (self.output, self.metadata):
+ with self.subTest(existing=existing):
+ existing.write_text("keep this file", encoding="utf-8")
+ with self.assertRaises(FileExistsError):
+ SCORER.write_scores([self.query()], "a" * 64, self.output, self.metadata,
+ lambda query, texts: [0, 0])
+ self.assertEqual(existing.read_text(encoding="utf-8"), "keep this file")
+ existing.unlink()
+ with self.assertRaises(ValueError):
+ SCORER.write_scores([self.query()], "a" * 64, self.output, self.output,
+ lambda query, texts: [0, 0])
+
+ def test_check_environment_does_not_load_model(self):
+ with patch.object(SCORER, "check_environment", return_value={"torch": "2.8.0+cpu"}), \
+ patch.object(SCORER, "TransformerScorer", side_effect=AssertionError("loaded model")), \
+ patch("sys.stdout", new=io.StringIO()):
+ self.assertEqual(SCORER.main(["--check-environment"]), 0)
+
+ def test_cpu_environment_check_does_not_import_torch(self):
+ with patch.object(SCORER.importlib.metadata, "version",
+ side_effect=lambda name: SCORER.DEPENDENCIES[name]), \
+ patch.dict(sys.modules, {"torch": None}):
+ self.assertEqual(SCORER.check_environment(), SCORER.DEPENDENCIES)
+
+ def test_cuda_environment_requires_available_gpu_without_fallback(self):
+ for available, count, cuda_version in ((False, 0, None), (True, 0, "12.8"),
+ (True, 1, None)):
+ with self.subTest(available=available, count=count, cuda_version=cuda_version):
+ torch = SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: available,
+ device_count=lambda: count),
+ version=SimpleNamespace(cuda=cuda_version))
+ with patch.object(SCORER.importlib.metadata, "version",
+ side_effect=lambda name: SCORER.DEPENDENCIES[name]), \
+ patch.dict(sys.modules, {"torch": torch}), \
+ patch.dict(SCORER.os.environ):
+ with self.assertRaisesRegex(RuntimeError, "-Device cuda"):
+ SCORER.check_environment("cuda")
+
+ def test_cuda_environment_reports_actual_device_without_loading_model(self):
+ properties = SimpleNamespace(name="Fake GPU", total_memory=8589934592)
+ torch = SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: True,
+ device_count=lambda: 1,
+ get_device_properties=lambda index: properties),
+ version=SimpleNamespace(cuda="12.8"))
+ with patch.object(SCORER.importlib.metadata, "version",
+ side_effect=lambda name: SCORER.DEPENDENCIES[name]), \
+ patch.dict(sys.modules, {"torch": torch}), patch.dict(SCORER.os.environ), \
+ patch.object(SCORER, "TransformerScorer", side_effect=AssertionError("loaded model")):
+ result = SCORER.check_environment("cuda")
+ self.assertEqual(result["cudaVersion"], "12.8")
+ self.assertEqual(result["deviceName"], "Fake GPU")
+ self.assertEqual(result["deviceMemoryBytes"], 8589934592)
+ self.assertEqual(SCORER.os.environ["CUBLAS_WORKSPACE_CONFIG"], ":4096:8")
+
+ def test_cuda_scoring_synchronizes_timer_and_records_device(self):
+ events = []
+ scorer = Mock(side_effect=lambda query, texts: events.append("score") or [-1, 1])
+ scorer.synchronize.side_effect = lambda: events.append("sync")
+ ticks = iter((10, 30))
+
+ def clock():
+ events.append("clock")
+ return next(ticks)
+
+ versions = {"torch": "2.8.0+cu128", "transformers": "4.57.1", "cudaVersion": "12.8",
+ "deviceName": "Fake GPU", "deviceMemoryBytes": 8589934592}
+ with patch.object(SCORER.time, "perf_counter_ns", side_effect=clock):
+ result = SCORER.write_scores([self.query()], "a" * 64, self.output, self.metadata,
+ scorer, versions=versions, device="cuda")
+ self.assertEqual(events, ["sync", "clock", "score", "sync", "clock"])
+ self.assertEqual(result["device"], "cuda")
+ self.assertEqual(result["cudaVersion"], "12.8")
+ self.assertEqual(result["deviceName"], "Fake GPU")
+ self.assertEqual(result["deviceMemoryBytes"], 8589934592)
+ self.assertEqual(result["inferenceNanos"], 20)
+
+ def test_cuda_memory_telemetry_reports_allocated_reserved_and_scope(self):
+ scorer = Mock(return_value=[-1, 1])
+ scorer.memory_metrics.return_value = {"cudaPeakAllocatedBytes": 1234,
+ "cudaPeakReservedBytes": 2048,
+ "cudaPeakScope": "process_lifetime_including_model_load"}
+ versions = {"cudaVersion": "12.8", "deviceName": "Fake GPU",
+ "deviceMemoryBytes": 8589934592}
+ with patch.object(SCORER, "peak_rss_bytes", return_value=4096):
+ result = SCORER.write_scores(iter([self.query()]), "a" * 64,
+ self.output, self.metadata, scorer,
+ versions=versions, device="cuda")
+ self.assertEqual(result["peakRssBytes"], 4096)
+ self.assertEqual(result["cudaPeakAllocatedBytes"], 1234)
+ self.assertEqual(result["cudaPeakReservedBytes"], 2048)
+ self.assertEqual(result["cudaPeakScope"], "process_lifetime_including_model_load")
+ self.assertEqual(result["timingScope"], "complete_pool")
+ self.assertIn("every candidate", result["queryElapsedNanosScope"])
+
+ def test_cuda_scoring_requires_device_metadata_before_output(self):
+ with self.assertRaisesRegex(ValueError, "cudaVersion"):
+ SCORER.write_scores([self.query()], "a" * 64, self.output, self.metadata,
+ Mock(), device="cuda")
+ self.assertFalse(self.output.exists())
+ self.assertFalse(self.metadata.exists())
+
+ def test_transformer_moves_model_and_tokens_to_selected_device(self):
+ tensor = Mock()
+ tensor.to.return_value = tensor
+ logits = Mock()
+ logits.shape = (2, 1)
+ logits.__getitem__ = Mock(return_value=SimpleNamespace(
+ cpu=lambda: SimpleNamespace(tolist=lambda: [-1.5, 2.0])))
+ model = Mock()
+ model.config.num_labels = 1
+ model.return_value.logits = logits
+ tokenizer = Mock(return_value={"input_ids": tensor, "attention_mask": tensor})
+ torch = SimpleNamespace(manual_seed=Mock(), set_num_threads=Mock(),
+ set_num_interop_threads=Mock(), use_deterministic_algorithms=Mock(),
+ float32="float32", inference_mode=nullcontext,
+ cuda=SimpleNamespace(synchronize=Mock(),
+ max_memory_allocated=Mock(return_value=8192),
+ max_memory_reserved=Mock(return_value=16384)),
+ backends=SimpleNamespace(cuda=SimpleNamespace(matmul=SimpleNamespace()),
+ cudnn=SimpleNamespace()))
+ transformers = SimpleNamespace(
+ AutoTokenizer=SimpleNamespace(from_pretrained=Mock(return_value=tokenizer)),
+ AutoModelForSequenceClassification=SimpleNamespace(from_pretrained=Mock(return_value=model)))
+ with patch.dict(sys.modules, {"torch": torch, "transformers": transformers}), \
+ patch.dict(SCORER.os.environ):
+ scorer = SCORER.TransformerScorer(self.root, 16, 512, 1, True, device="cuda")
+ self.assertEqual(scorer("query", ["first", "second"]), [-1.5, 2.0])
+ scorer.synchronize()
+ self.assertEqual(scorer.memory_metrics(), {
+ "cudaPeakAllocatedBytes": 8192, "cudaPeakReservedBytes": 16384,
+ "cudaPeakScope": "process_lifetime_including_model_load"})
+ model.to.assert_called_once_with(device="cuda", dtype="float32")
+ self.assertEqual(tensor.to.call_count, 2)
+ tensor.to.assert_called_with(device="cuda")
+ torch.cuda.synchronize.assert_called_once_with()
+ self.assertFalse(torch.backends.cuda.matmul.allow_tf32)
+ self.assertFalse(torch.backends.cudnn.allow_tf32)
+ expected = {"revision": SCORER.MODEL_REVISION, "cache_dir": str(self.root),
+ "local_files_only": True, "trust_remote_code": False}
+ transformers.AutoTokenizer.from_pretrained.assert_called_once_with(SCORER.MODEL_ID,
+ **expected)
+ transformers.AutoModelForSequenceClassification.from_pretrained.assert_called_once_with(
+ SCORER.MODEL_ID, **expected, use_safetensors=True, attn_implementation="eager",
+ dtype="float32")
+
+ def test_main_passes_device_and_waits_for_model_load(self):
+ self.write_input([self.query()])
+ scorer = Mock(return_value=[-1, 1])
+ versions = {"cudaVersion": "12.8", "deviceName": "Fake GPU",
+ "deviceMemoryBytes": 8589934592}
+ with patch.object(SCORER, "check_environment", return_value=versions) as check, \
+ patch.object(SCORER, "TransformerScorer", return_value=scorer) as create, \
+ patch("sys.stderr", new=io.StringIO()):
+ result = SCORER.main(["--input", str(self.input), "--output", str(self.output),
+ "--metadata", str(self.metadata), "--cache-dir", str(self.root),
+ "--device", "cuda"])
+ self.assertEqual(result, 0)
+ check.assert_called_once_with("cuda")
+ create.assert_called_once_with(self.root, 16, 512, 1, False, device="cuda")
+ self.assertEqual(scorer.synchronize.call_count, 3)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/geaflow-ai/pom.xml b/geaflow-ai/pom.xml
index 006215b74..e05b5c0eb 100644
--- a/geaflow-ai/pom.xml
+++ b/geaflow-ai/pom.xml
@@ -84,6 +84,13 @@
${lucene.version}
+
+
+ org.apache.lucene
+ lucene-analyzers-common
+ ${lucene.version}
+
+
org.apache.lucene
@@ -121,8 +128,31 @@
geaflow-api
${project.version}
+
+ org.apache.geaflow
+ geaflow-dsl-common
+ ${project.version}
+
+
+ org.apache.geaflow
+ geaflow-pipeline
+ ${project.version}
+ test
+
+
+ org.apache.geaflow
+ geaflow-on-local
+ ${project.version}
+ test
+
+
+ org.apache.geaflow
+ geaflow-dsl-runtime
+ ${project.version}
+ test
+
org.junit.jupiter
junit-jupiter
@@ -146,6 +176,17 @@
+
+ org.apache.rat
+ apache-rat-plugin
+
+
+
+ src/main/resources/temporal/example-events.jsonl
+ src/main/resources/temporal/example-queries.jsonl
+
+
+
maven-assembly-plugin
@@ -154,9 +195,9 @@
org.apache.geaflow.ai.client.GeaFlowMemoryClientCLI
-
- jar-with-dependencies
-
+
+ src/assembly/jar-with-dependencies.xml
+
diff --git a/geaflow-ai/src/assembly/jar-with-dependencies.xml b/geaflow-ai/src/assembly/jar-with-dependencies.xml
new file mode 100644
index 000000000..665e3d177
--- /dev/null
+++ b/geaflow-ai/src/assembly/jar-with-dependencies.xml
@@ -0,0 +1,42 @@
+
+
+
+ jar-with-dependencies
+
+ jar
+
+ false
+
+
+
+ metaInf-services
+
+
+
+
+ /
+ true
+ true
+ runtime
+
+
+
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/adapter/MemoryGraphAdapter.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/adapter/MemoryGraphAdapter.java
new file mode 100644
index 000000000..4b4786ddc
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/adapter/MemoryGraphAdapter.java
@@ -0,0 +1,1253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.adapter;
+
+import java.time.Instant;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeMap;
+import org.apache.geaflow.ai.graph.io.Edge;
+import org.apache.geaflow.ai.graph.io.EdgeGroup;
+import org.apache.geaflow.ai.graph.io.EdgeSchema;
+import org.apache.geaflow.ai.graph.io.EntityGroup;
+import org.apache.geaflow.ai.graph.io.GraphSchema;
+import org.apache.geaflow.ai.graph.io.MemoryGraph;
+import org.apache.geaflow.ai.graph.io.Vertex;
+import org.apache.geaflow.ai.graph.io.VertexGroup;
+import org.apache.geaflow.ai.graph.io.VertexSchema;
+import org.apache.geaflow.ai.temporal.model.Evidence;
+import org.apache.geaflow.ai.temporal.model.FactKey;
+import org.apache.geaflow.ai.temporal.model.FactValue;
+import org.apache.geaflow.ai.temporal.model.MemoryEntity;
+import org.apache.geaflow.ai.temporal.model.MemoryEvent;
+import org.apache.geaflow.ai.temporal.model.MemoryEventOperation;
+import org.apache.geaflow.ai.temporal.model.MemoryFact;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersion;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersionStatus;
+import org.apache.geaflow.ai.temporal.model.Source;
+import org.apache.geaflow.ai.temporal.model.TimeInterval;
+import org.apache.geaflow.ai.temporal.model.VersionRelation;
+import org.apache.geaflow.ai.temporal.model.VersionRelationType;
+import org.apache.geaflow.ai.temporal.semantics.CanonicalSnapshot;
+import org.apache.geaflow.ai.temporal.semantics.EventNormalizer;
+import org.apache.geaflow.ai.temporal.semantics.NormalizedMemoryEvent;
+import org.apache.geaflow.ai.temporal.semantics.TemporalState;
+
+/**
+ * Projects canonical temporal snapshots to the existing in-memory graph model.
+ */
+public final class MemoryGraphAdapter {
+
+ private static final String ENTITY = "entity";
+ private static final String FACT_VERSION = "fact_version";
+ private static final String MEMORY_EVENT = "memory_event";
+ private static final String EVIDENCE = "evidence";
+ private static final String SOURCE = "source";
+
+ private static final String SUBJECT = "subject";
+ private static final String OBJECT = "object";
+ private static final String GENERATES = "generates";
+ private static final String SUPPORTED_BY = "supported_by";
+ private static final String FROM_SOURCE = "from_source";
+ private static final String SUPERSEDES = "supersedes";
+ private static final String DUPLICATE_OF = "duplicate_of";
+ private static final String CONFLICTS_WITH = "conflicts_with";
+
+ private static final String ENTITY_PREFIX = "entity:";
+ private static final String VERSION_PREFIX = "version:";
+ private static final String EVENT_PREFIX = "event:";
+ private static final String EVIDENCE_PREFIX = "evidence:";
+ private static final String SOURCE_PREFIX = "source:";
+ private static final String EMPTY = "";
+
+ private static final List ENTITY_FIELDS = fields("label");
+ private static final List VERSION_FIELDS = fields(
+ "factId",
+ "predicate",
+ "scope",
+ "valueKind",
+ "literalValue",
+ "status",
+ "validStart",
+ "validEnd",
+ "transactionStart",
+ "transactionEnd");
+ private static final List EVENT_FIELDS = fields(
+ "operation",
+ "factId",
+ "subjectId",
+ "predicate",
+ "scope",
+ "valueKind",
+ "value",
+ "validStart",
+ "validEnd",
+ "recordedAt",
+ "targetVersionId",
+ "payloadHash");
+ private static final List EVIDENCE_FIELDS = fields("content");
+ private static final List SOURCE_FIELDS = fields("name");
+ private static final List NO_FIELDS = Collections.emptyList();
+ private static final List COUNT_FIELDS = fields(
+ "occurrenceCount");
+ private static final List RELATION_FIELDS = fields(
+ "relationId", "occurrenceCount");
+
+ private static final List VERTEX_LABELS = fields(
+ ENTITY, FACT_VERSION, MEMORY_EVENT, EVIDENCE, SOURCE);
+ private static final List EDGE_LABELS = fields(
+ SUBJECT,
+ OBJECT,
+ GENERATES,
+ SUPPORTED_BY,
+ FROM_SOURCE,
+ SUPERSEDES,
+ DUPLICATE_OF,
+ CONFLICTS_WITH);
+
+ private static final Comparator VERTEX_ORDER =
+ Comparator.comparing(Vertex::getId);
+ private static final Comparator EDGE_ORDER =
+ Comparator.comparing(Edge::getSrcId)
+ .thenComparing(Edge::getDstId)
+ .thenComparing(edge -> edge.getValues().toString());
+
+ public MemoryGraph toGraph(CanonicalSnapshot snapshot) {
+ Objects.requireNonNull(snapshot, "snapshot");
+
+ Map memoryEntities = new TreeMap<>();
+ Map evidenceById = new TreeMap<>();
+ Map sources = new TreeMap<>();
+ Map versionVertices = new TreeMap<>();
+ Map eventVertices = new TreeMap<>();
+ Map> edges = emptyEdgeLists();
+ Map supported = new TreeMap<>();
+ Map> relationEdges =
+ relationEdgeMaps();
+
+ for (NormalizedMemoryEvent event : snapshot.getEvents()) {
+ collectEventEntities(event, memoryEntities);
+ collectEvidence(event.getEvidence(), evidenceById, sources);
+ String eventGraphId = eventId(event.getEventId());
+ putUniqueVertex(
+ eventVertices,
+ new Vertex(MEMORY_EVENT, eventGraphId, eventValues(event)));
+ for (Evidence evidence : event.getEvidence()) {
+ addCountedEdge(
+ supported,
+ eventGraphId,
+ evidenceId(evidence.getId()));
+ }
+ }
+
+ for (Map.Entry> entry
+ : snapshot.getState().getVersionsByFactKey().entrySet()) {
+ FactKey key = entry.getKey();
+ for (MemoryFactVersion version : entry.getValue()) {
+ MemoryFact fact = version.getFact();
+ collectEntity(memoryEntities, fact.getSubject());
+ if (fact.isRelationship()) {
+ collectEntity(memoryEntities, fact.getTarget().get());
+ }
+ collectEvidence(version.getEvidence(), evidenceById, sources);
+
+ String versionGraphId = versionId(version.getId());
+ putUniqueVertex(
+ versionVertices,
+ new Vertex(
+ FACT_VERSION,
+ versionGraphId,
+ versionValues(version, key)));
+ edges.get(SUBJECT).add(new Edge(
+ SUBJECT,
+ versionGraphId,
+ entityId(fact.getSubject().getId()),
+ NO_FIELDS));
+ if (fact.isRelationship()) {
+ edges.get(OBJECT).add(new Edge(
+ OBJECT,
+ versionGraphId,
+ entityId(fact.getTarget().get().getId()),
+ NO_FIELDS));
+ }
+ String generatingEventId =
+ snapshot.getGeneratingEventIds().get(version.getId());
+ edges.get(GENERATES).add(new Edge(
+ GENERATES,
+ eventId(generatingEventId),
+ versionGraphId,
+ NO_FIELDS));
+ for (Evidence evidence : version.getEvidence()) {
+ addCountedEdge(
+ supported,
+ versionGraphId,
+ evidenceId(evidence.getId()));
+ }
+ }
+ }
+
+ for (VersionRelation relation
+ : snapshot.getState().getRelations()) {
+ String label = relationLabel(relation.getType());
+ addCountedEdge(
+ relationEdges.get(label),
+ versionId(relation.getFromVersionId()),
+ versionId(relation.getToVersionId()));
+ }
+
+ List entityVertices = new ArrayList<>();
+ for (MemoryEntity entity : memoryEntities.values()) {
+ entityVertices.add(new Vertex(
+ ENTITY,
+ entityId(entity.getId()),
+ fields(entity.getLabel())));
+ }
+ List evidenceVertices = new ArrayList<>();
+ for (Evidence evidence : evidenceById.values()) {
+ evidenceVertices.add(new Vertex(
+ EVIDENCE,
+ evidenceId(evidence.getId()),
+ fields(evidence.getContent())));
+ edges.get(FROM_SOURCE).add(new Edge(
+ FROM_SOURCE,
+ evidenceId(evidence.getId()),
+ sourceId(evidence.getSource().getId()),
+ NO_FIELDS));
+ }
+ List sourceVertices = new ArrayList<>();
+ for (Source source : sources.values()) {
+ sourceVertices.add(new Vertex(
+ SOURCE,
+ sourceId(source.getId()),
+ fields(source.getName())));
+ }
+
+ edges.put(
+ SUPPORTED_BY,
+ countedEdges(SUPPORTED_BY, supported, false));
+ for (String label : Arrays.asList(
+ SUPERSEDES, DUPLICATE_OF, CONFLICTS_WITH)) {
+ edges.put(
+ label,
+ countedEdges(label, relationEdges.get(label), true));
+ }
+
+ Map> vertices = new LinkedHashMap<>();
+ vertices.put(ENTITY, entityVertices);
+ vertices.put(
+ FACT_VERSION,
+ new ArrayList<>(versionVertices.values()));
+ vertices.put(
+ MEMORY_EVENT,
+ new ArrayList<>(eventVertices.values()));
+ vertices.put(EVIDENCE, evidenceVertices);
+ vertices.put(SOURCE, sourceVertices);
+ return createGraph(vertices, edges);
+ }
+
+ public CanonicalSnapshot fromGraph(MemoryGraph graph) {
+ Objects.requireNonNull(graph, "graph");
+ validateGraphShape(graph);
+
+ Map entityVertices =
+ vertexIndex(graph, ENTITY, ENTITY_PREFIX, ENTITY_FIELDS);
+ Map versionVertices =
+ vertexIndex(
+ graph,
+ FACT_VERSION,
+ VERSION_PREFIX,
+ VERSION_FIELDS);
+ Map eventVertices =
+ vertexIndex(
+ graph,
+ MEMORY_EVENT,
+ EVENT_PREFIX,
+ EVENT_FIELDS);
+ Map evidenceVertices =
+ vertexIndex(
+ graph,
+ EVIDENCE,
+ EVIDENCE_PREFIX,
+ EVIDENCE_FIELDS);
+ Map sourceVertices =
+ vertexIndex(graph, SOURCE, SOURCE_PREFIX, SOURCE_FIELDS);
+ Map> edges = edgeIndex(graph);
+
+ Map memoryEntities =
+ readEntities(entityVertices);
+ Map sources = readSources(sourceVertices);
+ Map evidence = readEvidence(
+ evidenceVertices,
+ sources,
+ edges.get(FROM_SOURCE));
+ Map> evidenceByOwner =
+ readSupportedEvidence(
+ edges.get(SUPPORTED_BY),
+ eventVertices,
+ versionVertices,
+ evidence);
+
+ Map events = readEvents(
+ eventVertices,
+ memoryEntities,
+ evidenceByOwner);
+ Map> versions = readVersions(
+ versionVertices,
+ memoryEntities,
+ evidenceByOwner,
+ edges.get(SUBJECT),
+ edges.get(OBJECT));
+ validateNoOrphanVertices(
+ memoryEntities,
+ evidence,
+ sources,
+ events,
+ versions);
+ Map generating = readGeneratingEvents(
+ edges.get(GENERATES),
+ events,
+ versionVertices);
+ List relations = readRelations(
+ edges,
+ versionVertices);
+
+ return new CanonicalSnapshot(
+ new TemporalState(versions, relations),
+ new ArrayList<>(events.values()),
+ generating);
+ }
+
+ private static MemoryGraph createGraph(
+ Map> vertices,
+ Map> edges) {
+ GraphSchema schema = new GraphSchema();
+ Map vertexSchemas = new LinkedHashMap<>();
+ addVertexSchema(schema, vertexSchemas, ENTITY, ENTITY_FIELDS);
+ addVertexSchema(
+ schema, vertexSchemas, FACT_VERSION, VERSION_FIELDS);
+ addVertexSchema(
+ schema, vertexSchemas, MEMORY_EVENT, EVENT_FIELDS);
+ addVertexSchema(schema, vertexSchemas, EVIDENCE, EVIDENCE_FIELDS);
+ addVertexSchema(schema, vertexSchemas, SOURCE, SOURCE_FIELDS);
+
+ Map edgeSchemas = new LinkedHashMap<>();
+ addEdgeSchema(schema, edgeSchemas, SUBJECT, NO_FIELDS);
+ addEdgeSchema(schema, edgeSchemas, OBJECT, NO_FIELDS);
+ addEdgeSchema(schema, edgeSchemas, GENERATES, NO_FIELDS);
+ addEdgeSchema(schema, edgeSchemas, SUPPORTED_BY, COUNT_FIELDS);
+ addEdgeSchema(schema, edgeSchemas, FROM_SOURCE, NO_FIELDS);
+ addEdgeSchema(schema, edgeSchemas, SUPERSEDES, RELATION_FIELDS);
+ addEdgeSchema(schema, edgeSchemas, DUPLICATE_OF, RELATION_FIELDS);
+ addEdgeSchema(
+ schema, edgeSchemas, CONFLICTS_WITH, RELATION_FIELDS);
+
+ Map groups = new LinkedHashMap<>();
+ for (String label : VERTEX_LABELS) {
+ List rows = vertices.get(label);
+ Collections.sort(rows, VERTEX_ORDER);
+ groups.put(
+ label,
+ new VertexGroup(vertexSchemas.get(label), rows));
+ }
+ for (String label : EDGE_LABELS) {
+ List rows = edges.get(label);
+ Collections.sort(rows, EDGE_ORDER);
+ groups.put(
+ label,
+ new EdgeGroup(edgeSchemas.get(label), rows));
+ }
+ return new MemoryGraph(schema, groups);
+ }
+
+ private static void addVertexSchema(
+ GraphSchema graphSchema,
+ Map schemas,
+ String label,
+ List fields) {
+ VertexSchema schema = new VertexSchema(label, "id", fields);
+ graphSchema.addVertex(schema);
+ schemas.put(label, schema);
+ }
+
+ private static void addEdgeSchema(
+ GraphSchema graphSchema,
+ Map schemas,
+ String label,
+ List fields) {
+ EdgeSchema schema = new EdgeSchema(
+ label, "srcId", "dstId", fields);
+ graphSchema.addEdge(schema);
+ schemas.put(label, schema);
+ }
+
+ private static Map> emptyEdgeLists() {
+ Map> edges = new LinkedHashMap<>();
+ for (String label : EDGE_LABELS) {
+ edges.put(label, new ArrayList<>());
+ }
+ return edges;
+ }
+
+ private static Map>
+ relationEdgeMaps() {
+ Map> relations =
+ new LinkedHashMap<>();
+ relations.put(SUPERSEDES, new TreeMap<>());
+ relations.put(DUPLICATE_OF, new TreeMap<>());
+ relations.put(CONFLICTS_WITH, new TreeMap<>());
+ return relations;
+ }
+
+ private static List countedEdges(
+ String label,
+ Map counted,
+ boolean relation) {
+ List edges = new ArrayList<>();
+ for (CountedEdge item : counted.values()) {
+ List values = relation
+ ? fields(
+ relationId(label, item.sourceId, item.targetId),
+ Integer.toString(item.count))
+ : fields(Integer.toString(item.count));
+ edges.add(new Edge(
+ label, item.sourceId, item.targetId, values));
+ }
+ return edges;
+ }
+
+ private static void addCountedEdge(
+ Map edges,
+ String sourceId,
+ String targetId) {
+ String key = tuple(sourceId, targetId);
+ CountedEdge current = edges.get(key);
+ if (current == null) {
+ edges.put(key, new CountedEdge(sourceId, targetId));
+ } else {
+ current.count++;
+ }
+ }
+
+ private static List eventValues(NormalizedMemoryEvent event) {
+ String kind = EMPTY;
+ String value = EMPTY;
+ if (event.getFactValue().isPresent()) {
+ kind = event.getFactValue().get().getKind().name();
+ value = event.getFactValue().get().getValue();
+ }
+ return fields(
+ event.getOperation().name(),
+ event.getFactId(),
+ event.getFactKey().getSubjectId(),
+ event.getFactKey().getPredicate(),
+ event.getFactKey().getScope(),
+ kind,
+ value,
+ instant(event.getValidTime().getStart()),
+ optionalInstant(event.getValidTime()),
+ instant(event.getRecordedAt()),
+ event.getTargetVersionId().orElse(EMPTY),
+ event.getPayloadHash());
+ }
+
+ private static List versionValues(
+ MemoryFactVersion version,
+ FactKey key) {
+ MemoryFact fact = version.getFact();
+ String kind = fact.isRelationship()
+ ? FactValue.Kind.ENTITY_REF.name()
+ : FactValue.Kind.LITERAL.name();
+ String literalValue = fact.getLiteralValue().orElse(EMPTY);
+ return fields(
+ fact.getId(),
+ fact.getPredicate(),
+ key.getScope(),
+ kind,
+ literalValue,
+ version.getStatus().name(),
+ instant(version.getValidTime().getStart()),
+ optionalInstant(version.getValidTime()),
+ instant(version.getTransactionTime().getStart()),
+ optionalInstant(version.getTransactionTime()));
+ }
+
+ private static void collectEventEntities(
+ NormalizedMemoryEvent event,
+ Map entities) {
+ if (!event.getEvent().getFact().isPresent()) {
+ return;
+ }
+ MemoryFact fact = event.getEvent().getFact().get();
+ collectEntity(entities, fact.getSubject());
+ if (fact.isRelationship()) {
+ collectEntity(entities, fact.getTarget().get());
+ }
+ }
+
+ private static void collectEntity(
+ Map entities,
+ MemoryEntity entity) {
+ putUnique(entities, entity.getId(), entity, "entity");
+ }
+
+ private static void collectEvidence(
+ List evidence,
+ Map evidenceById,
+ Map sources) {
+ for (Evidence item : evidence) {
+ putUnique(evidenceById, item.getId(), item, "evidence");
+ putUnique(
+ sources,
+ item.getSource().getId(),
+ item.getSource(),
+ "source");
+ }
+ }
+
+ private static void putUnique(
+ Map values,
+ String id,
+ T value,
+ String type) {
+ T previous = values.get(id);
+ if (previous != null && !previous.equals(value)) {
+ throw new IllegalArgumentException(
+ "Conflicting " + type + " id: " + id);
+ }
+ values.put(id, value);
+ }
+
+ private static void putUniqueVertex(
+ Map vertices,
+ Vertex vertex) {
+ Vertex previous = vertices.put(vertex.getId(), vertex);
+ if (previous != null
+ && !previous.getValues().equals(vertex.getValues())) {
+ throw new IllegalArgumentException(
+ "Conflicting vertex id: " + vertex.getId());
+ }
+ }
+
+ private static void validateGraphShape(MemoryGraph graph) {
+ GraphSchema schema = Objects.requireNonNull(
+ graph.getGraphSchema(), "graphSchema");
+ require(
+ schema.getVertexSchemaList().size() == VERTEX_LABELS.size(),
+ "Unexpected vertex schema count");
+ require(
+ schema.getEdgeSchemaList().size() == EDGE_LABELS.size(),
+ "Unexpected edge schema count");
+
+ for (int index = 0; index < VERTEX_LABELS.size(); index++) {
+ String label = VERTEX_LABELS.get(index);
+ VertexSchema actual = schema.getVertexSchemaList().get(index);
+ require(label.equals(actual.getLabel()),
+ "Unexpected vertex schema: " + actual.getLabel());
+ require("id".equals(actual.getIdField()),
+ "Unexpected vertex id field: " + label);
+ require(vertexFields(label).equals(actual.getFields()),
+ "Unexpected vertex fields: " + label);
+ }
+ for (int index = 0; index < EDGE_LABELS.size(); index++) {
+ String label = EDGE_LABELS.get(index);
+ EdgeSchema actual = schema.getEdgeSchemaList().get(index);
+ require(label.equals(actual.getLabel()),
+ "Unexpected edge schema: " + actual.getLabel());
+ require("srcId".equals(actual.getSrcIdField()),
+ "Unexpected edge source field: " + label);
+ require("dstId".equals(actual.getDstIdField()),
+ "Unexpected edge target field: " + label);
+ require(edgeFields(label).equals(actual.getFields()),
+ "Unexpected edge fields: " + label);
+ }
+
+ Map groups = Objects.requireNonNull(
+ graph.entities, "graph.entities");
+ List expectedGroups = new ArrayList<>(VERTEX_LABELS);
+ expectedGroups.addAll(EDGE_LABELS);
+ require(
+ expectedGroups.equals(new ArrayList<>(groups.keySet())),
+ "Unexpected graph entity groups");
+ for (String label : VERTEX_LABELS) {
+ require(groups.get(label) instanceof VertexGroup,
+ "Expected vertex group: " + label);
+ }
+ for (String label : EDGE_LABELS) {
+ require(groups.get(label) instanceof EdgeGroup,
+ "Expected edge group: " + label);
+ }
+ }
+
+ private static Map vertexIndex(
+ MemoryGraph graph,
+ String label,
+ String prefix,
+ List expectedFields) {
+ Map result = new TreeMap<>();
+ for (Vertex vertex
+ : ((VertexGroup) graph.entities.get(label)).getVertices()) {
+ require(vertex != null, "Null vertex in group: " + label);
+ require(label.equals(vertex.getLabel()),
+ "Vertex label does not match group: " + vertex.getId());
+ rawId(vertex.getId(), prefix);
+ require(vertex.getValues() != null,
+ "Null vertex values: " + vertex.getId());
+ require(vertex.getValues().size() == expectedFields.size(),
+ "Unexpected vertex value count: " + vertex.getId());
+ for (String fieldValue : vertex.getValues()) {
+ require(fieldValue != null,
+ "Null vertex field: " + vertex.getId());
+ }
+ require(result.put(vertex.getId(), vertex) == null,
+ "Duplicate vertex id: " + vertex.getId());
+ }
+ return result;
+ }
+
+ private static Map> edgeIndex(MemoryGraph graph) {
+ Map> result = new LinkedHashMap<>();
+ for (String label : EDGE_LABELS) {
+ List rows = new ArrayList<>();
+ Set identities = new HashSet<>();
+ for (Edge edge
+ : ((EdgeGroup) graph.entities.get(label)).getOutEdges()) {
+ require(edge != null, "Null edge in group: " + label);
+ require(label.equals(edge.getLabel()),
+ "Edge label does not match group: " + label);
+ require(edge.getSrcId() != null && edge.getDstId() != null,
+ "Null edge endpoint: " + label);
+ require(edge.getValues() != null,
+ "Null edge values: " + label);
+ require(edge.getValues().size() == edgeFields(label).size(),
+ "Unexpected edge value count: " + label);
+ for (String fieldValue : edge.getValues()) {
+ require(fieldValue != null,
+ "Null edge field: " + label);
+ }
+ require(identities.add(edge),
+ "Duplicate edge identity: " + edge);
+ rows.add(edge);
+ }
+ Collections.sort(rows, EDGE_ORDER);
+ result.put(label, rows);
+ }
+ return result;
+ }
+
+ private static Map readEntities(
+ Map vertices) {
+ Map result = new TreeMap<>();
+ for (Vertex vertex : vertices.values()) {
+ String rawId = rawId(vertex.getId(), ENTITY_PREFIX);
+ result.put(
+ vertex.getId(),
+ new MemoryEntity(
+ rawId,
+ value(vertex, ENTITY_FIELDS, "label")));
+ }
+ return result;
+ }
+
+ private static Map readSources(
+ Map vertices) {
+ Map result = new TreeMap<>();
+ for (Vertex vertex : vertices.values()) {
+ String rawId = rawId(vertex.getId(), SOURCE_PREFIX);
+ result.put(
+ vertex.getId(),
+ new Source(
+ rawId,
+ value(vertex, SOURCE_FIELDS, "name")));
+ }
+ return result;
+ }
+
+ private static Map readEvidence(
+ Map vertices,
+ Map sources,
+ List fromSourceEdges) {
+ Map sourceByEvidence = new HashMap<>();
+ for (Edge edge : fromSourceEdges) {
+ require(vertices.containsKey(edge.getSrcId()),
+ "Unknown evidence in from_source edge");
+ require(sources.containsKey(edge.getDstId()),
+ "Unknown source in from_source edge");
+ require(sourceByEvidence.put(
+ edge.getSrcId(), edge.getDstId()) == null,
+ "Evidence has multiple sources: " + edge.getSrcId());
+ }
+ require(sourceByEvidence.keySet().equals(vertices.keySet()),
+ "Every evidence must have exactly one source");
+
+ Map result = new TreeMap<>();
+ for (Vertex vertex : vertices.values()) {
+ String rawId = rawId(vertex.getId(), EVIDENCE_PREFIX);
+ result.put(
+ vertex.getId(),
+ new Evidence(
+ rawId,
+ sources.get(sourceByEvidence.get(vertex.getId())),
+ value(vertex, EVIDENCE_FIELDS, "content")));
+ }
+ return result;
+ }
+
+ private static Map> readSupportedEvidence(
+ List supportedEdges,
+ Map eventVertices,
+ Map versionVertices,
+ Map evidence) {
+ Map> result = new HashMap<>();
+ for (Edge edge : supportedEdges) {
+ require(
+ eventVertices.containsKey(edge.getSrcId())
+ || versionVertices.containsKey(edge.getSrcId()),
+ "Unknown supported_by owner: " + edge.getSrcId());
+ Evidence item = evidence.get(edge.getDstId());
+ require(item != null,
+ "Unknown supported_by evidence: " + edge.getDstId());
+ int count = positiveCount(edge.getValues().get(0));
+ List ownerEvidence = result.computeIfAbsent(
+ edge.getSrcId(), ignored -> new ArrayList<>());
+ for (int occurrence = 0; occurrence < count; occurrence++) {
+ ownerEvidence.add(item);
+ }
+ }
+ return result;
+ }
+
+ private static Map readEvents(
+ Map vertices,
+ Map entities,
+ Map> evidenceByOwner) {
+ Map result = new TreeMap<>();
+ EventNormalizer normalizer = new EventNormalizer();
+ for (Vertex vertex : vertices.values()) {
+ String rawEventId = rawId(vertex.getId(), EVENT_PREFIX);
+ MemoryEventOperation operation = enumValue(
+ MemoryEventOperation.class,
+ value(vertex, EVENT_FIELDS, "operation"),
+ "event operation");
+ String factId = value(vertex, EVENT_FIELDS, "factId");
+ String subjectId = value(vertex, EVENT_FIELDS, "subjectId");
+ String predicate = value(vertex, EVENT_FIELDS, "predicate");
+ final FactKey key = new FactKey(
+ subjectId,
+ predicate,
+ value(vertex, EVENT_FIELDS, "scope"));
+ TimeInterval validTime = interval(
+ value(vertex, EVENT_FIELDS, "validStart"),
+ value(vertex, EVENT_FIELDS, "validEnd"));
+ Instant recordedAt = parseInstant(
+ value(vertex, EVENT_FIELDS, "recordedAt"));
+ List eventEvidence = evidenceByOwner.get(
+ vertex.getId());
+ require(eventEvidence != null && !eventEvidence.isEmpty(),
+ "Memory event must have evidence: " + rawEventId);
+
+ MemoryEvent event;
+ if (operation == MemoryEventOperation.RETRACT) {
+ require(value(vertex, EVENT_FIELDS, "valueKind").isEmpty(),
+ "Retract event must not have a value kind");
+ require(value(vertex, EVENT_FIELDS, "value").isEmpty(),
+ "Retract event must not have a value");
+ event = MemoryEvent.retract(
+ rawEventId,
+ factId,
+ validTime,
+ recordedAt,
+ eventEvidence);
+ } else {
+ MemoryFact fact = eventFact(
+ vertex,
+ factId,
+ subjectId,
+ predicate,
+ entities);
+ if (operation == MemoryEventOperation.ADD) {
+ event = MemoryEvent.add(
+ rawEventId,
+ fact,
+ validTime,
+ recordedAt,
+ eventEvidence);
+ } else {
+ require(operation == MemoryEventOperation.CORRECT,
+ "Unsupported memory event operation");
+ event = MemoryEvent.correct(
+ rawEventId,
+ fact,
+ validTime,
+ recordedAt,
+ eventEvidence);
+ }
+ }
+
+ String targetVersionId = value(vertex, EVENT_FIELDS, "targetVersionId");
+ if (!targetVersionId.isEmpty()) {
+ event = event.targeting(targetVersionId);
+ }
+ NormalizedMemoryEvent normalized = normalizer.normalize(event, key);
+ require(eventId(normalized.getEventId()).equals(vertex.getId()),
+ "Event id is not canonical: " + rawEventId);
+ require(
+ normalized.getPayloadHash().equals(
+ value(vertex, EVENT_FIELDS, "payloadHash")),
+ "Event payload hash does not match: " + rawEventId);
+ result.put(vertex.getId(), normalized);
+ }
+ return result;
+ }
+
+ private static MemoryFact eventFact(
+ Vertex vertex,
+ String factId,
+ String subjectId,
+ String predicate,
+ Map entities) {
+ MemoryEntity subject = entities.get(entityId(subjectId));
+ require(subject != null,
+ "Unknown event subject: " + subjectId);
+ FactValue.Kind kind = enumValue(
+ FactValue.Kind.class,
+ value(vertex, EVENT_FIELDS, "valueKind"),
+ "event value kind");
+ String factValue = value(vertex, EVENT_FIELDS, "value");
+ if (kind == FactValue.Kind.LITERAL) {
+ return MemoryFact.attribute(
+ factId, subject, predicate, factValue);
+ }
+ MemoryEntity target = entities.get(entityId(factValue));
+ require(target != null,
+ "Unknown event object: " + factValue);
+ return MemoryFact.relationship(
+ factId, subject, predicate, target);
+ }
+
+ private static Map> readVersions(
+ Map vertices,
+ Map entities,
+ Map> evidenceByOwner,
+ List subjectEdges,
+ List objectEdges) {
+ Map subjects = uniqueTargets(
+ subjectEdges,
+ vertices,
+ entities,
+ "subject");
+ require(subjects.keySet().equals(vertices.keySet()),
+ "Every fact version must have exactly one subject");
+ Map objects = uniqueTargets(
+ objectEdges,
+ vertices,
+ entities,
+ "object");
+
+ Map> result = new TreeMap<>();
+ for (Vertex vertex : vertices.values()) {
+ String rawVersionId = rawId(vertex.getId(), VERSION_PREFIX);
+ MemoryEntity subject = entities.get(subjects.get(vertex.getId()));
+ String predicate = value(vertex, VERSION_FIELDS, "predicate");
+ FactKey key = new FactKey(
+ subject.getId(),
+ predicate,
+ value(vertex, VERSION_FIELDS, "scope"));
+
+ FactValue.Kind kind = enumValue(
+ FactValue.Kind.class,
+ value(vertex, VERSION_FIELDS, "valueKind"),
+ "version value kind");
+ String factId = value(vertex, VERSION_FIELDS, "factId");
+ String literalValue = value(
+ vertex, VERSION_FIELDS, "literalValue");
+ MemoryFact fact;
+ if (kind == FactValue.Kind.LITERAL) {
+ require(!objects.containsKey(vertex.getId()),
+ "Literal version must not have an object");
+ fact = MemoryFact.attribute(
+ factId, subject, predicate, literalValue);
+ } else {
+ require(literalValue.isEmpty(),
+ "Entity-reference version must not have a literal value");
+ String objectId = objects.get(vertex.getId());
+ require(objectId != null,
+ "Entity-reference version must have an object");
+ fact = MemoryFact.relationship(
+ factId, subject, predicate, entities.get(objectId));
+ }
+
+ List versionEvidence = evidenceByOwner.get(
+ vertex.getId());
+ require(versionEvidence != null && !versionEvidence.isEmpty(),
+ "Fact version must have evidence: " + rawVersionId);
+ MemoryFactVersion version = new MemoryFactVersion(
+ rawVersionId,
+ fact,
+ enumValue(
+ MemoryFactVersionStatus.class,
+ value(vertex, VERSION_FIELDS, "status"),
+ "version status"),
+ interval(
+ value(vertex, VERSION_FIELDS, "validStart"),
+ value(vertex, VERSION_FIELDS, "validEnd")),
+ interval(
+ value(vertex, VERSION_FIELDS, "transactionStart"),
+ value(vertex, VERSION_FIELDS, "transactionEnd")),
+ versionEvidence);
+ result.computeIfAbsent(
+ key, ignored -> new ArrayList<>()).add(version);
+ }
+ return result;
+ }
+
+ private static Map uniqueTargets(
+ List edges,
+ Map sourceVertices,
+ Map targets,
+ String relationName) {
+ Map result = new HashMap<>();
+ for (Edge edge : edges) {
+ require(sourceVertices.containsKey(edge.getSrcId()),
+ "Unknown " + relationName + " source");
+ require(targets.containsKey(edge.getDstId()),
+ "Unknown " + relationName + " target");
+ require(result.put(edge.getSrcId(), edge.getDstId()) == null,
+ "Multiple " + relationName + " targets");
+ }
+ return result;
+ }
+
+ private static void validateNoOrphanVertices(
+ Map entities,
+ Map evidence,
+ Map sources,
+ Map events,
+ Map> versions) {
+ Map referencedEntities = new TreeMap<>();
+ Map referencedEvidence = new TreeMap<>();
+ Map referencedSources = new TreeMap<>();
+ for (NormalizedMemoryEvent event : events.values()) {
+ collectEventEntities(event, referencedEntities);
+ collectEvidence(
+ event.getEvidence(),
+ referencedEvidence,
+ referencedSources);
+ }
+ for (List factVersions : versions.values()) {
+ for (MemoryFactVersion version : factVersions) {
+ MemoryFact fact = version.getFact();
+ collectEntity(referencedEntities, fact.getSubject());
+ if (fact.isRelationship()) {
+ collectEntity(
+ referencedEntities,
+ fact.getTarget().get());
+ }
+ collectEvidence(
+ version.getEvidence(),
+ referencedEvidence,
+ referencedSources);
+ }
+ }
+ requireAllReferenced(
+ entities, referencedEntities, ENTITY_PREFIX, "entity");
+ requireAllReferenced(
+ evidence, referencedEvidence, EVIDENCE_PREFIX, "evidence");
+ requireAllReferenced(
+ sources, referencedSources, SOURCE_PREFIX, "source");
+ }
+
+ private static void requireAllReferenced(
+ Map graphValues,
+ Map referencedValues,
+ String prefix,
+ String type) {
+ for (String graphId : graphValues.keySet()) {
+ require(referencedValues.containsKey(rawId(graphId, prefix)),
+ "Orphan " + type + " vertex: " + graphId);
+ }
+ }
+
+ private static Map readGeneratingEvents(
+ List generateEdges,
+ Map events,
+ Map versions) {
+ Map byVersion = new HashMap<>();
+ for (Edge edge : generateEdges) {
+ require(events.containsKey(edge.getSrcId()),
+ "Unknown generating event: " + edge.getSrcId());
+ require(versions.containsKey(edge.getDstId()),
+ "Unknown generated version: " + edge.getDstId());
+ require(byVersion.put(
+ edge.getDstId(), edge.getSrcId()) == null,
+ "Version has multiple generating events: "
+ + edge.getDstId());
+ }
+ require(byVersion.keySet().equals(versions.keySet()),
+ "Every fact version must have one generating event");
+
+ Map result = new TreeMap<>();
+ for (Map.Entry entry : byVersion.entrySet()) {
+ result.put(
+ rawId(entry.getKey(), VERSION_PREFIX),
+ rawId(entry.getValue(), EVENT_PREFIX));
+ }
+ return result;
+ }
+
+ private static List readRelations(
+ Map> edges,
+ Map versions) {
+ List result = new ArrayList<>();
+ for (String label : Arrays.asList(
+ SUPERSEDES, DUPLICATE_OF, CONFLICTS_WITH)) {
+ for (Edge edge : edges.get(label)) {
+ require(versions.containsKey(edge.getSrcId()),
+ "Unknown relation source: " + edge.getSrcId());
+ require(versions.containsKey(edge.getDstId()),
+ "Unknown relation target: " + edge.getDstId());
+ require(
+ relationId(label, edge.getSrcId(), edge.getDstId())
+ .equals(edge.getValues().get(0)),
+ "Version relation id does not match endpoints");
+ int count = positiveCount(edge.getValues().get(1));
+ for (int occurrence = 0; occurrence < count; occurrence++) {
+ VersionRelation relation = new VersionRelation(
+ relationType(label),
+ rawId(edge.getSrcId(), VERSION_PREFIX),
+ rawId(edge.getDstId(), VERSION_PREFIX));
+ require(
+ versionId(relation.getFromVersionId()).equals(
+ edge.getSrcId())
+ && versionId(relation.getToVersionId()).equals(
+ edge.getDstId()),
+ "Version relation endpoints are not canonical");
+ result.add(relation);
+ }
+ }
+ }
+ return result;
+ }
+
+ private static List vertexFields(String label) {
+ if (ENTITY.equals(label)) {
+ return ENTITY_FIELDS;
+ }
+ if (FACT_VERSION.equals(label)) {
+ return VERSION_FIELDS;
+ }
+ if (MEMORY_EVENT.equals(label)) {
+ return EVENT_FIELDS;
+ }
+ if (EVIDENCE.equals(label)) {
+ return EVIDENCE_FIELDS;
+ }
+ if (SOURCE.equals(label)) {
+ return SOURCE_FIELDS;
+ }
+ throw new IllegalArgumentException(
+ "Unknown vertex label: " + label);
+ }
+
+ private static List edgeFields(String label) {
+ if (SUPPORTED_BY.equals(label)) {
+ return COUNT_FIELDS;
+ }
+ if (SUPERSEDES.equals(label)
+ || DUPLICATE_OF.equals(label)
+ || CONFLICTS_WITH.equals(label)) {
+ return RELATION_FIELDS;
+ }
+ if (SUBJECT.equals(label)
+ || OBJECT.equals(label)
+ || GENERATES.equals(label)
+ || FROM_SOURCE.equals(label)) {
+ return NO_FIELDS;
+ }
+ throw new IllegalArgumentException(
+ "Unknown edge label: " + label);
+ }
+
+ private static String value(
+ Vertex vertex,
+ List fields,
+ String field) {
+ int index = fields.indexOf(field);
+ if (index < 0) {
+ throw new IllegalArgumentException("Unknown field: " + field);
+ }
+ return vertex.getValues().get(index);
+ }
+
+ private static TimeInterval interval(String start, String end) {
+ Instant parsedStart = parseInstant(start);
+ return end.isEmpty()
+ ? TimeInterval.unboundedFrom(parsedStart)
+ : new TimeInterval(parsedStart, parseInstant(end));
+ }
+
+ private static Instant parseInstant(String value) {
+ try {
+ return Instant.parse(value);
+ } catch (DateTimeParseException exception) {
+ throw new IllegalArgumentException(
+ "Invalid instant: " + value,
+ exception);
+ }
+ }
+
+ private static int positiveCount(String value) {
+ try {
+ int count = Integer.parseInt(value);
+ require(count > 0, "Occurrence count must be positive");
+ return count;
+ } catch (NumberFormatException exception) {
+ throw new IllegalArgumentException(
+ "Invalid occurrence count: " + value,
+ exception);
+ }
+ }
+
+ private static > T enumValue(
+ Class type,
+ String value,
+ String fieldName) {
+ try {
+ return Enum.valueOf(type, value);
+ } catch (IllegalArgumentException exception) {
+ throw new IllegalArgumentException(
+ "Invalid " + fieldName + ": " + value,
+ exception);
+ }
+ }
+
+ private static String relationLabel(VersionRelationType type) {
+ if (type == VersionRelationType.SUPERSEDES) {
+ return SUPERSEDES;
+ }
+ if (type == VersionRelationType.DUPLICATE_OF) {
+ return DUPLICATE_OF;
+ }
+ if (type == VersionRelationType.CONFLICTS_WITH) {
+ return CONFLICTS_WITH;
+ }
+ throw new IllegalArgumentException(
+ "Unsupported version relation type: " + type);
+ }
+
+ private static VersionRelationType relationType(String label) {
+ if (SUPERSEDES.equals(label)) {
+ return VersionRelationType.SUPERSEDES;
+ }
+ if (DUPLICATE_OF.equals(label)) {
+ return VersionRelationType.DUPLICATE_OF;
+ }
+ if (CONFLICTS_WITH.equals(label)) {
+ return VersionRelationType.CONFLICTS_WITH;
+ }
+ throw new IllegalArgumentException(
+ "Unsupported version relation label: " + label);
+ }
+
+ private static String relationId(
+ String label,
+ String sourceId,
+ String targetId) {
+ return "relation:" + tuple(label, sourceId, targetId);
+ }
+
+ private static String tuple(String... values) {
+ StringBuilder result = new StringBuilder();
+ for (String value : values) {
+ result.append(value.length()).append(':').append(value);
+ }
+ return result.toString();
+ }
+
+ private static String instant(Instant instant) {
+ return instant.toString();
+ }
+
+ private static String optionalInstant(TimeInterval interval) {
+ return interval.getEnd().isPresent()
+ ? instant(interval.getEnd().get()) : EMPTY;
+ }
+
+ private static String entityId(String rawId) {
+ return ENTITY_PREFIX + rawId;
+ }
+
+ private static String versionId(String rawId) {
+ return VERSION_PREFIX + rawId;
+ }
+
+ private static String eventId(String rawId) {
+ return EVENT_PREFIX + rawId;
+ }
+
+ private static String evidenceId(String rawId) {
+ return EVIDENCE_PREFIX + rawId;
+ }
+
+ private static String sourceId(String rawId) {
+ return SOURCE_PREFIX + rawId;
+ }
+
+ private static String rawId(String graphId, String prefix) {
+ require(graphId != null && graphId.startsWith(prefix),
+ "Graph id has the wrong prefix: " + graphId);
+ String rawId = graphId.substring(prefix.length());
+ require(!rawId.trim().isEmpty(),
+ "Graph id has an empty raw id: " + graphId);
+ return rawId;
+ }
+
+ private static List fields(String... values) {
+ return Collections.unmodifiableList(Arrays.asList(values));
+ }
+
+ private static void require(boolean condition, String message) {
+ if (!condition) {
+ throw new IllegalArgumentException(message);
+ }
+ }
+
+ private static final class CountedEdge {
+
+ private final String sourceId;
+ private final String targetId;
+ private int count;
+
+ private CountedEdge(String sourceId, String targetId) {
+ this.sourceId = sourceId;
+ this.targetId = targetId;
+ this.count = 1;
+ }
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/CandidateResult.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/CandidateResult.java
new file mode 100644
index 000000000..86f725837
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/CandidateResult.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.api;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.geaflow.ai.temporal.candidate.EvidenceCandidate;
+import org.apache.geaflow.ai.temporal.candidate.FactCandidate;
+import org.apache.geaflow.ai.temporal.candidate.RetrievalProfile;
+import org.apache.geaflow.ai.temporal.codec.TemporalJsonCodec;
+import org.apache.geaflow.ai.temporal.model.FactKey;
+import org.apache.geaflow.ai.temporal.semantics.NormalizedMemoryEvent;
+
+public final class CandidateResult {
+
+ private final QueryMode mode;
+ private final FactKey factKey;
+ private final String text;
+ private final Instant validAt;
+ private final Instant systemAt;
+ private final int topK;
+ private final List facts;
+ private final List evidence;
+ private final Map generatingEventIds;
+ private final RetrievalProfile profile;
+
+ CandidateResult(QueryMode mode, FactKey factKey, String text, Instant validAt,
+ Instant systemAt, int topK, List facts,
+ List evidence, Map generatingEventIds,
+ RetrievalProfile profile) {
+ this.mode = mode;
+ this.factKey = factKey;
+ this.text = text;
+ this.validAt = validAt;
+ this.systemAt = systemAt;
+ this.topK = topK;
+ this.facts = Collections.unmodifiableList(new ArrayList<>(facts));
+ this.evidence = Collections.unmodifiableList(new ArrayList<>(evidence));
+ this.generatingEventIds = Collections.unmodifiableMap(new TreeMap<>(generatingEventIds));
+ this.profile = profile;
+ }
+
+ public List getFacts() {
+ return facts;
+ }
+
+ public List getEvidence() {
+ return evidence;
+ }
+
+ public JsonObject toJson() {
+ TemporalJsonCodec codec = new TemporalJsonCodec();
+ JsonObject result = new JsonObject();
+ result.addProperty("schemaVersion", NormalizedMemoryEvent.SCHEMA_VERSION);
+ result.addProperty("mode", mode.name());
+ result.add("factKey", codec.factKeyToJson(factKey));
+ result.addProperty("validAt", validAt.toString());
+ result.addProperty("systemAt", systemAt == null ? null : systemAt.toString());
+ result.addProperty("text", text);
+ result.addProperty("topK", topK);
+ result.addProperty("retrievalProfile", profile.name());
+ JsonArray encodedFacts = new JsonArray();
+ for (FactCandidate candidate : facts) {
+ JsonObject value = new JsonObject();
+ value.addProperty("rank", candidate.getRank());
+ value.addProperty("mandatory", candidate.isMandatory());
+ JsonObject scores = new JsonObject();
+ for (Map.Entry score : candidate.getScores().entrySet()) {
+ scores.addProperty(score.getKey(), score.getValue());
+ }
+ value.add("scores", scores);
+ if (profile.isCandidateStrategy()) {
+ value.addProperty("score", candidate.getScore());
+ JsonObject ranks = new JsonObject();
+ for (Map.Entry rank : candidate.getChannelRanks().entrySet()) {
+ ranks.addProperty(rank.getKey(), rank.getValue());
+ }
+ value.add("channelRanks", ranks);
+ }
+ value.add("version", codec.versionToJson(candidate.getFactKey(),
+ candidate.getVersion(), generatingEventIds.get(candidate.getVersion().getId())));
+ encodedFacts.add(value);
+ }
+ result.add("facts", encodedFacts);
+ JsonArray encodedEvidence = new JsonArray();
+ for (EvidenceCandidate candidate : evidence) {
+ JsonObject value = new JsonObject();
+ value.addProperty("rank", candidate.getRank());
+ JsonObject scores = new JsonObject();
+ for (Map.Entry score : candidate.getChannelScores().entrySet()) {
+ scores.addProperty(score.getKey(), score.getValue());
+ }
+ value.add("scores", scores);
+ if (profile.usesLexicalFusion() || profile.isCandidateStrategy()) {
+ value.addProperty("score", candidate.getScore());
+ JsonObject ranks = new JsonObject();
+ for (Map.Entry rank : candidate.getChannelRanks().entrySet()) {
+ ranks.addProperty(rank.getKey(), rank.getValue());
+ }
+ value.add("channelRanks", ranks);
+ }
+ value.add("evidence", codec.evidenceToJson(candidate.getEvidence()));
+ JsonArray versionIds = new JsonArray();
+ for (String id : candidate.getVersionIds()) {
+ versionIds.add(new JsonPrimitive(id));
+ }
+ value.add("versionIds", versionIds);
+ encodedEvidence.add(value);
+ }
+ result.add("evidence", encodedEvidence);
+ return result;
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/QueryMode.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/QueryMode.java
new file mode 100644
index 000000000..45fa060c7
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/QueryMode.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.api;
+
+/** Public query modes for the experimental temporal API. */
+public enum QueryMode {
+ CURRENT,
+ HISTORICAL,
+ AS_OF
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/QueryResult.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/QueryResult.java
new file mode 100644
index 000000000..f51a2ee9a
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/QueryResult.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.api;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersion;
+import org.apache.geaflow.ai.temporal.trace.QueryTrace;
+
+/** Query matches and the explanation generated for this request. */
+public final class QueryResult {
+
+ private final List versions;
+ private final QueryTrace trace;
+
+ QueryResult(List versions, QueryTrace trace) {
+ this.versions = Collections.unmodifiableList(new ArrayList<>(versions));
+ this.trace = trace;
+ }
+
+ public List getVersions() {
+ return versions;
+ }
+
+ public QueryTrace getTrace() {
+ return trace;
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/ReplayResult.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/ReplayResult.java
new file mode 100644
index 000000000..dfbdca2a2
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/ReplayResult.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.api;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.geaflow.ai.temporal.semantics.CanonicalSnapshot;
+import org.apache.geaflow.ai.temporal.trace.DecisionTrace;
+
+/** A committed snapshot and the decisions for this submitted finite batch. */
+public final class ReplayResult {
+
+ private final CanonicalSnapshot snapshot;
+ private final List traces;
+ private final Map statistics;
+ private final String method;
+
+ ReplayResult(CanonicalSnapshot snapshot, List traces,
+ Map statistics, String method) {
+ this.snapshot = snapshot;
+ this.traces = Collections.unmodifiableList(new ArrayList<>(traces));
+ this.statistics = Collections.unmodifiableMap(new LinkedHashMap<>(statistics));
+ this.method = method;
+ }
+
+ public CanonicalSnapshot getSnapshot() {
+ return snapshot;
+ }
+
+ public List getDecisionTraces() {
+ return traces;
+ }
+
+ public Map getStatistics() {
+ return statistics;
+ }
+
+ public String getMethod() {
+ return method;
+ }
+
+ public String getFallbackStatus() {
+ return "NOT_APPLICABLE";
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/TemporalMemoryFacade.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/TemporalMemoryFacade.java
new file mode 100644
index 000000000..8d8579f2d
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/TemporalMemoryFacade.java
@@ -0,0 +1,331 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.api;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
+import org.apache.geaflow.ai.temporal.adapter.MemoryGraphAdapter;
+import org.apache.geaflow.ai.temporal.candidate.RetrievalProfile;
+import org.apache.geaflow.ai.temporal.candidate.TemporalCandidateService;
+import org.apache.geaflow.ai.temporal.codec.TemporalJsonCodec;
+import org.apache.geaflow.ai.temporal.integration.IncrementalTemporalIntegrator;
+import org.apache.geaflow.ai.temporal.model.FactKey;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersion;
+import org.apache.geaflow.ai.temporal.oracle.FullReplayOracle;
+import org.apache.geaflow.ai.temporal.query.TemporalQueryService;
+import org.apache.geaflow.ai.temporal.semantics.CanonicalSnapshot;
+import org.apache.geaflow.ai.temporal.semantics.EventLedgerDecision;
+import org.apache.geaflow.ai.temporal.semantics.NormalizedMemoryEvent;
+import org.apache.geaflow.ai.temporal.semantics.SnapshotComparator;
+import org.apache.geaflow.ai.temporal.semantics.TemporalErrorCode;
+import org.apache.geaflow.ai.temporal.semantics.TemporalValidationException;
+import org.apache.geaflow.ai.temporal.trace.DecisionTrace;
+import org.apache.geaflow.ai.temporal.trace.QueryTrace;
+import org.apache.geaflow.ai.temporal.trace.TraceBuilder;
+
+/**
+ * Single-session, finite-batch API. A staged graph and explanations publish together.
+ * Staging rebuilds Java state; this facade makes no incremental-performance claim.
+ */
+public final class TemporalMemoryFacade {
+
+ private static final Comparator EVENT_ORDER =
+ Comparator.comparing(NormalizedMemoryEvent::getRecordedAt)
+ .thenComparing(NormalizedMemoryEvent::getEventId);
+
+ private final TemporalJsonCodec codec = new TemporalJsonCodec();
+ private final List decisions = new ArrayList<>();
+ private final List queries = new ArrayList<>();
+ private final List operations = new ArrayList<>();
+ private CanonicalSnapshot snapshot = new IncrementalTemporalIntegrator().canonicalSnapshot();
+ private int revision;
+ private int attemptSequence;
+
+ public ReplayResult replay(List events) {
+ return commit(events, true, null, 20);
+ }
+
+ public ReplayResult applyBatch(List events) {
+ return commit(events, false, null, 20);
+ }
+
+ public ReplayResult replay(List events, RetrievalProfile profile, int topK) {
+ validateDecisionRetrieval(profile, topK);
+ return commit(events, true, profile, topK);
+ }
+
+ public ReplayResult applyBatch(List events, RetrievalProfile profile, int topK) {
+ validateDecisionRetrieval(profile, topK);
+ return commit(events, false, profile, topK);
+ }
+
+ private ReplayResult commit(List events, boolean reset,
+ RetrievalProfile profile, int topK) {
+ if (events == null) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "events must not be null");
+ }
+ List ordered = new ArrayList<>(events);
+ for (NormalizedMemoryEvent event : ordered) {
+ if (event == null) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "event must not be null");
+ }
+ }
+ ordered.sort(EVENT_ORDER);
+ JsonObject operation = eventOperation(reset ? "REPLAY" : "APPLY_BATCH", ordered);
+ if (profile != null) {
+ operation.addProperty("candidateProfile", profile.name());
+ operation.addProperty("candidateTopK", topK);
+ }
+ List attempts = new ArrayList<>();
+ try {
+ IncrementalTemporalIntegrator staged = new IncrementalTemporalIntegrator();
+ if (!reset) {
+ staged.applyBatch(snapshot.getEvents());
+ }
+ List codes = staged.applyBatch(ordered);
+ CanonicalSnapshot stagedSnapshot = reset && profile == null
+ ? new FullReplayOracle().replayToSnapshot(ordered)
+ : roundTrip(staged.canonicalSnapshot());
+ if (!new SnapshotComparator().compare(staged.canonicalSnapshot(), stagedSnapshot)
+ .isEquivalent()) {
+ throw new IllegalStateException("Java and formal Oracle snapshots differ");
+ }
+ if (profile != null && !new SnapshotComparator().compare(stagedSnapshot,
+ new FullReplayOracle().replayToSnapshot(stagedSnapshot.getEvents())).isEquivalent()) {
+ throw new IllegalStateException("Conflict resolver and independent Oracle snapshots differ");
+ }
+ boolean replaySemantics = profile == null || codes.contains(EventLedgerDecision.ACCEPTED);
+ Map semantics = profile == null
+ ? TraceBuilder.replay(stagedSnapshot.getEvents())
+ : replaySemantics
+ ? TraceBuilder.replay(stagedSnapshot.getEvents(), profile, topK)
+ : Collections.emptyMap();
+ int nextRevision = revision + 1;
+ int nextAttempt = attemptSequence;
+ Set acceptedIds = new TreeSet<>();
+ int accepted = 0;
+ for (int index = 0; index < ordered.size(); index++) {
+ NormalizedMemoryEvent event = ordered.get(index);
+ String code = codes.get(index).name();
+ boolean changed = codes.get(index) == EventLedgerDecision.ACCEPTED;
+ if (changed) {
+ accepted++;
+ acceptedIds.add(event.getEventId());
+ }
+ attempts.add(TraceBuilder.decision(event,
+ changed ? semantics.get(event.getEventId()) : null,
+ "attempt-" + (++nextAttempt), nextRevision, "INPUT", code, code, null, profile, topK));
+ }
+ List replayed = new ArrayList<>();
+ if (!reset && accepted > 0) {
+ for (NormalizedMemoryEvent event : stagedSnapshot.getEvents()) {
+ if (!acceptedIds.contains(event.getEventId())) {
+ replayed.add(TraceBuilder.decision(event, semantics.get(event.getEventId()),
+ "replay-" + nextRevision + "-" + event.getEventId(), nextRevision,
+ "REPLAY", "ACCEPTED", "ACCEPTED", null, profile, topK));
+ }
+ }
+ }
+ Map statistics = new LinkedHashMap<>();
+ statistics.put("submittedCount", ordered.size());
+ statistics.put("acceptedCount", accepted);
+ statistics.put("duplicateCount", ordered.size() - accepted);
+ statistics.put("stateEventCount", stagedSnapshot.getEvents().size());
+ statistics.put("traceReplayEventCount", replaySemantics ? stagedSnapshot.getEvents().size() : 0);
+ statistics.put("versionCount", stagedSnapshot.getState().getVersions().size());
+ statistics.put("relationCount", stagedSnapshot.getState().getRelations().size());
+ final ReplayResult result = new ReplayResult(stagedSnapshot, attempts, statistics,
+ profile != null ? "JAVA_CONFLICT_RESOLVER_MEMORY_GRAPH"
+ : reset ? "FULL_REPLAY_MEMORY_GRAPH" : "JAVA_LOCAL_STAGED_REPLAY");
+ snapshot = stagedSnapshot;
+ revision = nextRevision;
+ attemptSequence = nextAttempt;
+ decisions.addAll(attempts);
+ decisions.addAll(replayed);
+ operations.add(operation);
+ return result;
+ } catch (IllegalArgumentException error) {
+ TemporalValidationException failure = error instanceof TemporalValidationException
+ ? (TemporalValidationException) error
+ : new TemporalValidationException(TemporalErrorCode.REJECT_REFERENCE_INTEGRITY,
+ error.getMessage());
+ operation.addProperty("expectedError", failure.getCode().name());
+ for (NormalizedMemoryEvent event : ordered) {
+ decisions.add(TraceBuilder.decision(event, null,
+ "attempt-" + (++attemptSequence), revision, "INPUT", "BATCH_ROLLED_BACK",
+ failure.getCode().name(), failure.getMessage(), profile, topK));
+ }
+ operations.add(operation);
+ throw failure;
+ }
+ }
+
+ private static void validateDecisionRetrieval(RetrievalProfile profile, int topK) {
+ if (profile == null || !profile.isCandidateStrategy() || topK <= 0) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "Decision retrieval requires RULE_ONLY, RULE_BM25, RULE_MINHASH or RULE_ALL and positive topK");
+ }
+ }
+
+ public CanonicalSnapshot getSnapshot() {
+ return snapshot;
+ }
+
+ public CandidateResult candidates(QueryMode mode, FactKey factKey, String text,
+ Instant validAt, Instant systemAt, int topK)
+ throws IOException {
+ return candidates(mode, factKey, text, validAt, systemAt, topK, RetrievalProfile.BASELINE);
+ }
+
+ public CandidateResult candidates(QueryMode mode, FactKey factKey, String text,
+ Instant validAt, Instant systemAt, int topK,
+ RetrievalProfile profile) throws IOException {
+ if (mode == null || factKey == null || text == null || validAt == null || topK <= 0) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "mode, factKey, text and validAt are required; topK must be positive");
+ }
+ if ((mode == QueryMode.AS_OF) != (systemAt != null)) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "systemAt is required only for AS_OF");
+ }
+ if (profile == null) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "profile must not be null");
+ }
+ TemporalCandidateService service = new TemporalCandidateService(profile);
+ return new CandidateResult(mode, factKey, text, validAt, systemAt, topK,
+ service.facts(snapshot, mode, factKey, validAt, systemAt, text, topK),
+ service.evidence(snapshot, mode, factKey, validAt, systemAt, text, topK),
+ snapshot.getGeneratingEventIds(), profile);
+ }
+
+ public QueryResult query(QueryMode mode, FactKey factKey, Instant validAt, Instant systemAt) {
+ Objects.requireNonNull(mode, "mode");
+ Objects.requireNonNull(factKey, "factKey");
+ Objects.requireNonNull(validAt, "validAt");
+ if ((mode == QueryMode.AS_OF) != (systemAt != null)) {
+ throw new TemporalValidationException(TemporalErrorCode.REJECT_INPUT,
+ "systemAt is required only for AS_OF");
+ }
+ TemporalQueryService service = new TemporalQueryService();
+ List matches;
+ if (mode == QueryMode.CURRENT) {
+ matches = service.current(snapshot.getState(), factKey, validAt);
+ } else if (mode == QueryMode.HISTORICAL) {
+ matches = service.historical(snapshot.getState(), factKey, validAt);
+ } else {
+ matches = service.asOf(snapshot.getState(), factKey, validAt, systemAt);
+ }
+ final QueryTrace trace = TraceBuilder.query("query-" + (queries.size() + 1), revision,
+ mode, factKey, validAt, systemAt, matches, snapshot.getState());
+ JsonObject operation = new JsonObject();
+ operation.addProperty("method", "QUERY");
+ operation.addProperty("mode", mode.name());
+ operation.add("factKey", codec.factKeyToJson(factKey));
+ operation.addProperty("validAt", validAt.toString());
+ operation.addProperty("systemAt", systemAt == null ? null : systemAt.toString());
+ queries.add(trace);
+ operations.add(operation);
+ return new QueryResult(matches, trace);
+ }
+
+ public List getDecisionTraces() {
+ return Collections.unmodifiableList(new ArrayList<>(decisions));
+ }
+
+ public List getQueryTraces() {
+ return Collections.unmodifiableList(new ArrayList<>(queries));
+ }
+
+ public List getTrace(String eventId) {
+ Objects.requireNonNull(eventId, "eventId");
+ List matches = new ArrayList<>();
+ for (DecisionTrace trace : decisions) {
+ if (eventId.equals(trace.getEventId()) || eventId.equals(trace.getId())) {
+ matches.add(trace);
+ }
+ }
+ return Collections.unmodifiableList(matches);
+ }
+
+ public QueryTrace getQueryTrace(String queryId) {
+ for (QueryTrace trace : queries) {
+ if (trace.getId().equals(queryId)) {
+ return trace;
+ }
+ }
+ return null;
+ }
+
+ public JsonArray getOperations() {
+ JsonArray result = new JsonArray();
+ for (JsonObject operation : operations) {
+ result.add(codec.parseObject(codec.toJson(operation)));
+ }
+ return result;
+ }
+
+ public ValidationResult validateInvariants() {
+ List violations = new ArrayList<>();
+ try {
+ CanonicalSnapshot reloaded = roundTrip(snapshot);
+ if (!new SnapshotComparator().compare(snapshot, reloaded).isEquivalent()) {
+ violations.add("GRAPH_ROUND_TRIP_DIFF");
+ }
+ CanonicalSnapshot oracle = new FullReplayOracle().replayToSnapshot(snapshot.getEvents());
+ if (!new SnapshotComparator().compare(snapshot, oracle).isEquivalent()) {
+ violations.add("ORACLE_DIFF");
+ }
+ } catch (IllegalArgumentException error) {
+ violations.add("REFERENCE_INTEGRITY: " + error.getMessage());
+ }
+ return new ValidationResult(violations);
+ }
+
+ private JsonObject eventOperation(String method, List events) {
+ JsonObject operation = new JsonObject();
+ operation.addProperty("method", method);
+ JsonArray encoded = new JsonArray();
+ for (NormalizedMemoryEvent event : events) {
+ encoded.add(codec.eventToJson(event));
+ }
+ operation.add("events", encoded);
+ operation.addProperty("expectedError", (String) null);
+ return operation;
+ }
+
+ private static CanonicalSnapshot roundTrip(CanonicalSnapshot value) {
+ MemoryGraphAdapter adapter = new MemoryGraphAdapter();
+ return adapter.fromGraph(adapter.toGraph(value));
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/ValidationResult.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/ValidationResult.java
new file mode 100644
index 000000000..9c329c5b5
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/api/ValidationResult.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.api;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Invariant validation has no side effects on committed state. */
+public final class ValidationResult {
+
+ private final List violations;
+
+ ValidationResult(List violations) {
+ this.violations = Collections.unmodifiableList(new ArrayList<>(violations));
+ }
+
+ public boolean isValid() {
+ return violations.isEmpty();
+ }
+
+ public List getViolations() {
+ return violations;
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/baseline/LwwBaseline.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/baseline/LwwBaseline.java
new file mode 100644
index 000000000..cdcfd7ef3
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/baseline/LwwBaseline.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.baseline;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.TreeMap;
+import org.apache.geaflow.ai.temporal.model.FactKey;
+import org.apache.geaflow.ai.temporal.model.MemoryEventOperation;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersion;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersionStatus;
+import org.apache.geaflow.ai.temporal.model.TimeInterval;
+import org.apache.geaflow.ai.temporal.oracle.ReplayMethod;
+import org.apache.geaflow.ai.temporal.semantics.CanonicalSnapshot;
+import org.apache.geaflow.ai.temporal.semantics.EventLedger;
+import org.apache.geaflow.ai.temporal.semantics.EventLedgerDecision;
+import org.apache.geaflow.ai.temporal.semantics.NormalizedMemoryEvent;
+import org.apache.geaflow.ai.temporal.semantics.TemporalState;
+
+/**
+ * Replays events using deterministic last-write-wins current-state semantics.
+ */
+public final class LwwBaseline implements ReplayMethod {
+
+ private static final Comparator EVENT_ORDER =
+ Comparator.comparing(NormalizedMemoryEvent::getRecordedAt)
+ .thenComparing(NormalizedMemoryEvent::getEventId);
+
+ private static final TimeInterval ALL_VALID_TIME =
+ TimeInterval.unboundedFrom(
+ Instant.ofEpochMilli(Long.MIN_VALUE));
+
+ @Override
+ public CanonicalSnapshot replayToSnapshot(
+ List events) {
+ List ordered = new ArrayList<>(
+ Objects.requireNonNull(events, "events"));
+ for (NormalizedMemoryEvent event : ordered) {
+ Objects.requireNonNull(event, "event");
+ }
+ Collections.sort(ordered, EVENT_ORDER);
+
+ EventLedger ledger = new EventLedger();
+ List accepted = new ArrayList<>();
+ Map current = new TreeMap<>();
+ Map generatingEventIds = new TreeMap<>();
+ for (NormalizedMemoryEvent event : ordered) {
+ EventLedgerDecision decision = ledger.check(event);
+ if (decision == EventLedgerDecision.DUPLICATE_NOOP) {
+ continue;
+ }
+ if (decision == EventLedgerDecision.REJECT_EVENT_ID_REUSE) {
+ throw new IllegalArgumentException(
+ "Event id reused with a different payload: "
+ + event.getEventId());
+ }
+
+ apply(event, current, generatingEventIds);
+ ledger.commit(event);
+ accepted.add(event);
+ }
+
+ Map> versions =
+ new TreeMap<>();
+ for (Map.Entry entry
+ : current.entrySet()) {
+ versions.put(
+ entry.getKey(),
+ Collections.singletonList(entry.getValue()));
+ }
+ return new CanonicalSnapshot(
+ new TemporalState(versions, Collections.emptyList()),
+ accepted,
+ generatingEventIds);
+ }
+
+ private static void apply(
+ NormalizedMemoryEvent event,
+ Map current,
+ Map generatingEventIds) {
+ FactKey key = event.getFactKey();
+ MemoryEventOperation operation = event.getOperation();
+ if (operation == MemoryEventOperation.RETRACT) {
+ MemoryFactVersion removed = current.remove(key);
+ if (removed == null) {
+ throw new IllegalArgumentException(
+ "Cannot retract a fact without current state: "
+ + event.getFactId());
+ }
+ generatingEventIds.remove(removed.getId());
+ return;
+ }
+ if (operation == MemoryEventOperation.CORRECT
+ && !current.containsKey(key)) {
+ throw new IllegalArgumentException(
+ "Cannot correct a fact without current state: "
+ + event.getFactId());
+ }
+ if (operation != MemoryEventOperation.ADD
+ && operation != MemoryEventOperation.CORRECT) {
+ throw new UnsupportedOperationException(
+ "Unsupported memory event operation: " + operation);
+ }
+
+ MemoryFactVersion version = new MemoryFactVersion(
+ event.getEventId() + ":version:0",
+ event.getEvent().getFact().get(),
+ MemoryFactVersionStatus.ACTIVE,
+ ALL_VALID_TIME,
+ TimeInterval.unboundedFrom(event.getRecordedAt()),
+ event.getEvidence());
+ MemoryFactVersion replaced = current.put(key, version);
+ if (replaced != null) {
+ generatingEventIds.remove(replaced.getId());
+ }
+ generatingEventIds.put(version.getId(), event.getEventId());
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/baseline/SingleTimestampBaseline.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/baseline/SingleTimestampBaseline.java
new file mode 100644
index 000000000..c58d23efc
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/baseline/SingleTimestampBaseline.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.baseline;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.TreeMap;
+import org.apache.geaflow.ai.temporal.model.FactKey;
+import org.apache.geaflow.ai.temporal.model.FactValue;
+import org.apache.geaflow.ai.temporal.model.MemoryEventOperation;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersion;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersionStatus;
+import org.apache.geaflow.ai.temporal.model.TimeInterval;
+import org.apache.geaflow.ai.temporal.model.VersionRelation;
+import org.apache.geaflow.ai.temporal.model.VersionRelationType;
+import org.apache.geaflow.ai.temporal.oracle.ReplayMethod;
+import org.apache.geaflow.ai.temporal.semantics.CanonicalSnapshot;
+import org.apache.geaflow.ai.temporal.semantics.EventLedger;
+import org.apache.geaflow.ai.temporal.semantics.EventLedgerDecision;
+import org.apache.geaflow.ai.temporal.semantics.NormalizedMemoryEvent;
+import org.apache.geaflow.ai.temporal.semantics.TemporalState;
+
+/**
+ * Keeps the current distinct values while ignoring valid-time intervals.
+ */
+public final class SingleTimestampBaseline implements ReplayMethod {
+
+ private static final Comparator EVENT_ORDER =
+ Comparator.comparing(NormalizedMemoryEvent::getRecordedAt)
+ .thenComparing(NormalizedMemoryEvent::getEventId);
+
+ private static final TimeInterval ALL_VALID_TIME =
+ TimeInterval.unboundedFrom(Instant.ofEpochMilli(Long.MIN_VALUE));
+
+ @Override
+ public CanonicalSnapshot replayToSnapshot(
+ List events) {
+ List ordered = new ArrayList<>(
+ Objects.requireNonNull(events, "events"));
+ for (NormalizedMemoryEvent event : ordered) {
+ Objects.requireNonNull(event, "event");
+ }
+ Collections.sort(ordered, EVENT_ORDER);
+
+ EventLedger ledger = new EventLedger();
+ List accepted = new ArrayList<>();
+ Map> current =
+ new TreeMap<>();
+ for (NormalizedMemoryEvent event : ordered) {
+ EventLedgerDecision decision = ledger.check(event);
+ if (decision == EventLedgerDecision.DUPLICATE_NOOP) {
+ continue;
+ }
+ if (decision == EventLedgerDecision.REJECT_EVENT_ID_REUSE) {
+ throw new IllegalArgumentException(
+ "Event id reused with a different payload: "
+ + event.getEventId());
+ }
+
+ Map next = new TreeMap<>();
+ Map existing =
+ current.get(event.getFactKey());
+ if (existing != null) {
+ next.putAll(existing);
+ }
+ apply(event, next);
+ ledger.commit(event);
+ if (next.isEmpty()) {
+ current.remove(event.getFactKey());
+ } else {
+ current.put(event.getFactKey(), next);
+ }
+ accepted.add(event);
+ }
+
+ Map> versions =
+ new TreeMap<>();
+ Map generatingEventIds = new TreeMap<>();
+ List relations = new ArrayList<>();
+ for (Map.Entry> entry
+ : current.entrySet()) {
+ List versionsForKey = new ArrayList<>();
+ for (NormalizedMemoryEvent event : entry.getValue().values()) {
+ MemoryFactVersion version = version(event);
+ versionsForKey.add(version);
+ generatingEventIds.put(
+ version.getId(), event.getEventId());
+ }
+ versions.put(entry.getKey(), versionsForKey);
+ addConflicts(versionsForKey, relations);
+ }
+ return new CanonicalSnapshot(
+ new TemporalState(versions, relations),
+ accepted,
+ generatingEventIds);
+ }
+
+ private static void apply(
+ NormalizedMemoryEvent event,
+ Map current) {
+ MemoryEventOperation operation = event.getOperation();
+ if (operation == MemoryEventOperation.ADD) {
+ current.put(event.getFactValue().get(), event);
+ return;
+ }
+ if (current.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Operation requires current values for fact key: "
+ + event.getFactKey());
+ }
+ current.clear();
+ if (operation == MemoryEventOperation.CORRECT) {
+ current.put(event.getFactValue().get(), event);
+ } else if (operation != MemoryEventOperation.RETRACT) {
+ throw new UnsupportedOperationException(
+ "Unsupported memory event operation: " + operation);
+ }
+ }
+
+ private static MemoryFactVersion version(
+ NormalizedMemoryEvent event) {
+ return new MemoryFactVersion(
+ versionId(event),
+ event.getEvent().getFact().get(),
+ MemoryFactVersionStatus.ACTIVE,
+ ALL_VALID_TIME,
+ TimeInterval.unboundedFrom(event.getRecordedAt()),
+ event.getEvidence());
+ }
+
+ private static String versionId(NormalizedMemoryEvent event) {
+ return event.getEventId() + ":version:0";
+ }
+
+ private static void addConflicts(
+ List versions,
+ List relations) {
+ for (int left = 0; left < versions.size(); left++) {
+ for (int right = left + 1; right < versions.size(); right++) {
+ relations.add(new VersionRelation(
+ VersionRelationType.CONFLICTS_WITH,
+ versions.get(left).getId(),
+ versions.get(right).getId()));
+ }
+ }
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/benchmark/ClimateFeverDataset.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/benchmark/ClimateFeverDataset.java
new file mode 100644
index 000000000..1a52b474b
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/benchmark/ClimateFeverDataset.java
@@ -0,0 +1,390 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.benchmark;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import org.apache.geaflow.ai.temporal.codec.TemporalJsonCodec;
+import org.apache.geaflow.ai.temporal.model.Evidence;
+import org.apache.geaflow.ai.temporal.model.Source;
+
+public final class ClimateFeverDataset {
+
+ public static final int SUBSET_SIZE = 1000;
+ public static final String RELEASE_COMMIT = "336f0a46c98920a6f40926385ce71ba877a7b863";
+ public static final String RELEASE_SHA256 =
+ "8a4b9032d861be482ffb49dddfd283ffa6089e654f1e968040011882c5eb6e0b";
+ public static final String SOURCE_URL = "https://raw.githubusercontent.com/tdiggelm/"
+ + "climate-fever-dataset/" + RELEASE_COMMIT + "/dataset/climate-fever.jsonl";
+ private static final Gson GSON = new GsonBuilder().disableHtmlEscaping().create();
+
+ private final List claims;
+ private final List corpus;
+ private final String subsetJsonl;
+ private final String corpusJsonl;
+ private final String mappingsJsonl;
+ private final String goldJsonl;
+ private final Map> titlesByEvidenceId;
+ private final String titlesJsonl;
+ private final JsonObject manifest;
+
+ private ClimateFeverDataset(List claims, List corpus, String subsetJsonl,
+ String corpusJsonl, String mappingsJsonl, String goldJsonl,
+ Map> titlesByEvidenceId, String titlesJsonl,
+ JsonObject manifest) {
+ this.claims = Collections.unmodifiableList(claims);
+ this.corpus = Collections.unmodifiableList(corpus);
+ this.subsetJsonl = subsetJsonl;
+ this.corpusJsonl = corpusJsonl;
+ this.mappingsJsonl = mappingsJsonl;
+ this.goldJsonl = goldJsonl;
+ this.titlesByEvidenceId = Collections.unmodifiableMap(titlesByEvidenceId);
+ this.titlesJsonl = titlesJsonl;
+ this.manifest = manifest;
+ }
+
+ public static ClimateFeverDataset load(Path raw, String expectedSha256) throws IOException {
+ if (expectedSha256 == null || !expectedSha256.matches("[0-9a-fA-F]{64}")) {
+ throw new IllegalArgumentException("Expected SHA-256 must contain exactly 64 hex digits");
+ }
+ byte[] bytes = Files.readAllBytes(raw);
+ String actualSha256 = sha256(bytes);
+ if (!actualSha256.equals(expectedSha256.toLowerCase(Locale.ROOT))) {
+ throw new IllegalArgumentException("Raw dataset SHA-256 does not match: " + actualSha256);
+ }
+ TemporalJsonCodec codec = new TemporalJsonCodec();
+ Map records = new TreeMap<>();
+ Map rawEvidenceContents = new HashMap<>();
+ int lineNumber = 0;
+ for (String line : new String(bytes, StandardCharsets.UTF_8).split("\\r?\\n", -1)) {
+ lineNumber++;
+ if (line.trim().isEmpty()) {
+ continue;
+ }
+ JsonObject record;
+ try {
+ record = codec.parseObject(line);
+ validateRecord(record, rawEvidenceContents);
+ } catch (IllegalArgumentException exception) {
+ throw new IllegalArgumentException("Invalid dataset line " + lineNumber + ": "
+ + exception.getMessage(), exception);
+ }
+ String id = text(record, "claim_id");
+ if (!id.matches("0|[1-9][0-9]*")) {
+ throw new IllegalArgumentException("claim_id must be a canonical nonnegative integer: " + id);
+ }
+ if (records.put(new BigInteger(id), record) != null) {
+ throw new IllegalArgumentException("Duplicate claim_id: " + id);
+ }
+ }
+ if (records.isEmpty()) {
+ throw new IllegalArgumentException("Dataset must contain at least one claim");
+ }
+
+ List claims = new ArrayList<>();
+ Map corpus = new TreeMap<>();
+ Map> titleSets = new TreeMap<>();
+ StringBuilder subset = new StringBuilder();
+ StringBuilder mappings = new StringBuilder();
+ StringBuilder gold = new StringBuilder();
+ int noGoldCount = 0;
+ int neiCount = 0;
+ int evidencePairCount = 0;
+ long goldPairCount = 0;
+ for (JsonObject record : records.values()) {
+ if (claims.size() == SUBSET_SIZE) {
+ break;
+ }
+ String claimId = text(record, "claim_id");
+ String claimLabel = text(record, "claim_label");
+ Set goldIds = new TreeSet<>();
+ JsonObject selected = new JsonObject();
+ selected.addProperty("claim_id", claimId);
+ selected.addProperty("claim", text(record, "claim"));
+ selected.addProperty("claim_label", claimLabel);
+ JsonArray selectedEvidence = new JsonArray();
+ Map sortedEvidence = new TreeMap<>();
+ for (JsonElement item : record.getAsJsonArray("evidences")) {
+ JsonObject evidence = item.getAsJsonObject();
+ sortedEvidence.put(text(evidence, "evidence_id"), evidence);
+ }
+ for (JsonObject evidence : sortedEvidence.values()) {
+ String content = text(evidence, "evidence");
+ String evidenceId = "climate-evidence-" + sha256(content.getBytes(StandardCharsets.UTF_8));
+ String article = text(evidence, "article");
+ titleSets.computeIfAbsent(evidenceId, ignored -> new TreeSet<>()).add(article);
+ String sourceId = "wikipedia-" + sha256(article.getBytes(StandardCharsets.UTF_8));
+ Evidence candidate = new Evidence(evidenceId, new Source(sourceId, article), content);
+ Evidence existing = corpus.get(evidenceId);
+ if (existing != null && !existing.getContent().equals(content)) {
+ throw new IllegalArgumentException("Evidence SHA-256 collision: " + evidenceId);
+ }
+ // One text may occur in several articles; all original sources stay in mappings.
+ if (existing == null || article.compareTo(existing.getSource().getName()) < 0) {
+ corpus.put(evidenceId, candidate);
+ }
+ String label = text(evidence, "evidence_label");
+ if ("SUPPORTS".equals(label) || "REFUTES".equals(label)) {
+ goldIds.add(evidenceId);
+ }
+ JsonObject mapping = new JsonObject();
+ mapping.addProperty("claim_id", claimId);
+ mapping.addProperty("raw_evidence_id", text(evidence, "evidence_id"));
+ mapping.addProperty("evidence_id", evidenceId);
+ mapping.addProperty("evidence_label", label);
+ mapping.addProperty("source_id", sourceId);
+ mapping.addProperty("article", article);
+ append(mappings, mapping);
+ JsonObject entry = new JsonObject();
+ entry.addProperty("raw_evidence_id", text(evidence, "evidence_id"));
+ entry.addProperty("evidence_id", evidenceId);
+ entry.addProperty("evidence_label", label);
+ entry.addProperty("article", article);
+ selectedEvidence.add(entry);
+ evidencePairCount++;
+ }
+ selected.add("evidences", selectedEvidence);
+ append(subset, selected);
+ Claim claim = new Claim(claimId, text(record, "claim"), claimLabel, goldIds);
+ claims.add(claim);
+ JsonObject queryGold = new JsonObject();
+ queryGold.addProperty("claim_id", claimId);
+ queryGold.add("gold_evidence_ids", strings(goldIds));
+ append(gold, queryGold);
+ goldPairCount += goldIds.size();
+ if (goldIds.isEmpty()) {
+ noGoldCount++;
+ }
+ if ("NOT_ENOUGH_INFO".equals(claimLabel)) {
+ neiCount++;
+ }
+ }
+ StringBuilder corpusJsonl = new StringBuilder();
+ for (Evidence evidence : corpus.values()) {
+ JsonObject item = new JsonObject();
+ item.addProperty("evidence_id", evidence.getId());
+ item.addProperty("text", evidence.getContent());
+ item.addProperty("source_id", evidence.getSource().getId());
+ item.addProperty("article", evidence.getSource().getName());
+ append(corpusJsonl, item);
+ }
+ Map> titlesByEvidenceId = new TreeMap<>();
+ StringBuilder titlesJsonl = new StringBuilder();
+ for (Map.Entry> entry : titleSets.entrySet()) {
+ titlesByEvidenceId.put(entry.getKey(),
+ Collections.unmodifiableList(new ArrayList<>(entry.getValue())));
+ JsonObject item = new JsonObject();
+ item.addProperty("evidence_id", entry.getKey());
+ item.add("titles", strings(entry.getValue()));
+ append(titlesJsonl, item);
+ }
+ JsonObject manifest = new JsonObject();
+ manifest.addProperty("processorVersion", "climate-fever-prefix-v1");
+ manifest.addProperty("dataset", "CLIMATE-FEVER");
+ boolean official = RELEASE_SHA256.equals(actualSha256);
+ manifest.addProperty("release", official ? "1.0.1" : "custom-input");
+ manifest.addProperty("officialReleaseBytes", official);
+ if (official) {
+ manifest.addProperty("sourceCommit", RELEASE_COMMIT);
+ manifest.addProperty("sourceUrl", SOURCE_URL);
+ manifest.addProperty("sourcePage", "https://github.com/tdiggelm/climate-fever-dataset/releases/tag/1.0.1");
+ }
+ manifest.addProperty("rawSha256", actualSha256);
+ manifest.addProperty("rawBytes", bytes.length);
+ manifest.addProperty("license", "unknown");
+ manifest.addProperty("licenseSource", "https://huggingface.co/datasets/tdiggelm/climate_fever/blob/main/README.md");
+ manifest.addProperty("licenseNote", "Author dataset card declares unknown; no dataset license in source repository");
+ manifest.addProperty("attribution", "Diggelmann, Boyd-Graber, Bulian, Ciaramita, Leippold (2020), arXiv:2012.00614");
+ manifest.addProperty("selection", "Ascending numeric claim_id; first min(1000, raw claim count)");
+ manifest.addProperty("evaluationCorpus", "Deduplicated attached evidence of selected claims; not full Wikipedia");
+ manifest.addProperty("evidenceIdentity", "Exact original UTF-8 text SHA-256; no lexical normalization");
+ manifest.addProperty("sourceSelection", "Lexicographically smallest article per identical text; retain all source mappings");
+ manifest.addProperty("goldDefinition", "Distinct evidence text IDs with SUPPORTS or REFUTES; NOT_ENOUGH_INFO is not positive");
+ manifest.addProperty("zeroGoldPolicy", "Exclude zero-positive queries from recall denominators; report separately");
+ manifest.addProperty("rawClaimCount", records.size());
+ manifest.addProperty("selectedClaimCount", claims.size());
+ manifest.addProperty("selectedEvidencePairCount", evidencePairCount);
+ manifest.addProperty("corpusCount", corpus.size());
+ manifest.addProperty("positivePairCount", goldPairCount);
+ manifest.addProperty("zeroGoldClaimCount", noGoldCount);
+ manifest.addProperty("notEnoughInfoClaimCount", neiCount);
+ List claimIds = new ArrayList<>();
+ for (Claim claim : claims) {
+ claimIds.add(claim.getId());
+ }
+ manifest.add("selectedClaimIds", strings(claimIds));
+ manifest.addProperty("subsetSha256", sha256(subset.toString().getBytes(StandardCharsets.UTF_8)));
+ manifest.addProperty("corpusSha256", sha256(corpusJsonl.toString().getBytes(StandardCharsets.UTF_8)));
+ manifest.addProperty("mappingsSha256", sha256(mappings.toString().getBytes(StandardCharsets.UTF_8)));
+ manifest.addProperty("goldSha256", sha256(gold.toString().getBytes(StandardCharsets.UTF_8)));
+ manifest.addProperty("titleSelection", "All distinct original article titles per selected evidence text, regardless of label");
+ manifest.addProperty("titlesSha256", sha256(titlesJsonl.toString().getBytes(StandardCharsets.UTF_8)));
+ return new ClimateFeverDataset(claims, new ArrayList<>(corpus.values()), subset.toString(),
+ corpusJsonl.toString(), mappings.toString(), gold.toString(),
+ titlesByEvidenceId, titlesJsonl.toString(), manifest);
+ }
+
+ private static void validateRecord(JsonObject record, Map rawEvidenceContents) {
+ text(record, "claim_id");
+ text(record, "claim");
+ label(record, "claim_label", true);
+ if (!record.has("evidences") || !record.get("evidences").isJsonArray()) {
+ throw new IllegalArgumentException("evidences must be an array");
+ }
+ Set evidenceIds = new HashSet<>();
+ for (JsonElement item : record.getAsJsonArray("evidences")) {
+ if (!item.isJsonObject()) {
+ throw new IllegalArgumentException("Evidence must be an object");
+ }
+ JsonObject evidence = item.getAsJsonObject();
+ String id = text(evidence, "evidence_id");
+ if (!evidenceIds.add(id)) {
+ throw new IllegalArgumentException("Duplicate evidence_id in one claim: " + id);
+ }
+ String content = text(evidence, "evidence");
+ String article = text(evidence, "article");
+ label(evidence, "evidence_label", false);
+ String signature = GSON.toJson(Arrays.asList(article, content));
+ String existing = rawEvidenceContents.put(id, signature);
+ if (existing != null && !existing.equals(signature)) {
+ throw new IllegalArgumentException("Conflicting content/source for evidence_id: " + id);
+ }
+ }
+ }
+
+ private static void label(JsonObject object, String name, boolean claim) {
+ String value = text(object, name);
+ if (!Arrays.asList("SUPPORTS", "REFUTES", "NOT_ENOUGH_INFO").contains(value)
+ && !(claim && "DISPUTED".equals(value))) {
+ throw new IllegalArgumentException("Unknown " + name + ": " + value);
+ }
+ }
+
+ private static String text(JsonObject object, String name) {
+ JsonElement value = object.get(name);
+ if (value == null || !value.isJsonPrimitive() || !value.getAsJsonPrimitive().isString()
+ || value.getAsString().trim().isEmpty()) {
+ throw new IllegalArgumentException(name + " must be a nonempty string");
+ }
+ return value.getAsString();
+ }
+
+ static String sha256(byte[] bytes) {
+ try {
+ byte[] digest = MessageDigest.getInstance("SHA-256").digest(bytes);
+ char[] result = new char[digest.length * 2];
+ for (int index = 0; index < digest.length; index++) {
+ result[index * 2] = Character.forDigit((digest[index] & 0xff) >>> 4, 16);
+ result[index * 2 + 1] = Character.forDigit(digest[index] & 0x0f, 16);
+ }
+ return new String(result);
+ } catch (NoSuchAlgorithmException exception) {
+ throw new IllegalStateException("SHA-256 is unavailable", exception);
+ }
+ }
+
+ private static JsonArray strings(Iterable values) {
+ JsonArray array = new JsonArray();
+ for (String value : values) {
+ array.add(new com.google.gson.JsonPrimitive(value));
+ }
+ return array;
+ }
+
+ private static void append(StringBuilder builder, JsonObject object) {
+ builder.append(GSON.toJson(object)).append('\n');
+ }
+
+ public List getClaims() {
+ return claims;
+ }
+
+ public List getCorpus() {
+ return corpus;
+ }
+
+ public Map> getTitlesByEvidenceId() {
+ return titlesByEvidenceId;
+ }
+
+ public JsonObject getManifest() {
+ return new com.google.gson.JsonParser().parse(GSON.toJson(manifest)).getAsJsonObject();
+ }
+
+ public void writeArtifacts(Path output) throws IOException {
+ Files.write(output.resolve("subset.jsonl"), subsetJsonl.getBytes(StandardCharsets.UTF_8));
+ Files.write(output.resolve("corpus.jsonl"), corpusJsonl.getBytes(StandardCharsets.UTF_8));
+ Files.write(output.resolve("source-mappings.jsonl"), mappingsJsonl.getBytes(StandardCharsets.UTF_8));
+ Files.write(output.resolve("gold.jsonl"), goldJsonl.getBytes(StandardCharsets.UTF_8));
+ Files.write(output.resolve("titles.jsonl"), titlesJsonl.getBytes(StandardCharsets.UTF_8));
+ }
+
+ public static final class Claim {
+
+ private final String id;
+ private final String text;
+ private final String label;
+ private final Set goldIds;
+
+ private Claim(String id, String text, String label, Set goldIds) {
+ this.id = id;
+ this.text = text;
+ this.label = label;
+ this.goldIds = Collections.unmodifiableSet(goldIds);
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public String getText() {
+ return text;
+ }
+
+ public String getLabel() {
+ return label;
+ }
+
+ public Set getGoldIds() {
+ return goldIds;
+ }
+ }
+}
diff --git a/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/benchmark/FactGoldEvaluation.java b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/benchmark/FactGoldEvaluation.java
new file mode 100644
index 000000000..5d051e83c
--- /dev/null
+++ b/geaflow-ai/src/main/java/org/apache/geaflow/ai/temporal/benchmark/FactGoldEvaluation.java
@@ -0,0 +1,640 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.ai.temporal.benchmark;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonNull;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Stream;
+import org.apache.geaflow.ai.temporal.adapter.MemoryGraphAdapter;
+import org.apache.geaflow.ai.temporal.api.QueryMode;
+import org.apache.geaflow.ai.temporal.api.QueryResult;
+import org.apache.geaflow.ai.temporal.api.ReplayResult;
+import org.apache.geaflow.ai.temporal.api.TemporalMemoryFacade;
+import org.apache.geaflow.ai.temporal.candidate.RetrievalProfile;
+import org.apache.geaflow.ai.temporal.codec.TemporalJsonCodec;
+import org.apache.geaflow.ai.temporal.model.Evidence;
+import org.apache.geaflow.ai.temporal.model.FactKey;
+import org.apache.geaflow.ai.temporal.model.MemoryEntity;
+import org.apache.geaflow.ai.temporal.model.MemoryFact;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersion;
+import org.apache.geaflow.ai.temporal.model.MemoryFactVersionStatus;
+import org.apache.geaflow.ai.temporal.model.Source;
+import org.apache.geaflow.ai.temporal.model.TimeInterval;
+import org.apache.geaflow.ai.temporal.model.VersionRelation;
+import org.apache.geaflow.ai.temporal.model.VersionRelationType;
+import org.apache.geaflow.ai.temporal.oracle.FullReplayOracle;
+import org.apache.geaflow.ai.temporal.semantics.CanonicalSnapshot;
+import org.apache.geaflow.ai.temporal.semantics.NormalizedMemoryEvent;
+import org.apache.geaflow.ai.temporal.semantics.SnapshotComparator;
+import org.apache.geaflow.ai.temporal.semantics.TemporalState;
+import org.apache.geaflow.ai.temporal.semantics.TemporalValidationException;
+import org.apache.geaflow.ai.temporal.semantics.VersionIdentity;
+import org.apache.geaflow.ai.temporal.trace.DecisionTrace;
+import org.apache.geaflow.ai.temporal.trace.QueryTrace;
+
+/** Runs hand-authored fact Gold through every production candidate strategy. */
+public final class FactGoldEvaluation {
+
+ private static final TemporalJsonCodec CODEC = new TemporalJsonCodec();
+ private static final int TOP_K = 1;
+ private static final List FIXTURES = Arrays.asList(
+ "01-assert-entity-reference", "02-identical-duplicate", "03-event-id-reuse",
+ "04-evidence-merge", "05-conflict-coexistence", "06-adjacent-intervals",
+ "07-partial-correction", "08-partial-retraction", "09-valid-time-late",
+ "10-system-time-disorder", "11-same-timestamp", "12-invalid-target-atomicity");
+ private static final List PROFILES = Arrays.asList(
+ RetrievalProfile.RULE_ONLY, RetrievalProfile.RULE_BM25,
+ RetrievalProfile.RULE_MINHASH, RetrievalProfile.RULE_ALL);
+ private static final Comparator EVENT_ORDER =
+ Comparator.comparing(NormalizedMemoryEvent::getRecordedAt)
+ .thenComparing(NormalizedMemoryEvent::getEventId);
+ private static final Comparator END_ORDER =
+ Comparator.nullsLast(Comparator.naturalOrder());
+ private static final Comparator