test(fq): pin catalog playlist and reconnect semantics

tests/test_fq.py

@@ -0,0 +1,169 @@
+import json
+
+import pytest
+
+from core import fq
+from core.manifest_parser import ManifestParser, ManifestRegistry
+from core.signal_registry import SignalRegistry
+
+
+def test_agents_in_discovers_broad_catalog_but_tunes_nothing_by_default():
+    raw = """
+# Studio agent
+
+This agent can render video. This agent can generate 3D models.
+
+## Capabilities
+- Search the web
+- Execute Python code
+"""
+    profile = fq.agents_in(raw, project="studio")
+    frequencies = {x["frequency"] for x in profile["catalog"]}
+
+    assert {"media.video.render", "model.3d.generate", "research.web.search", "code.execute"} <= frequencies
+    assert profile["tune"] == []
+    assert profile["playlist"] == {"available": 4, "tuned": 0}
+    assert profile["runtime_genome"]["closure_state"] == "seed"
+
+
+def test_tune_is_a_playlist_not_a_catalog_copy():
+    profile = fq.agents_in(
+        "This agent can render video. This agent can generate 3D models. This agent can search the web.",
+        project="demo",
+    )
+    tuned = fq.retune(profile, ["video", "3d"])
+    assert tuned["tune"] == ["media.video.render", "model.3d.generate"]
+    assert len(tuned["catalog"]) == 3
+    assert tuned["playlist"] == {"available": 3, "tuned": 2}
+
+
+def test_agents_out_mentions_only_resident_playlist_not_every_available_tool():
+    profile = fq.agents_in(
+        "This agent can render video. This agent can generate 3D models. This agent can search the web.",
+        project="demo",
+        tune=["video"],
+    )
+    out = fq.agents_out(profile, endpoint="http://mesh.local:7860")
+    patch = out["agents_md_patch"]
+
+    assert "AGENTS.fq" in patch
+    assert "http://mesh.local:7860" in patch
+    assert "media.video.render" in patch
+    assert "model.3d.generate" not in patch
+    assert "research.web.search" not in patch
+    assert "full catalog is available" in patch
+
+
+def test_declared_signalmesh_frequencies_are_preserved_and_can_be_tuned_exactly():
+    raw = """
+# Existing node
+- **Frequency:** `repo.review`
+- **Mode:** `EMIT / BROADCAST`
+- **Frequency:** `tests.run`
+- **Mode:** `LISTEN`
+"""
+    profile = fq.agents_in(raw, tune=["repo.review", "tests.run"])
+    assert profile["tune"] == ["repo.review", "tests.run"]
+    by = {x["frequency"]: x for x in profile["catalog"]}
+    assert by["repo.review"]["origin"] == "declared_frequency"
+    assert by["tests.run"]["mode"] == "LISTEN"
+
+
+def test_manifest_parser_retains_fq_sidecar_and_only_explicit_frequencies_start_tuned():
+    declared = ManifestParser(
+        "- **Frequency:** `repo.review`\n- **Mode:** `EMIT`\nThis agent can render video.",
+        "https://example.test/AGENTS.md",
+    ).parse()
+    assert declared["agents_fq"]["tune"] == ["repo.review"]
+    assert "media.video.render" in {x["frequency"] for x in declared["agents_fq"]["catalog"]}
+
+    generic = ManifestParser(
+        "This agent can render video. This agent can generate 3D models.",
+        "https://example.test/generic/AGENTS.md",
+    ).parse()
+    assert generic["frequencies"] == []
+    assert generic["agents_fq"]["tune"] == []
+    assert generic["agents_fq"]["playlist"]["available"] == 2
+
+
+def test_manifest_registry_registers_sidecar_without_broadcasting_discovered_catalog():
+    registry = ManifestRegistry()
+    signal_registry = SignalRegistry()
+    before = set(signal_registry.streams)
+    parsed = registry.ingest(
+        "This agent can render video. This agent can generate 3D models.",
+        "https://example.test/no-broadcast/AGENTS.md",
+    )
+    assert fq.fq_registry.get(parsed["agents_fq"]["id"])
+    assert set(signal_registry.streams) == before
+
+
+def test_fq_registry_broadcasts_only_tuned_stations():
+    signal_registry = SignalRegistry()
+    for name in ("media.video.render", "model.3d.generate", "research.web.search"):
+        signal_registry.streams.pop(name, None)
+
+    profile = fq.agents_in(
+        "This agent can render video. This agent can generate 3D models. This agent can search the web.",
+        project="broadcast-demo",
+        tune=["video"],
+    )
+    local = fq.FQRegistry()
+    local.ingest(profile, signal_registry=signal_registry)
+
+    assert signal_registry.streams.get("media.video.render")
+    assert not signal_registry.streams.get("model.3d.generate")
+    assert not signal_registry.streams.get("research.web.search")
+    signal_registry.streams.pop("media.video.render", None)
+
+
+def test_roundtrip_keeps_extension_protocol_and_kind_dispatch_open():
+    profile = fq.agents_in("This agent can render video.", project="roundtrip", tune=["video"])
+    encoded = fq.dump_fq(profile)
+    decoded = fq.parse_fq(encoded)
+    assert decoded["schema"] == fq.SCHEMA
+    assert decoded["kind"] == "agents"
+    assert decoded["tune"] == ["media.video.render"]
+
+    # .fq is broader than AGENTS.fq: a future kind can use the same envelope.
+    custom = {
+        "schema": fq.SCHEMA,
+        "kind": "browser-flow",
+        "id": "fq.browser.demo",
+        "catalog": [{"frequency": "browser.checkout", "label": "Checkout"}],
+        "tune": ["browser.checkout"],
+        "signals": [{"frequency": "browser.checkout", "content": {"flow_ref": "flow.checkout"}}],
+    }
+    assert fq.parse_fq(custom)["kind"] == "browser-flow"
+
+
+def test_fq_refuses_credentials_but_allows_requirement_names_and_refs():
+    profile = fq.agents_in("This agent can render video.", project="safe")
+    profile["runtime_genome"]["env_names"] = ["OPENAI_API_KEY", "DATABASE_URL"]
+    profile["runtime_genome"]["session_ref"] = "mport.session.owner"
+    fq.parse_fq(profile)
+
+    poisoned = json.loads(fq.dump_fq(profile))
+    poisoned["runtime_genome"]["session_ref"] = "Bearer abcdefghijklmnopqrstuvwxyz"
+    with pytest.raises(ValueError, match="credential-shaped"):
+        fq.parse_fq(poisoned)
+
+    poisoned = json.loads(fq.dump_fq(profile))
+    poisoned["api_key"] = "sk-abcdefghijklmnopqrstuvwxyz"
+    with pytest.raises(ValueError, match="credential field"):
+        fq.parse_fq(poisoned)
+
+
+def test_selector_ambiguity_fails_instead_of_guessing_attention():
+    profile = {
+        "schema": fq.SCHEMA,
+        "kind": "agents",
+        "catalog": [
+            {"frequency": "image.generate", "label": "Generate image"},
+            {"frequency": "image.edit", "label": "Edit image"},
+        ],
+        "tune": [],
+        "signals": [],
+    }
+    with pytest.raises(ValueError, match="ambiguous"):
+        fq.retune(profile, ["image"])