ldp-protocol 0.2.0

LDP — LLM Delegate Protocol: identity-aware communication for multi-agent LLM systems
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

"""Tests for LDP delegate base class."""

import pytest
from ldp_protocol.delegate import LdpDelegate
from ldp_protocol.types import (
    LdpCapability,
    LdpEnvelope,
    LdpMessageBody,
    PayloadMode,
    QualityMetrics,
    TrustDomain,
)


class EchoDelegate(LdpDelegate):
    async def handle_task(self, skill, input_data, task_id):
        return {"echo": input_data}, 0.99


def _make_delegate() -> EchoDelegate:
    return EchoDelegate(
        delegate_id="ldp:delegate:echo-test",
        name="Echo Test",
        model_family="test",
        model_version="1.0",
        capabilities=[LdpCapability(name="echo", quality=QualityMetrics(quality_score=0.99))],
    )


class TestDelegate:
    @pytest.mark.asyncio
    async def test_hello_returns_capability_manifest(self):
        delegate = _make_delegate()
        env = LdpEnvelope.create(
            "", "client", "delegate", body=LdpMessageBody.hello("client", [PayloadMode.TEXT])
        )
        resp = await delegate.handle_message(env)
        assert resp.body.type == "CAPABILITY_MANIFEST"

    @pytest.mark.asyncio
    async def test_session_propose_returns_accept(self):
        delegate = _make_delegate()
        env = LdpEnvelope.create(
            "s1",
            "client",
            "delegate",
            body=LdpMessageBody.session_propose(
                config={
                    "preferred_payload_modes": ["text"],
                    "ttl_secs": 3600,
                    "trust_domain": "default",
                }
            ),
        )
        resp = await delegate.handle_message(env)
        assert resp.body.type == "SESSION_ACCEPT"

    @pytest.mark.asyncio
    async def test_task_submit_returns_result(self):
        delegate = _make_delegate()
        env = LdpEnvelope.create(
            "s1",
            "client",
            "delegate",
            body=LdpMessageBody.task_submit("t1", "echo", {"hello": "world"}),
        )
        resp = await delegate.handle_message(env)
        assert resp.body.type == "TASK_RESULT"
        assert resp.body.output == {"echo": {"hello": "world"}}
        assert resp.body.provenance is not None
        assert resp.body.provenance.confidence == 0.99

    @pytest.mark.asyncio
    async def test_task_cancel_returns_update(self):
        delegate = _make_delegate()
        env = LdpEnvelope.create("s1", "client", "delegate", body=LdpMessageBody.task_cancel("t1"))
        resp = await delegate.handle_message(env)
        assert resp.body.type == "TASK_UPDATE"
        assert resp.body.message == "cancelled"

    @pytest.mark.asyncio
    async def test_session_close_returns_ack(self):
        delegate = _make_delegate()
        env = LdpEnvelope.create(
            "s1", "client", "delegate", body=LdpMessageBody.session_close("done")
        )
        resp = await delegate.handle_message(env)
        assert resp.body.type == "SESSION_CLOSE"
        assert resp.body.reason == "acknowledged"

    @pytest.mark.asyncio
    async def test_unknown_message_returns_error(self):
        delegate = _make_delegate()
        env = LdpEnvelope.create("s1", "client", "delegate", body=LdpMessageBody(type="UNKNOWN"))
        resp = await delegate.handle_message(env)
        assert resp.body.type == "TASK_FAILED"

    def test_delegate_identity_card(self):
        delegate = _make_delegate()
        assert delegate.identity.delegate_id == "ldp:delegate:echo-test"
        assert delegate.identity.model_family == "test"
        assert len(delegate.identity.capabilities) == 1

    @pytest.mark.asyncio
    async def test_task_exception_returns_typed_error(self):
        class FailingDelegate(LdpDelegate):
            async def handle_task(self, skill, input_data, task_id):
                raise ValueError("Intentional test failure")

        delegate = FailingDelegate(
            delegate_id="ldp:delegate:fail",
            name="Failing",
            model_family="test",
            model_version="1.0",
        )
        env = LdpEnvelope.create(
            "s1", "client", "delegate", body=LdpMessageBody.task_submit("t1", "any", {})
        )
        resp = await delegate.handle_message(env)
        assert resp.body.type == "TASK_FAILED"

    @pytest.mark.asyncio
    async def test_untrusted_domain_rejected(self):
        delegate = EchoDelegate(
            delegate_id="ldp:delegate:strict",
            name="Strict",
            model_family="test",
            model_version="1.0",
            trust_domain=TrustDomain(name="internal"),
        )
        env = LdpEnvelope.create(
            "s1",
            "client",
            "delegate",
            body=LdpMessageBody.session_propose(
                config={"trust_domain": "external", "preferred_payload_modes": ["text"]}
            ),
        )
        resp = await delegate.handle_message(env)
        assert resp.body.type == "SESSION_REJECT"