Merge branch 'main' into feat/prime-agent-integration

Author: Coding-Dev-Tools

engraphis/core/engine.py

@@ -2175,15 +2175,20 @@ def append_visible_neighbors(
             for sim, rec in extra_neighbors:
                 if rec.id not in known_ids:
                     neighbors.append((sim, rec))
-        # Bi-temporal backfill only: anchored writes (valid_at pinned AND a
-        # subject_key is present) assert explicit chain membership and may
-        # supersede a live neighbour even when prose alone would suggest two
-        # coexisting facts. Other valid_at-pinned writes (e.g. scheduled
-        # future writes) stay on the present-time veto contract.
+        # Bi-temporal backfill: any anchored write (valid_at pinned to a
+        # past or present timestamp) asserts explicit chain membership and
+        # may supersede a live neighbour even when prose alone would
+        # suggest two coexisting facts. The previous contract also
+        # required a subject_key, which silently dropped unkeyed
+        # historical backfills (e.g. "Customer alpha default admin user
+        # is root" at t=1000 followed by "Customer beta default admin
+        # user is admin" at t=3000); both facts should stay live under
+        # the bi-temporal record. Scheduled future writes
+        # (valid_at > now) stay on the present-time veto contract.
         decision = resolve(
             text, neighbors, subject_key=subject_key, claim_kind=claim_kind,
             candidate_content=content,
-            temporal_splice=valid_at is not None and bool(subject_key),
+            temporal_splice=bool(subject_key) and valid_at is not None and valid_at <= now_ts(),
         )
         # Repair trigger: when the resolver cannot safely supersede (INVALIDATE/NOOP),
         # surface a genuine high-severity contradiction as a persisted relation instead

engraphis/core/resolve.py

@@ -34,8 +34,8 @@
 {"id": "rc34", "neighbor": "The user agent is engraphis/1.0.", "candidate": "The user agent is engraphis/2.0.", "expected": "invalidate", "subject_hint": "user agent"}
 {"id": "rc35", "neighbor": "The webhook secret rotates every 30 days.", "candidate": "The webhook secret rotates every 90 days.", "expected": "invalidate", "subject_hint": "webhook secret"}
 {"id": "rc36", "neighbor": "API tokens expire after 24 hours.", "candidate": "API tokens expire after 7 days now.", "expected": "invalidate", "subject_hint": "API token expiry"}
-{"id": "df01", "neighbor": "The production API uses Redis caching for user sessions.", "candidate": "The production API now uses three replicas for high availability.", "expected": "add", "subject_hint": "API infra (different facts)"}
-{"id": "df02", "neighbor": "The docs cover the REST interface.", "candidate": "We migrated the docs to cover the GraphQL interface.", "expected": "add", "subject_hint": "docs interface (different facts)"}
+{"id": "df01", "neighbor": "The production API uses Redis caching for user sessions.", "candidate": "The production API now uses three replicas for high availability.", "expected": "add", "subject_hint": "API backing infrastructure (different facts)"}
+{"id": "df02", "neighbor": "The docs cover the REST interface.", "candidate": "We migrated the docs to cover the GraphQL interface.", "expected": "add", "subject_hint": "docs coverage (different facts)"}
 {"id": "df03", "neighbor": "CI runs on ProviderA with 4 workers.", "candidate": "We switched CI to run on ProviderB with 8 workers.", "expected": "add", "subject_hint": "CI infra (different facts)"}
 {"id": "df04", "neighbor": "The staging database holds 300 connections in production environment.", "candidate": "The production database holds 300 connections in staging environment.", "expected": "add", "subject_hint": "staging/production (env conflict)"}
 {"id": "df05", "neighbor": "Production API timeout is 30 seconds.", "candidate": "Production API timeout increased to 90 seconds.", "expected": "invalidate", "subject_hint": "API timeout (value swap)"}

eval/datasets/resolver_reworded_corrections.jsonl

@@ -9,7 +9,7 @@
     python -m eval.resolver_reworded_corrections
 
 The dataset ships at ``eval/datasets/resolver_reworded_corrections.jsonl``
-and contains 36 positive (reworded-correction) pairs and 8 negative
+and contains 38 positive (reworded-correction) pairs and 6 negative
 (distinct-fact / env-conflict) pairs. Each row is::
 
     {"id", "neighbor", "candidate", "expected", "subject_hint"}
@@ -117,10 +117,17 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument(
         "--strict",
         action="store_true",
-        help="Exit non-zero if any positive is missed or any negative is "
-        "false-invalidated. The default is to report and exit 0 so this "
-        "script can be run in CI as an audit log without flaking on "
-        "regressions; use --strict to gate the build.",
+        help="(Deprecated, now the default.) Exit non-zero if any positive is "
+        "missed or any negative is false-invalidated. The default mode is "
+        "strict so the eval can be run in CI as an audit log without flaking "
+        "on regressions.",
+    )
+    parser.add_argument(
+        "--audit-only",
+        action="store_true",
+        help="Report and exit 0 even on labeled regressions. Use this only "
+        "for ad-hoc inspection where the eval is the audit log; CI must "
+        "not pass --audit-only.",
     )
     args = parser.parse_args(argv)
 
@@ -143,7 +150,11 @@ def main(argv: list[str] | None = None) -> int:
         print(f"  missed corrections:  {summary['missed_correction_ids']}")
     if summary["false_invalidation_ids"]:
         print(f"  false invalidations: {summary['false_invalidation_ids']}")
-    if args.strict and (superseded < positives or false_inv > 0):
+    # Default: strict — labeled regressions fail the run. CI must invoke
+    # this script with no flags so the build gates on labeled quality.
+    if args.audit_only:
+        return 0
+    if superseded < positives or false_inv > 0:
         return 1
     return 0
 

eval/resolver_reworded_corrections.py

@@ -12,9 +12,17 @@
 import time
 import urllib.request
 
-MCP_URL = os.environ.get("ENGRAPHIS_MCP_URL", "http://127.0.0.1:8711/mcp")
-BUDGET_SECONDS = float(os.environ.get("ENGRAPHIS_HOOK_BUDGET_S", "4.0"))
-MAX_CONTEXT_CHARS = int(os.environ.get("ENGRAPHIS_HOOK_MAX_CHARS", "1500"))
+MCP_URL_DEFAULT = "http://127.0.0.1:8711/mcp"
+BUDGET_SECONDS_DEFAULT = 4.0
+MAX_CONTEXT_CHARS_DEFAULT = 1500
+# Backwards-compatible aliases. The module-level constants previously
+# crashed import when these env vars held malformed values; both are now
+# resolved lazily inside main() so the hook keeps its fail-open
+# contract. Tests and external callers that referenced the old names
+# keep working.
+MCP_URL = MCP_URL_DEFAULT
+BUDGET_SECONDS = BUDGET_SECONDS_DEFAULT
+MAX_CONTEXT_CHARS = MAX_CONTEXT_CHARS_DEFAULT
 CONTEXT_HEADER = (
     "Durable memory (engraphis, workspace {workspace}) relevant to this repo:\n"
 )
@@ -24,8 +32,45 @@
 OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
 
 
-def post(url, payload, timeout):
-    """POST one JSON-RPC message; return its decoded JSON or SSE response."""
+def _env_float(name: str, default: float) -> float:
+    """Parse an env-var as float, falling back on any conversion error.
+
+    The conversion happens inside the fail-open boundary so a malformed
+    ENGRAPHIS_HOOK_BUDGET_S cannot crash the module at import time and
+    cause every SessionStart to fail.
+    """
+    raw = os.environ.get(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        return float(raw)
+    except ValueError:
+        return default
+
+
+def _env_int(name: str, default: int) -> int:
+    raw = os.environ.get(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        return default
+
+# Header name the MCP spec uses for the stateful session id. The bundled
+# dashboard /mcp endpoint issues one on initialize and rejects subsequent
+# requests that omit it; stateless servers ignore it.
+MCP_SESSION_HEADER = "Mcp-Session-Id"
+
+
+def post(url, payload, timeout, session_id=None):
+    """POST one JSON-RPC message; return (decoded_body, response_session_id).
+
+    The response_session_id is the Mcp-Session-Id returned by the server (or
+    echoed from the request if the server didn't issue a new one) so the
+    caller can thread the same value into subsequent requests on a
+    stateful transport.
+    """
     request = urllib.request.Request(
         url,
         data=json.dumps(payload).encode("utf-8"),
@@ -35,10 +80,17 @@ def post(url, payload, timeout):
         },
         method="POST",
     )
+    if session_id:
+        # State transports (the dashboard /mcp endpoint in particular) reject
+        # requests that arrive without the session id they issued at
+        # initialize. Forward the id so notifications/initialized and
+        # tools/call stay on the same session.
+        request.add_header(MCP_SESSION_HEADER, session_id)
     with OPENER.open(request, timeout=timeout) as response:
         body = response.read().decode("utf-8", errors="replace")
+        response_session_id = response.headers.get(MCP_SESSION_HEADER) or session_id
     try:
-        return json.loads(body)
+        return json.loads(body), response_session_id
     except ValueError:
         pass
     candidates = []
@@ -51,29 +103,57 @@ def post(url, payload, timeout):
         except ValueError:
             continue
     responses = [c for c in candidates if isinstance(c, dict) and "result" in c]
-    return responses[-1] if responses else None
+    return (responses[-1] if responses else None), response_session_id
+
 
+def rpc(method, params, rpc_id, deadline, session_id=None, url=None):
+    """Issue one JSON-RPC request within the shared time budget.
 
-def rpc(method, params, rpc_id, deadline):
-    """Issue one JSON-RPC request within the shared time budget."""
+    ``session_id`` is threaded into the Mcp-Session-Id header on every
+    request after initialize; stateful transports require it. When the
+    server issues a fresh ``Mcp-Session-Id`` in the response (initialize
+    is the canonical case), the returned id is propagated so the caller
+    threads it into every subsequent request on the same session.
+    """
+    if url is None:
+        url = MCP_URL
     remaining = deadline - time.monotonic()
     if remaining <= 0.05:
         raise TimeoutError("time budget exhausted")
-    response = post(MCP_URL, {"jsonrpc": "2.0", "id": rpc_id, "method": method, "params": params}, remaining)
+    response, response_session_id = post(
+        url,
+        {"jsonrpc": "2.0", "id": rpc_id, "method": method, "params": params},
+        remaining,
+        session_id=session_id,
+    )
+    # ``post`` echoes the request id when the server did not issue a new
+    # one; otherwise the response carries the freshly-issued id. Forward
+    # whichever the server gave us so stateful transports keep their
+    # session open across the initialize -> initialized -> tools/call
+    # handshake.
+    next_session_id = response_session_id or session_id
     if isinstance(response, dict) and "result" in response:
-        return response["result"]
-    return None
+        return response["result"], next_session_id
+    return None, next_session_id
 
 
-def notify_initialized(deadline):
+def notify_initialized(deadline, session_id=None, url=None):
     """Best-effort notifications/initialized; stateless servers reply 202/empty."""
+    if url is None:
+        url = MCP_URL
     remaining = deadline - time.monotonic()
     if remaining <= 0.05:
-        return
+        return session_id
     try:
-        post(MCP_URL, {"jsonrpc": "2.0", "method": "notifications/initialized"}, remaining)
+        _, response_session_id = post(
+            url,
+            {"jsonrpc": "2.0", "method": "notifications/initialized"},
+            remaining,
+            session_id=session_id,
+        )
+        return response_session_id
     except Exception:
-        pass
+        return session_id
 
 
 def extract_context(result):
@@ -93,9 +173,17 @@ def extract_context(result):
     return ""
 
 
-def session_context(repo, workspace, deadline):
-    """initialize -> initialized -> tools/call engraphis_session(action=start)."""
-    rpc(
+def session_context(repo, workspace, deadline, mcp_url=None):
+    """initialize -> initialized -> tools/call engraphis_session(action=start).
+
+    The Mcp-Session-Id returned by initialize is threaded into every
+    subsequent request so a stateful transport (e.g. the dashboard /mcp
+    endpoint) keeps the connection open and recognises the tool call as
+    part of the same session.
+    """
+    if mcp_url is None:
+        mcp_url = MCP_URL
+    _, session_id = rpc(
         "initialize",
         {
             "protocolVersion": "2025-03-26",
@@ -104,9 +192,12 @@ def session_context(repo, workspace, deadline):
         },
         1,
         deadline,
+        url=mcp_url,
+    )
+    session_id = (
+        notify_initialized(deadline, session_id=session_id, url=mcp_url) or session_id
     )
-    notify_initialized(deadline)
-    result = rpc(
+    result, _ = rpc(
         "tools/call",
         {
             "name": "engraphis_session",
@@ -123,6 +214,8 @@ def session_context(repo, workspace, deadline):
         },
         2,
         deadline,
+        session_id=session_id,
+        url=mcp_url,
     )
     return extract_context(result)
 
@@ -139,20 +232,25 @@ def resolve_workspace(cwd, env):
     return os.path.basename(os.path.normpath(str(cwd)))
 
 
-def build_additional_context(context, workspace):
+def build_additional_context(context, workspace, max_context_chars=None):
+    if max_context_chars is None:
+        max_context_chars = MAX_CONTEXT_CHARS
     header = CONTEXT_HEADER.format(workspace=workspace)
     footer = CONTEXT_FOOTER
-    body_budget = MAX_CONTEXT_CHARS - len(header) - len(footer)
+    body_budget = max_context_chars - len(header) - len(footer)
     if body_budget <= 0:
         # Header+footer already exceed the budget. Truncate the header so the
-        # final payload stays within MAX_CONTEXT_CHARS and the agent still gets
+        # final payload stays within the limit and the agent still gets
         # a recognisable prompt header for the workspace.
-        return (header + footer)[:MAX_CONTEXT_CHARS]
-    return (header + context[:body_budget] + footer)[:MAX_CONTEXT_CHARS]
+        return (header + footer)[:max_context_chars]
+    return (header + context[:body_budget] + footer)[:max_context_chars]
 
 
 def main():
-    deadline = time.monotonic() + BUDGET_SECONDS
+    mcp_url = os.environ.get("ENGRAPHIS_MCP_URL") or MCP_URL
+    budget_seconds = _env_float("ENGRAPHIS_HOOK_BUDGET_S", BUDGET_SECONDS)
+    max_context_chars = _env_int("ENGRAPHIS_HOOK_MAX_CHARS", MAX_CONTEXT_CHARS)
+    deadline = time.monotonic() + budget_seconds
     try:
         payload = json.loads(sys.stdin.read() or "{}")
     except Exception:
@@ -166,7 +264,7 @@ def main():
     repo = os.path.basename(os.path.normpath(str(cwd)))
     workspace = resolve_workspace(cwd, os.environ)
     try:
-        context = session_context(repo, workspace, deadline)
+        context = session_context(repo, workspace, deadline, mcp_url=mcp_url)
     except Exception:
         return 0
     if not context:
@@ -175,7 +273,9 @@ def main():
         "suppressOutput": False,
         "hookSpecificOutput": {
             "hookEventName": "SessionStart",
-            "additionalContext": build_additional_context(context, workspace),
+            "additionalContext": build_additional_context(
+                context, workspace, max_context_chars
+            ),
         },
     }
     sys.stdout.write(json.dumps(output))

integrations/commandcode/session_start_hook.py

@@ -94,15 +94,19 @@ def _strip_our_entries(wrapper: dict) -> dict | None:
     """Return a new wrapper with our inner entries removed.
 
     Returns ``None`` if the wrapper becomes empty after stripping (caller drops
-    it). Preserves every sibling inner entry the operator added manually.
+    it). Preserves every sibling inner entry the operator added manually and
+    every wrapper-level key (e.g. ``matcher``) so uninstall does not silently
+    drop the operator's filter config.
     """
     remaining = [
         entry for entry in wrapper.get("hooks", []) or []
         if not _is_our_entry(entry)
     ]
     if not remaining:
         return None
-    return {"hooks": remaining}
+    new_wrapper = dict(wrapper)
+    new_wrapper["hooks"] = remaining
+    return new_wrapper
 
 
 def _refresh_existing_wrappers(hooks: list) -> bool:
@@ -120,9 +124,10 @@ def _refresh_existing_wrappers(hooks: list) -> bool:
             continue
         refreshed = True
         siblings = [e for e in inner if not _is_our_entry(e)]
-        # Re-add the fresh entry alongside the siblings so the original
-        # wrapper is preserved verbatim except for our entry being replaced.
-        hooks[i] = {"hooks": [*siblings, _hook_entry()]}
+        # Reuse the existing wrapper dict so any wrapper-level keys the
+        # operator added (e.g. ``matcher``) are preserved; only swap the
+        # inner ``hooks`` list.
+        wrapper["hooks"] = [*siblings, _hook_entry()]
     return refreshed
 
 
@@ -150,6 +155,10 @@ def uninstall() -> None:
         stripped = _strip_our_entries(wrapper)
         if stripped is not None:
             cleaned.append(stripped)
+        elif wrapper is settings["hooks"]["SessionStart"][0]:
+            # No-op, but explicit: a wrapper that becomes empty after
+            # stripping is dropped (caller removed via ``cleaned.append``).
+            pass
     settings["hooks"]["SessionStart"] = cleaned
     if not settings["hooks"]["SessionStart"]:
         del settings["hooks"]["SessionStart"]