from __future__ import annotations import json from functools import lru_cache from pathlib import Path from typing import Any from slipcore import ( LangGraphSlipstreamAdapter, create_base_ucr, parse_slip, render_human, validate_wire, ) ROOT = Path(__file__).resolve().parents[1] VALID_VECTORS = ROOT / "spec" / "conformance" / "valid.jsonl" INVALID_VECTORS = ROOT / "spec" / "conformance" / "invalid.jsonl" SPACE_RAW_ASSET_BASE = "https://huggingface.co/spaces/anthonym21/slipcore/raw/main/assets" SPACE_URL = "https://anthonym21-slipcore.hf.space/" WEBSITE_URL = "https://slipstream.making-minds.ai" FAVICON_URL = f"{SPACE_RAW_ASSET_BASE}/slipstream-mark.svg" SOCIAL_CARD_URL = f"{SPACE_RAW_ASSET_BASE}/slipstream-social-card.svg" FALLBACK_VALID_WIRES = [ "SLIP v3 planner reviewer Request Review auth", "SLIP v3 ops dev Inform Status green", "SLIP v3 worker router Meta Ack", "SLIP v3 dev sre Fallback Generic ref7f3a1b2c", ] FALLBACK_INVALID_WIRES = [ "BAD v3 planner reviewer Request Review auth", "SLIP v3 planner reviewer Unknown Review auth", "SLIP v3 dev sre Fallback Generic", "SLIP v3 planner reviewer Request Review auth-module", ] @lru_cache(maxsize=1) def _base_ucr(): return create_base_ucr() @lru_cache(maxsize=1) def _adapter() -> LangGraphSlipstreamAdapter: return LangGraphSlipstreamAdapter() def build_ucr_rows(force_filter: str = "All", search: str = "") -> list[dict[str, str]]: query = search.strip().lower() rows: list[dict[str, str]] = [] for anchor in sorted(_base_ucr(), key=lambda item: item.index): if force_filter != "All" and anchor.force != force_filter: continue haystack = " ".join( [ anchor.force, anchor.obj, anchor.canonical, " ".join(str(value) for value in anchor.coords), ] ).lower() if query and query not in haystack: continue rows.append( { "index": f"0x{anchor.index:04x}", "force": anchor.force, "object": anchor.obj, "canonical": anchor.canonical, "coords": str(anchor.coords), "core": "yes" if anchor.is_core else "no", "state": anchor.state.value, } ) return rows def analyze_wire(wire: str) -> dict[str, Any]: wire = wire.strip() issues = validate_wire(wire) if issues: return { "status": "invalid", "issues": issues, "human": "", "fields": {}, } message = parse_slip(wire) fields = { "version": message.version, "src": message.src, "dst": message.dst, "force": message.force, "object": message.obj, "payload": " ".join(message.payload), "fallback_ref": message.fallback_ref or "", "token_count": str(message.token_count_estimate), } return { "status": "valid", "issues": [], "human": render_human(message), "fields": fields, } def load_example_wires(kind: str) -> list[str]: path = VALID_VECTORS if kind == "Valid" else INVALID_VECTORS fallback = FALLBACK_VALID_WIRES if kind == "Valid" else FALLBACK_INVALID_WIRES examples: list[str] = [] if not path.exists(): return fallback with path.open(encoding="utf-8") as handle: for line in handle: record = json.loads(line) examples.append(record["wire"]) if len(examples) == 8: break return examples or fallback LANGGRAPH_SNIPPETS = { "Boundary Encode/Decode": """from typing import TypedDict from langgraph.graph import StateGraph from slipcore import ( LangGraphSlipstreamAdapter, make_decode_node, make_encode_node, ) class AgentState(TypedDict, total=False): thought: str src: str dst: str slip_wire: str slip_force: str slip_obj: str adapter = LangGraphSlipstreamAdapter() encode_node = make_encode_node(adapter) decode_node = make_decode_node(adapter) builder = StateGraph(AgentState) builder.add_node("slip_encode", encode_node) builder.add_node("slip_decode", decode_node) """, "Force:Object Router": """from slipcore import ( LangGraphSlipstreamAdapter, make_force_object_router, ) adapter = LangGraphSlipstreamAdapter() route = make_force_object_router(adapter) builder.add_conditional_edges( "slip_decode", route, { "Request:Review": "review_agent", "Inform:Status": "status_agent", "Fallback:Generic": "fallback_agent", }, ) """, "Fallback-Aware Flow": """from slipcore import LangGraphSlipstreamAdapter adapter = LangGraphSlipstreamAdapter() wire, result = adapter.encode_thought( "Check kubernetes pod logs for OOMKilled events", src="devops", dst="sre", ) decoded = adapter.decode_wire(wire) print(decoded.message.force, decoded.message.obj) print(decoded.fallback_text) """, } def get_langgraph_snippet(topic: str) -> str: return LANGGRAPH_SNIPPETS[topic] TRAINING_GUIDANCE = { "When should I train?": ( "Training is optional. Start with the built-in keyword quantizer, " "strict wire validation, and pointer-based fallback. Measure " "fallback rate and routing correctness first. Train only if your " "workload needs higher intent recall than the rules-based path provides." ), "What does the dataset look like?": ( "The dataset is ShareGPT-style conversation data for Think -> " "Quantize -> Transmit. Typical records include THOUGHT, QUANTIZE, " "and SLIP lines so a model can learn the protocol without hiding " "the reasoning step." ), "What model artifacts exist?": ( "Reference artifacts live on Hugging Face under the anthonym21 " "namespace: the dataset `slipstream-tqt`, the LoRA adapter " "`slipstream-glm-z1-9b`, and companion merged and GGUF variants." ), "How should I evaluate first?": ( "Build a small gold eval set from your own agent traffic. Track " "Force accuracy, Object accuracy, fallback rate, and downstream " "routing correctness before considering any fine-tuning pass." ), } def get_training_guidance(topic: str) -> str: return TRAINING_GUIDANCE[topic] def get_overview_metrics() -> list[dict[str, str]]: return [ {"metric": "Current release", "value": "3.1.1"}, {"metric": "Core dependencies", "value": "0"}, {"metric": "Passing tests", "value": "594"}, {"metric": "Average token reduction", "value": "82%"}, ] def get_head_html() -> str: description = ( "Inspect Slipstream 3.1.1, validate SLIP v3 wires, browse UCR anchors, " "and copy LangGraph starter patterns without a live model dependency." ) return f""" """ def get_overview_markdown() -> str: return f""" ## Start here This Space is for engineers evaluating Slipstream in a real agent stack. 1. **Inspect the protocol surface** in `UCR Explorer`. 2. **Validate concrete messages** in `Conformance Lab`. 3. **Drop the adapter into LangGraph** from `LangGraph Starter`. 4. **Use the dataset/model tab last**, after you have measured fallback rate and routing quality. You do not need to train a model to adopt Slipstream. Start with the runtime, route on `Force:Object`, and train only if your workload needs better quantization than the built-in path provides. For the release narrative, benchmarks, and positioning, use the website: {WEBSITE_URL} """ def get_resource_rows() -> list[dict[str, str]]: return [ {"resource": "Website", "link": "https://slipstream.making-minds.ai"}, {"resource": "GitHub", "link": "https://github.com/anthony-maio/slipcore"}, {"resource": "PyPI", "link": "https://pypi.org/project/slipcore/"}, {"resource": "Paper", "link": "https://doi.org/10.5281/zenodo.18063451"}, { "resource": "Dataset", "link": "https://huggingface.co/datasets/anthonym21/slipstream-tqt", }, { "resource": "Reference model", "link": "https://huggingface.co/anthonym21/slipstream-glm-z1-9b", }, ]