Langertha-Knarr
view release on metacpan or search on metacpan
.claude/agents/knarr-test-writer.md view on Meta::CPAN
---
name: knarr-test-writer
description: "Write Knarr tests with Test2::V0 and the Handler::Code fake handler — protocol round-trips, streaming end markers, routing/passthrough behavior. Tests never require live API keys. Use for test additions, regression scaffolding, and co...
model: sonnet
allowed-tools: Read, Edit, Write, Bash, Glob, Grep
briefing:
skills:
- perl-core
- perl-ai-langertha
- perl-io-async-future
- karr
---
.claude/agents/knarr-test-writer.md view on Meta::CPAN
intent is unclear or the briefed behavior seems wrong, stop and ask.
Hard rule: **no test may require a live API key or external service.** Even the `*_live.t`
tests only start a real Knarr server on a local port and talk to it over loopback with
`Net::Async::HTTP` — backed by `Langertha::Knarr::Handler::Code` fakes, never a real
engine.
Mechanics of this suite:
- Test2::V0, flat `t/` directory, numbered `NN_topic.t` (00 load → 10s config/protocol →
20s handlers/routing → 30 protocols → 40s streaming → 50+ translator/psgi/passthrough/
tracing/cli/auth).
- `Langertha::Knarr::Handler::Code->new(code => sub {…}, stream_code => sub {…})` is the
canonical fake: `code` returns the response (string or `Langertha::Knarr::Response`),
`stream_code` returns an iterator sub that yields chunks then `undef`.
- Streaming assertions must check the exact per-protocol end marker (`data: [DONE]`,
`event: message_stop`, `{"done": true}`) — a tolerant assertion here hides hung-client
bugs.
- Follow the existing tests' IO::Async pattern: one `IO::Async::Loop`, server on an
ephemeral local port, await the response Futures.
.claude/agents/knarr-worker.md view on Meta::CPAN
- **Raw passthrough pipes bytes 1:1.** Unconfigured models forward all HTTP bytes
(headers, body, SSE chunks) untouched to the upstream API — that is what preserves
tool_use, usage, cache_control and protocol metadata. Never parse-and-reserialize on the
passthrough path; protocol translation lives exclusively in the Protocol::* modules on
the routed path.
- **`Langertha::Knarr::Response` is the single shape** every handler returns and every
protocol formatter consumes; `Response->coerce()` is the only upgrade path for legacy
shapes. Do not surface a second, parallel response representation.
- **Streaming end markers are per-protocol and exact** (`data: [DONE]` / `event:
message_stop` / `{"done": true}` — table in CLAUDE.md). A wrong or missing marker means
a hung client; any streaming change must run the streaming tests for all three formats.
## Verification
`prove -l t/` (flat directory, no subdirs) or `dzil test`. Tests are Test2::V0;
`Langertha::Knarr::Handler::Code` (code + stream_code) is the canonical fake handler. The
`*_live.t` tests start a real Knarr server on a local port inside the test — they need no
API keys and must stay key-free.
Never run `dzil release` — release is the maintainer's call (see house rules).
.claude/rules/knarr-rules.md view on Meta::CPAN
| Task | Agent |
|---|---|
| Implement / refactor / debug behavior-relevant code | `knarr-worker` (default) |
| Write/extend tests | `knarr-test-writer` |
| Pre-release audit | `knarr-release-checker` |
- **You cannot spawn subagents** (you ARE a `knarr-*` agent): The delegation lock does not
apply to you — implement, refactor, debug, and test per these rules.
Behavior-relevant = runtime behavior, public API, request/response handling, the raw
passthrough path, protocol formatting and streaming, routing, tracing, config parsing,
error handling, tests, performance. Pure prose docs and `Changes` notes are not.
## Coordination — karr board (always in scope)
Ticket coordination is the orchestrating agent's job, so `karr` is always in scope — don't
invoke the `karr` skill first, just use it. Git-native kanban; board state lives in
`refs/karr/*` in this repo (own board; the sibling Langertha repos each have their own —
cross-repo work is a ticket on that repo's board, never a direct edit). Day-to-day:
- `karr list --compact` / `karr board` — open work · `karr show ID` — detail
.claude/rules/knarr-rules.md view on Meta::CPAN
and `gh release`. For anything heading toward release: stop and ask.
## Knarr-specific hazards
- **The passthrough seam is bytes, not messages.** Unconfigured models pipe all HTTP bytes
1:1 to the upstream API — that is the feature (preserves tool_use, usage,
cache_control, SSE framing). A "cleanup" that parses and re-serializes on that path
looks harmless and passes the fake-backed tests, but silently strips real-provider
metadata. Protocol translation belongs only in `Protocol::*` on the routed path.
- **Streaming end markers** differ per protocol (table in CLAUDE.md); a wrong marker means
a hung client, not a test failure. Any streaming change runs
`prove -lv t/40_streaming_live.t t/41_streaming_protocols_live.t`.
- **`*_live.t` needs no API keys** — those tests start a real server on a local port with
`Handler::Code` fakes. Keep them key-free; nothing in `t/` may hit an external service.
## Perl specifics — reference, don't restate
Module loading, Moose vs. Moo house patterns, cpanfile versioning, POD directives, and
commit style live in the briefed skills (`perl-core`, `perl-moose`, `perl-moo`,
`perl-io-async-future`, `perl-release-author-getty`, `git-commit-style`). The Moose/Moo
module split itself is in CLAUDE.md. Do not duplicate that content here.
.claude/skills/perl-ai-langertha/SKILL.md view on Meta::CPAN
<capabilities>
## Capability Queries
Every engine reports its capabilities via `Langertha::Role::Capabilities`
(composed by `Role::Chat`, so present on every engine):
```perl
$engine->supports('tool_choice_named') or die "engine cannot force named tool";
$engine->supports('response_format_json_schema') # safe to pass json_schema response_format
$engine->supports('streaming') # chat_stream_request wired up
$engine->supports('tools_native') # accepts a tools array on the wire
$engine->supports('tools_hermes') # Hermes XML-tag tool path
my $caps = $engine->engine_capabilities;
# { chat=>1, streaming=>1, tools_native=>1, tool_choice_named=>1,
# response_format_json_schema=>1, embedding=>1, transcription=>1, ... }
```
The flag set is derived from which capability roles the engine composes
(central role→flag map in `Role::Capabilities`); engines override via
`around engine_capabilities` for wire-reality corrections.
</capabilities>
<chat-f>
## chat_f — Single-Turn with Named Args
.claude/skills/perl-ai-langertha/SKILL.md view on Meta::CPAN
Engines compose feature roles:
| Role | Feature |
|------|---------|
| `Langertha::Role::Capabilities` | `engine_capabilities` registry + `supports($cap)` |
| `Langertha::Role::Chat` | `simple_chat`, `simple_chat_f`, `chat_f` (named args), `aggregate_tool_calls` |
| `Langertha::Role::Tools` | `chat_with_tools_f` (MCP loop) |
| `Langertha::Role::HermesTools` | XML-tag tool calling for models without native support |
| `Langertha::Role::ParallelToolUse` | `parallel_tool_use` boolean (canonical name) |
| `Langertha::Role::Streaming` | SSE/NDJSON streaming |
| `Langertha::Role::Embedding` | Vector embeddings |
| `Langertha::Role::Transcription` | Audio-to-text |
| `Langertha::Role::ImageGeneration` | Image generation |
| `Langertha::Role::SystemPrompt` | System prompt management |
| `Langertha::Role::Temperature` | Sampling temperature |
| `Langertha::Role::Seed` | Deterministic seed (`seed`, `randomize_seed`) |
| `Langertha::Role::ContextSize` | `context_size` parameter |
| `Langertha::Role::ResponseSize` | `response_size` / max_tokens parameter |
| `Langertha::Role::ResponseFormat` | JSON mode / structured output, plus `$self->decode_loose_json($text)` (overridable) |
| `Langertha::Role::Models` | Model listing |
.claude/skills/perl-ai-langertha/SKILL.md view on Meta::CPAN
<value-objects>
## Value Objects
| Class | Purpose |
|-------|---------|
| `Langertha::Tool` | Canonical tool definition. `from_openai/from_anthropic/from_mcp/from_gemini/from_hash` accept any shape; `to_openai/to_anthropic/to_gemini/to_mcp/to_json_schema` emit per-provider wire payloads. |
| `Langertha::ToolChoice` | Canonical tool-selection policy (`auto`/`any`/`none`/`tool`). `to_openai/to_anthropic/to_gemini/to_perplexity` per-provider serializers. |
| `Langertha::ToolCall` | Tool invocation emitted by an LLM. `name`, `arguments`, `id`, `synthetic`. `from_openai/from_anthropic/from_ollama/from_gemini`; `extract($raw)` pulls every call out of any known response shape. |
| `Langertha::Content::Image` | Provider-agnostic vision input. `from_url/from_file/from_data`; `to_openai/to_anthropic/to_gemini`. |
| `Langertha::Response` | LLM response with metadata. Stringifies to `content`. `tool_calls` is `ArrayRef[Langertha::ToolCall]` — single source of truth. |
| `Langertha::Stream::Chunk` | Single streaming chunk. Optional `tool_calls` for engines that emit them mid-stream; `Role::Chat::aggregate_tool_calls(\@chunks)` flattens. |
Use these instead of hand-rolled hashes when normalizing across
providers. `Tool->from_hash` auto-detects MCP camelCase, Anthropic
snake_case, OpenAI envelope, and Gemini-flat shapes.
</value-objects>
usage passes through as the same instance. Note the 1.100 entry
above describes a ToolCall/Usage symmetry that never existed on
the usage side — Langertha upgrades tool_calls to value objects
in BUILDARGS but leaves usage a hashref.
- Fixed: with Langfuse configured, a routed response carrying usage
then died a second time, in the Langfuse batch encode. Langertha's
value objects have no TO_JSON, and Tracing::flush encoded without
an eval, so a serialization failure inside end_trace turned an
already-answered, already-paid-for upstream call into a 500 — and
on the streaming path, where the trace closes after the last
chunk, into a client left waiting for an end marker. The JSONL
request log hit the same object but encodes inside an eval that
only warns, so it silently dropped whole log lines instead.
Usage is now flattened at the JSON boundary in both Tracing and
RequestLog, mirroring the _rate_limit_hash that already sat next
to it. Neither defect was reachable before, because the
constructor died first.
- Tracing::flush no longer throws. A failed encode is logged at
error level and the batch dropped, exactly as an ingestion HTTP
to detect. New engine_catalog class method exposes the list.
- Langfuse generations carry real latency. Langertha::Response grew
timing with ttft_seconds / total_seconds, but Knarr dropped it
along with id, thinking and rate_limit when normalizing. The
generation's endTime is now startTime + total_seconds and
completionStartTime (the Langfuse field for time-to-first-token)
is startTime + ttft_seconds, both anchored to the high-resolution
instant start_trace recorded rather than to a wall clock that has
moved on; the raw timing hash is kept in metadata so provider-native
stage durations survive. Only the routed non-streaming path has
engine-measured timing — streaming and raw passthrough keep the
proxy's own wall clock, and a new POD section says which path
measures what. rate_limit is flattened to its quota scalars, never
its raw headers.
- response_format is capability-gated like every other generation
parameter, and gated on the right one: Langertha registers
response_format_json_object and response_format_json_schema
separately, so an explicit json_schema demands the schema flag
while OpenAI's json_object and Ollama's bare 'json' need only the
object flag. Gating both on the schema flag would have dropped a
- Tracing flush is async via Net::Async::HTTP. The previous
LWP::UserAgent call blocked the IO::Async event loop on every
end_trace; the new flush fires the request and returns
immediately, with a warn-on-fail logger attached.
- Streaming pump consolidated into Knarr::Stream::from_callback.
Removes ~50 lines of duplicated queue/pending/finished/error
bookkeeping from Handler::Engine and Handler::Router.
- supports()-aware streaming detection. The old
$engine->can('simple_chat_stream_realtime_f') heuristic now
defers to $engine->supports('streaming') when available
(Langertha 0.500+) and falls back to the can()-check for older
engines.
- Dead steerboard attribute removed from all six Protocol classes
(never read), and Knarr.pm no longer threads $self into
protocol-object construction. Knarr::PSGI's constructor argument
renamed steerboard => knarr.
- 'steerboard' string fallbacks replaced with 'unknown' (model
default) and 'knarr-code' / 'knarr-raider' (handler defaults).
* Passthrough with and without tracing enabled
- Documentation updated: CLAUDE.md, README.md, bin/knarr POD
reflect new architecture, --from-env, repeatable --port,
KNARR_DEBUG, raw passthrough behavior.
1.000 2026-04-10 00:00:20Z
- MAJOR REWRITE. Mojolicious is gone; the new core is built on
IO::Async + Net::Async::HTTP::Server + Future::AsyncAwait for
native async streaming and seamless integration with Langertha
engines. Existing knarr.yaml configs and the knarr CLI commands
(start, container, models, check, init) keep working.
- Six wire protocols loaded by default on every listening port:
OpenAI (/v1/chat/completions, /v1/models, SSE), Anthropic
(/v1/messages, named-event SSE), Ollama (/api/chat, /api/tags,
NDJSON streaming), A2A (Google Agent2Agent JSON-RPC at / with
/.well-known/agent.json discovery), ACP (BeeAI/IBM /agents,
/runs), and AG-UI (CopilotKit /awp event stream).
- Pluggable Handler architecture replaces the old Knarr::Proxy
classes:
Knarr::Handler::Router — model→engine via Knarr::Router
Knarr::Handler::Engine — single Langertha engine wrapper
Knarr::Handler::Raider — per-session Langertha::Raider
Knarr::Handler::Code — coderef-backed for tests/fakes
Knarr::Handler::Passthrough — raw HTTP forward to upstream
reconfiguration.
- proxy_api_key authentication enforced by the new dispatcher.
Clients must present 'Authorization: Bearer <token>' or
'x-api-key: <token>' when KNARR_API_KEY (or yaml proxy_api_key)
is set. The /.well-known/agent.json discovery route stays
anonymous.
- PSGI adapter (Knarr::PSGI) for deploying behind any Plack
server (Starman, Twiggy, mod_perl, ...). Streaming is buffered
in this mode; use the native server for real-time streaming.
- Universal protocol translator pattern: an OpenAI-fronted Knarr
with a Handler::A2AClient backend exposes a remote A2A agent to
OpenAI clients (and similarly for ACP).
- Drops the Knarr namespace facades for Langertha utility code
(Knarr::Metrics, Knarr::Input, Knarr::Output, Knarr::Input::Tools,
Knarr::Output::Tools). Use Langertha::Usage / Tool / ToolCall /
ToolChoice / Pricing / Cost / UsageRecord directly from
Langertha core.
HTTP::Message, File::ShareDir::ProjectDistDir; adds IO::Async,
Net::Async::HTTP, Net::Async::HTTP::Server, Future::AsyncAwait,
Moose, Path::Tiny (test), Capture::Tiny (test).
- Bump Langertha floor to 0.400 for the new value object API.
- dist.ini sets irc = #langertha (on irc.perl.org).
- Test suite expanded from 200 to 320 unit tests + author POD
syntax. New coverage:
* end-to-end live streaming for Anthropic / A2A / ACP / AG-UI
(OpenAI and Ollama already covered)
* Handler::Passthrough sync + streaming through a real backend
* Handler::Router with passthrough fallback
* Multi-listen on real sockets
* Tracing and RequestLog decorators
* CLI smoketest for knarr models / check / init
* proxy_api_key auth enforcement
0.007 2026-03-10 20:05:00Z
- Resolve configured engines through both `Langertha::Engine::*` and
`LangerthaX::Engine::*` (including fully-qualified class names).
t/00-load.t
t/10-config.t
t/10_openai_protocol.t
t/15_response.t
t/20-router.t
t/20_handlers.t
t/25_chat_f_params.t
t/26_tool_calls_routing.t
t/27_usage_routing.t
t/30_all_protocols.t
t/40_streaming_live.t
t/41_streaming_protocols_live.t
t/50_translator.t
t/60_psgi.t
t/70_passthrough.t
t/71_router_passthrough_fallback.t
t/72_multi_listen.t
t/73_tracing_handler.t
t/74_requestlog_handler.t
t/75_cli_smoketest.t
t/76_auth.t
t/77_raw_passthrough_tracing.t
lib/Langertha/Knarr.pm view on Meta::CPAN
Ollama SDKs, and the agent ecosystems around A2A, ACP, and AG-UI.
By default a single running Knarr answers OpenAI
C</v1/chat/completions>, Anthropic C</v1/messages>, Ollama
C</api/chat>, A2A's C</.well-known/agent.json> plus JSON-RPC C</>,
ACP's C</runs>, and AG-UI's C</awp> simultaneously on every listening
port. The same handler implementation drives all of them.
Knarr is built on L<IO::Async> and L<Net::Async::HTTP::Server>
with native L<Future::AsyncAwait> integration into Langertha engines,
so streaming works end-to-end token-by-token without any thread or
event-loop bridges.
=head1 ARCHITECTURE
Three pluggable layers:
=over
=item B<Protocols>
lib/Langertha/Knarr.pm view on Meta::CPAN
L<Langertha::Knarr::Handler::Code> (coderef-backed for tests). Implement
L<Langertha::Knarr::Handler> to write your own. Decorators
(L<Langertha::Knarr::Handler::Tracing>,
L<Langertha::Knarr::Handler::RequestLog>) wrap any inner handler and
add behavior on top — they themselves consume the Handler role and
compose freely.
=item B<Transport>
Default is L<Net::Async::HTTP::Server> with chunked SSE / NDJSON
streaming on one or more listen sockets. For Plack deployments,
L<Langertha::Knarr::PSGI> wraps the same Knarr instance into a PSGI
app (buffered — see its docs for the streaming caveat).
=back
=head2 handler
Required. An object consuming L<Langertha::Knarr::Handler>.
=head2 listen
ArrayRef of C<host:port> strings or C<< { host => ..., port => ... } >>
lib/Langertha/Knarr/Handler.pm view on Meta::CPAN
our $VERSION = '1.101';
use Moose::Role;
use Future::AsyncAwait;
use Langertha::Knarr::Stream;
use Langertha::Knarr::Response;
requires 'handle_chat_f';
requires 'list_models';
# Default streaming = run handle_chat_f and emit one chunk.
# Handlers that natively stream should override.
async sub handle_stream_f {
my ($self, $session, $request) = @_;
my $r = Langertha::Knarr::Response->coerce( await $self->handle_chat_f($session, $request) );
return Langertha::Knarr::Stream->from_list( $r->content );
}
1;
__END__
lib/Langertha/Knarr/Handler/Code.pm view on Meta::CPAN
with 'Langertha::Knarr::Handler';
has code => (
is => 'ro',
isa => 'CodeRef',
required => 1,
);
# Optional: separate generator for streaming. Returns a coderef that itself
# returns next-chunk strings (undef = done) when called.
has stream_code => (
is => 'ro',
isa => 'Maybe[CodeRef]',
default => sub { undef },
);
has models => (
is => 'ro',
isa => 'ArrayRef',
lib/Langertha/Knarr/Handler/Code.pm view on Meta::CPAN
},
stream_code => sub {
my @parts = ('hel', 'lo');
return sub { @parts ? shift @parts : undef };
},
);
=head1 DESCRIPTION
The simplest possible handler: pass coderefs that return strings (or
chunk generators for streaming) and you get a working Knarr handler.
Useful for tests, fakes, smoketests, and "fake LLM" demos.
=head2 code
Required. Coderef called as C<< $code->($session, $request) >> for
non-streaming requests; must return a scalar string.
=head2 stream_code
Optional. Coderef returning another coderef that yields the next chunk
per call, C<undef> to signal end.
=head2 models
Optional. Arrayref of model descriptors. Defaults to a single
C<knarr-code> entry.
lib/Langertha/Knarr/Handler/Engine.pm view on Meta::CPAN
async sub handle_chat_f {
my ($self, $session, $request) = @_;
my $response = await $self->engine->chat_f( $request->chat_f_args($self->engine) );
my $r = Langertha::Knarr::Response->coerce($response);
return $r->clone_with( model => $r->model // $self->_model_id );
}
async sub handle_stream_f {
my ($self, $session, $request) = @_;
my $engine = $self->engine;
unless ( _supports_streaming($engine) ) {
# Engine doesn't support native streaming — fall back to single-chunk.
my $r = await $self->handle_chat_f($session, $request);
return Langertha::Knarr::Stream->from_list( $r->content );
}
return Langertha::Knarr::Stream->from_callback( sub {
my ($emit, $done, $fail) = @_;
my $cb = sub {
my ($chunk) = @_;
my $text = ref $chunk && $chunk->can('content') ? $chunk->content : "$chunk";
$emit->($text);
};
my $f = $engine->simple_chat_stream_realtime_f( $cb, @{ $request->messages } );
$f->on_done( $done );
$f->on_fail( $fail );
$f->retain;
});
}
sub _supports_streaming {
my ($engine) = @_;
return $engine->supports('streaming') if $engine->can('supports');
return $engine->can('simple_chat_stream_realtime_f') && $engine->can('chat_stream_request');
}
sub list_models {
my ($self) = @_;
return [ { id => $self->_model_id, object => 'model' } ];
}
__PACKAGE__->meta->make_immutable;
1;
lib/Langertha/Knarr/Handler/Engine.pm view on Meta::CPAN
);
my $handler = Langertha::Knarr::Handler::Engine->new(
engine => $engine,
model_id => 'groq-llama-3.3-70b',
);
=head1 DESCRIPTION
Wraps a single L<Langertha::Engine::*> instance and exposes it as a
Knarr handler. Non-streaming requests are dispatched via
C<< $engine->chat_f >> with the full set of generation parameters
(C<tools>, C<tool_choice>, C<response_format>, C<temperature>,
C<max_tokens>) forwarded from the client request — subject to the
engine's reported capabilities. Streaming requests use
C<simple_chat_stream_realtime_f> for native token-by-token delivery;
engines that don't support streaming fall back to a single-chunk
emission.
For routing across multiple engines based on model name, use
L<Langertha::Knarr::Handler::Router> with a L<Langertha::Knarr::Router>
config instead.
=head2 engine
Required. Any object consuming L<Langertha::Role::Chat>. Streaming
support is detected via C<< $engine->supports('streaming') >> (Langertha
0.500+) or the presence of C<simple_chat_stream_realtime_f>.
=head2 model_id
Optional. The id reported by L</list_models> and surfaced in responses.
Defaults to the engine's C<chat_model>, falling back to a derived name
from the engine class.
=head1 SUPPORT
lib/Langertha/Knarr/Handler/Passthrough.pm view on Meta::CPAN
}
}
if ( my $auth = $self->default_auth ) {
$http_req->header( Authorization => $auth ) unless $http_req->header('Authorization');
}
$http_req->content( $self->_json->encode($body) );
return $http_req;
}
# Extract assistant text from an upstream response body for the protocol it
# came from. Bare-minimum extractor for sync mode; the streaming path
# forwards bytes verbatim and doesn't need this.
sub _extract_text {
my ($self, $protocol_name, $resp_body) = @_;
my $data = eval { $self->_json->decode($resp_body) };
return '' unless ref $data eq 'HASH';
if ( $protocol_name eq 'openai' ) {
return $data->{choices}[0]{message}{content} // '';
}
if ( $protocol_name eq 'anthropic' ) {
my $bits = '';
lib/Langertha/Knarr/Handler/Passthrough.pm view on Meta::CPAN
},
);
=head1 DESCRIPTION
Forwards the original wire-format request verbatim to a real upstream
API. The protocol's parser already turned the body into a
L<Langertha::Knarr::Request>; Passthrough rebuilds the upstream JSON
from C<$request-E<gt>raw> and re-POSTs it.
Both sync and streaming requests are supported. For streaming, the
upstream's protocol-native chunks are extracted into plain text deltas
which the front-side protocol then re-frames — keeping symmetry even
when client and upstream use the same protocol.
This is the building block behind Knarr's classic "configure your API
keys once, point everything at me" use case.
=head2 upstreams
Required. HashRef mapping protocol name (C<openai>, C<anthropic>,
lib/Langertha/Knarr/Handler/RequestLog.pm view on Meta::CPAN
my $rlog = Langertha::Knarr::RequestLog->new(config => $config);
$handler = Langertha::Knarr::Handler::RequestLog->new(
wrapped => $handler,
request_log => $rlog,
);
=head1 DESCRIPTION
Decorator handler that writes a structured per-request log entry for
every chat or stream request via L<Langertha::Knarr::RequestLog>.
Sync requests log a single line with the result; streaming requests
accumulate every delta and log one line with the assembled output
when the stream closes.
C<knarr start> mounts this automatically when
C<KNARR_LOG_FILE> / C<KNARR_LOG_DIR> (or the YAML C<logging:> section)
is set.
=head2 wrapped
Required. The inner L<Langertha::Knarr::Handler> being decorated.
lib/Langertha/Knarr/Handler/Router.pm view on Meta::CPAN
}
async sub handle_stream_f {
my ($self, $session, $request) = @_;
my ($engine) = $self->_resolve( $request->model );
unless ( $engine ) {
return await $self->passthrough->handle_stream_f( $session, $request );
}
unless ( _supports_streaming($engine) ) {
my $r = await $self->handle_chat_f($session, $request);
return Langertha::Knarr::Stream->from_list( $r->content );
}
return Langertha::Knarr::Stream->from_callback( sub {
my ($emit, $done, $fail) = @_;
my $cb = sub {
my ($chunk) = @_;
my $text = ref $chunk && $chunk->can('content') ? $chunk->content : "$chunk";
$emit->($text);
};
my $f = $engine->simple_chat_stream_realtime_f( $cb, @{ $request->messages } );
$f->on_done( $done );
$f->on_fail( $fail );
$f->retain;
});
}
sub _supports_streaming {
my ($engine) = @_;
return $engine->supports('streaming') if $engine->can('supports');
return $engine->can('simple_chat_stream_realtime_f') && $engine->can('chat_stream_request');
}
sub list_models {
my ($self) = @_;
my $models = $self->router->list_models;
return [ map { ref $_ eq 'HASH' ? $_ : { id => "$_", object => 'model' } } @{ $models || [] } ];
}
__PACKAGE__->meta->make_immutable;
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
package Langertha::Knarr::PSGI;
# ABSTRACT: PSGI adapter for Langertha::Knarr (buffered, no streaming)
our $VERSION = '1.101';
use Moose;
use JSON::MaybeXS;
use Langertha::Knarr::Request;
# Wraps a Langertha::Knarr instance and returns a PSGI app coderef.
# Streaming requests are coerced into buffered responses: the full body is
# assembled (open + chunks + close + done) before being returned to the
# PSGI server. Use the native Net::Async::HTTP::Server entrypoint
# (Langertha::Knarr->run) if you need real streaming.
has knarr => ( is => 'ro', required => 1 );
has _json => (
is => 'ro',
default => sub { JSON::MaybeXS->new( utf8 => 1, canonical => 1 ) },
);
sub to_app {
my ($self) = @_;
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
[ $self->_json->encode({ error => { message => "unknown action $action" } }) ] ];
}
my $body = $self->_read_body($env);
my $fake_http = Langertha::Knarr::PSGI::FakeReq->new( $env );
my $sb_req = $proto->parse_chat_request( $fake_http, \$body );
my $session = $sb->session( $sb_req->session_id );
my $handler = $sb->handler;
if ( $sb_req->stream ) {
# Buffered streaming: drive the stream to completion, concatenate frames.
my $stream = $handler->handle_stream_f( $session, $sb_req )->get;
my $out = $proto->format_stream_open($sb_req);
while ( defined( my $delta = $stream->next_chunk_f->get ) ) {
$out .= $proto->format_stream_chunk( $delta, $sb_req );
}
$out .= $proto->format_stream_close($sb_req);
$out .= $proto->format_stream_done($sb_req);
return [ 200, [ 'Content-Type' => $proto->stream_content_type ], [ $out ] ];
}
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
1;
__END__
=pod
=encoding UTF-8
=head1 NAME
Langertha::Knarr::PSGI - PSGI adapter for Langertha::Knarr (buffered, no streaming)
=head1 VERSION
version 1.101
=head1 SYNOPSIS
use Langertha::Knarr;
use Langertha::Knarr::PSGI;
lib/Langertha/Knarr/PSGI.pm view on Meta::CPAN
# Run with any PSGI server:
# plackup -s Starman -p 8088 app.psgi
=head1 DESCRIPTION
Adapter that wraps a L<Langertha::Knarr> instance and exposes it as a
PSGI app, so you can deploy Knarr behind any Plack server (Starman,
Twiggy, Gazelle, mod_perl, etc.) instead of running its native
L<Net::Async::HTTP::Server> loop.
B<Streaming responses are buffered.> The PSGI streaming protocol's
delayed-response form does work in theory but mixes badly with
L<IO::Async> in the same process; for honesty's sake this adapter
just drives the inner stream to completion in a blocking loop and
returns the full assembled body. Use the native
L<Langertha::Knarr/run> entry point if you need real-time streaming.
=head2 knarr
Required. The L<Langertha::Knarr> instance to expose.
=head2 to_app
Returns the PSGI coderef.
=head1 SUPPORT
lib/Langertha/Knarr/Protocol.pm view on Meta::CPAN
my ($self) = @_;
return "data: [DONE]\n\n";
}
# Optional lifecycle hooks for protocols that need to frame the stream
# (Anthropic message_start/stop, A2A status events, ACP run.created, AGUI RUN_STARTED).
# Default: empty — protocols like OpenAI / Ollama don't need them.
sub format_stream_open { '' }
sub format_stream_close { '' }
# Content-Type for streaming responses. Default is SSE; Ollama overrides.
sub stream_content_type { 'text/event-stream' }
# format_models_response(\@models) -> ($status, \%headers, $body)
sub format_models_response {
my ($self, $models) = @_;
return ( 200, { 'Content-Type' => 'application/json' }, '{"data":[]}' );
}
1;
lib/Langertha/Knarr/Protocol.pm view on Meta::CPAN
via the matched protocol's parser/formatter.
Knarr ships with six concrete protocols, all loaded by default:
=over
=item * L<Langertha::Knarr::Protocol::OpenAI> — C</v1/chat/completions>, SSE
=item * L<Langertha::Knarr::Protocol::Anthropic> — C</v1/messages>, named SSE events
=item * L<Langertha::Knarr::Protocol::Ollama> — C</api/chat>, NDJSON streaming
=item * L<Langertha::Knarr::Protocol::A2A> — Google Agent2Agent JSON-RPC
=item * L<Langertha::Knarr::Protocol::ACP> — IBM/BeeAI Agent Communication Protocol
=item * L<Langertha::Knarr::Protocol::AGUI> — CopilotKit AG-UI event protocol
=back
=head2 protocol_name
lib/Langertha/Knarr/Protocol.pm view on Meta::CPAN
Required. Returns a L<Langertha::Knarr::Request>.
=head2 format_chat_response
my ($status, \%headers, $body) = $proto->format_chat_response($response, $request);
Required. Returns the HTTP response triple for sync mode.
=head2 format_stream_open / format_stream_chunk / format_stream_close / format_stream_done
Lifecycle hooks for streaming responses. Defaults are no-ops where the
protocol doesn't need framing — Anthropic/A2A/ACP/AG-UI override these
to emit their named events around the chunk stream.
=head2 stream_content_type
Returns the HTTP C<Content-Type> for streaming responses. Default
C<text/event-stream>; Ollama overrides to C<application/x-ndjson>.
=head1 SUPPORT
=head2 Issues
Please report bugs and feature requests on GitHub at
L<https://github.com/Getty/langertha-knarr/issues>.
=head2 IRC
lib/Langertha/Knarr/Protocol/A2A.pm view on Meta::CPAN
with 'Langertha::Knarr::Protocol';
# --- A2A overview ---
# Discovery: GET /.well-known/agent.json -> AgentCard JSON
# Method bus: POST / body: JSON-RPC 2.0 envelope
# methods: "tasks/send", "tasks/sendSubscribe", "tasks/get",
# "tasks/cancel", "tasks/pushNotification/set", ...
#
# tasks/send (sync): returns Task JSON in JSON-RPC result
# tasks/sendSubscribe (streaming): returns SSE stream of JSON-RPC responses
# Each event is a full JSON-RPC response wrapping a TaskStatusUpdateEvent or
# TaskArtifactUpdateEvent. Final event has status.state = "completed".
#
# Streaming model on the wire:
# data: {"jsonrpc":"2.0","id":<reqid>,"result":{"id":"<task>","status":{"state":"working","message":{...}}, "final":false}}\n\n
# data: {"jsonrpc":"2.0","id":<reqid>,"result":{"id":"<task>","artifact":{"parts":[{"type":"text","text":"Hi"}], "index":0,"append":true}, "final":false}}\n\n
# data: {"jsonrpc":"2.0","id":<reqid>,"result":{"id":"<task>","status":{"state":"completed"},"final":true}}\n\n
# ---------------------------------------------------------------------------
has _json => ( is => 'ro', default => sub { JSON::MaybeXS->new( utf8 => 1, canonical => 1 ) } );
lib/Langertha/Knarr/Protocol/A2A.pm view on Meta::CPAN
);
sub _build_agent_card {
my ($self) = @_;
return {
name => 'Langertha Steerboard Agent',
description => 'Steerboard-exposed agent',
url => '/',
version => '0.0.1',
capabilities => {
streaming => JSON::MaybeXS::true(),
pushNotifications => JSON::MaybeXS::false(),
stateTransitionHistory => JSON::MaybeXS::false(),
},
defaultInputModes => ['text'],
defaultOutputModes => ['text'],
skills => [],
};
}
sub protocol_name { 'a2a' }
lib/Langertha/Knarr/Protocol/A2A.pm view on Meta::CPAN
=over
=item * C<GET /.well-known/agent.json> — agent card discovery (anonymous)
=item * C<POST /> — JSON-RPC 2.0 method bus
=back
Supported methods: C<tasks/send> (sync) and C<tasks/sendSubscribe>
(streaming). Streaming wraps every chunk in a JSON-RPC envelope with
the original C<id> preserved, transitions C<status.state> from
C<working> to C<completed>, and emits artifact append events for the
text deltas.
=head2 agent_card
The HashRef returned by the discovery endpoint. Defaults to a generic
"Knarr Agent" card with C<streaming: true>; override to advertise
specific skills, version, etc.
=head1 SUPPORT
=head2 Issues
Please report bugs and feature requests on GitHub at
L<https://github.com/Getty/langertha-knarr/issues>.
=head2 IRC
lib/Langertha/Knarr/Protocol/AGUI.pm view on Meta::CPAN
use Moose;
use JSON::MaybeXS;
use Data::UUID;
use Time::HiRes qw( time );
use Langertha::Knarr::Request;
use Langertha::Knarr::Response;
with 'Langertha::Knarr::Protocol';
# --- AG-UI overview ---
# CopilotKit-driven event protocol meant for agent <-> UI streaming.
# Transport-agnostic; SSE is the canonical wire form.
# Endpoint convention: POST /awp (Agent Wire Protocol) — body is the run input,
# response is an SSE stream of typed events.
#
# Event types (subset, all framed as SSE "data: { ... }" with "type" field):
# RUN_STARTED { type, threadId, runId }
# TEXT_MESSAGE_START { type, messageId, role:"assistant" }
# TEXT_MESSAGE_CONTENT { type, messageId, delta:"..." }
# TEXT_MESSAGE_END { type, messageId }
# TOOL_CALL_START { type, toolCallId, toolCallName, parentMessageId }
lib/Langertha/Knarr/Protocol/AGUI.pm view on Meta::CPAN
my $data = $self->_json->decode( $$body_ref || '{}' );
my @msgs;
for my $m ( @{ $data->{messages} || [] } ) {
push @msgs, { role => $m->{role} // 'user', content => $m->{content} // '' };
}
return Langertha::Knarr::Request->new(
protocol => 'agui',
raw => $data,
model => $data->{model},
messages => \@msgs,
stream => 1, # AG-UI is always streaming
session_id => $data->{threadId},
extra => {
thread_id => $data->{threadId} // $self->_uuid->create_str,
run_id => $data->{runId} // $self->_uuid->create_str,
message_id => $self->_uuid->create_str,
},
);
}
sub format_chat_response {
lib/Langertha/Knarr/Protocol/AGUI.pm view on Meta::CPAN
version 1.101
=head1 DESCRIPTION
Implements the CopilotKit AG-UI agent-to-UI event protocol on top of
L<Langertha::Knarr::Protocol>. Loaded by default.
=over
=item * C<POST /awp> — Agent Wire Protocol endpoint, always streaming
=back
Streaming emits the typed AG-UI event sequence:
C<RUN_STARTED>, C<TEXT_MESSAGE_START>, C<TEXT_MESSAGE_CONTENT>×N,
C<TEXT_MESSAGE_END>, C<RUN_FINISHED>. Sync mode synthesizes the same
sequence into a single response body for non-streaming clients.
=head1 SUPPORT
=head2 Issues
Please report bugs and feature requests on GitHub at
L<https://github.com/Getty/langertha-knarr/issues>.
=head2 IRC
lib/Langertha/Knarr/Protocol/Anthropic.pm view on Meta::CPAN
role => 'assistant',
model => $r->model // $request->model // 'unknown',
content => \@blocks,
stop_reason => $stop_reason,
stop_sequence => undef,
usage => $usage,
};
return ( 200, { 'Content-Type' => 'application/json' }, $self->_json->encode($payload) );
}
# Anthropic streaming uses named SSE events. We render full event blocks.
sub _sse_event {
my ($self, $event, $data) = @_;
return "event: $event\ndata: " . $self->_json->encode($data) . "\n\n";
}
sub format_stream_open {
my ($self, $request) = @_;
my $id = _msg_id();
my $model = $request->model // 'unknown';
return join( '',
lib/Langertha/Knarr/Protocol/Anthropic.pm view on Meta::CPAN
version 1.101
=head1 DESCRIPTION
Implements the Anthropic Messages wire format on top of
L<Langertha::Knarr::Protocol>. Loaded by default.
=over
=item * C<POST /v1/messages> — sync and named-event SSE streaming
=back
Streaming emits the full event sequence the Anthropic SDK expects:
C<message_start>, C<content_block_start>, C<content_block_delta>×N,
C<content_block_stop>, C<message_delta>, C<message_stop>.
=head1 SUPPORT
=head2 Issues
lib/Langertha/Knarr/Protocol/Ollama.pm view on Meta::CPAN
version 1.101
=head1 DESCRIPTION
Implements the Ollama wire format on top of
L<Langertha::Knarr::Protocol>. Loaded by default.
=over
=item * C<POST /api/chat>, C<POST /api/generate> — chat with NDJSON streaming
=item * C<GET /api/tags> — model listing
=item * C<GET /api/version> — version probe
=back
Streaming uses newline-delimited JSON (NDJSON) rather than SSE — the
C<Content-Type> is C<application/x-ndjson> and each chunk is a single
JSON object per line. The final chunk has C<done: true>.
lib/Langertha/Knarr/Protocol/OpenAI.pm view on Meta::CPAN
version 1.101
=head1 DESCRIPTION
Implements the OpenAI Chat Completions wire format on top of
L<Langertha::Knarr::Protocol>. Loaded by default in every
L<Langertha::Knarr> instance.
=over
=item * C<POST /v1/chat/completions> — sync and SSE streaming
=item * C<GET /v1/models> — model listing
=back
Streaming uses the standard SSE chunk format with C<data: [DONE]> as
the terminator. C<tools>, C<tool_choice>, and C<response_format> are
extracted into L<Langertha::Knarr::Request> attributes and forwarded to
the engine via C<chat_f>. Tool-call responses are serialised into
C<message.tool_calls> with C<finish_reason: "tool_calls">.
lib/Langertha/Knarr/Request.pm view on Meta::CPAN
=head2 model
Optional model id from the request body.
=head2 messages
ArrayRef of message hashes (C<< { role => ..., content => ... } >>).
=head2 stream
Boolean. Whether the client requested streaming.
=head2 temperature, max_tokens, tools, tool_choice, response_format, system
Optional generation parameters and tool definitions, if the protocol
extracted them. C<tool_choice> and C<response_format> are passed to
L<Langertha::Engine> via C<chat_f> in their canonical form; Langertha
normalizes them to the target engine's wire format.
=head2 session_id
lib/Langertha/Knarr/Response.pm view on Meta::CPAN
L<Langertha::Usage> on the way in; see L</usage>.
Every field is read behind a C<can()> guard so Knarr keeps working
against a L<Langertha> release that predates one of them — the older
attributes were added over several Langertha versions and C<timing> /
C<rate_limit> / C<thinking> are the most recent.
=head2 ttft_seconds
Time-to-first-token in seconds (Float) out of L</timing>, or C<undef>
when the engine did not measure it (non-streaming calls, or any handler
that is not engine-backed).
=head2 total_seconds
Total engine-measured call duration in seconds (Float) out of
L</timing>, or C<undef>.
=head2 has_tool_calls
True when C<tool_calls> contains at least one entry.
lib/Langertha/Knarr/Stream.pm view on Meta::CPAN
package Langertha::Knarr::Stream;
# ABSTRACT: Async chunk iterator returned by streaming Knarr handlers
our $VERSION = '1.101';
use Moose;
use Future;
sub from_callback {
my ($class, $setup) = @_;
my @queue;
my $pending;
my $finished = 0;
lib/Langertha/Knarr/Stream.pm view on Meta::CPAN
1;
__END__
=pod
=encoding UTF-8
=head1 NAME
Langertha::Knarr::Stream - Async chunk iterator returned by streaming Knarr handlers
=head1 VERSION
version 1.101
=head1 SYNOPSIS
use Langertha::Knarr::Stream;
# From a fixed list of strings
lib/Langertha/Knarr/Stream.pm view on Meta::CPAN
source => sub { $next_chunk_future },
);
# Drain it
while ( defined( my $chunk = $stream->next_chunk_f->get ) ) {
print $chunk;
}
=head1 DESCRIPTION
The chunk iterator that streaming Knarr handlers return. Supports two
construction modes: a sync C<generator> coderef that returns the next
chunk string each call (or C<undef> for end), or a C<source> coderef
that returns a L<Future> resolving to the next chunk string. The
Future form is the one real async backends like L<Net::Async::HTTP>
use; the generator form is for tests and simple cases.
=head2 generator
Optional. CodeRef returning the next chunk synchronously.
lib/Langertha/Knarr/Tracing.pm view on Meta::CPAN
variable values, which Docker C<--env-file> sometimes adds literally.
=head2 Timing sources
Knarr has two request paths and they do not measure latency the same way.
The generation's C<startTime> always marks the moment L</start_trace> ran;
what differs is where C<endTime> and C<completionStartTime> come from.
=over
=item * B<Routed, non-streaming> — a L<Langertha> engine produced a
L<Langertha::Response>, so L<Langertha::Knarr::Handler::Tracing> hands the
engine-measured C<timing> hash to L</end_trace>. C<endTime> becomes
C<startTime + total_seconds> and C<completionStartTime> becomes
C<startTime + ttft_seconds>, both anchored to the high-resolution
timestamp L</start_trace> recorded. This is the only path with a real
time-to-first-token, and the durations exclude the proxy's own
formatting overhead.
=item * B<Routed, streaming> — the decorator accumulates deltas and never
sees a response object, so there is no C<timing>. C<endTime> is the
wall-clock moment the stream was exhausted and no C<completionStartTime>
is emitted.
=item * B<Raw passthrough> — bytes are piped 1:1 and never parsed, so no
L<Langertha::Response> exists at all. C<endTime> is again the proxy's own
wall clock at L</end_trace>, which includes network time to the upstream
provider.
=back
t/40_streaming_live.t view on Meta::CPAN
$sb->start;
# Discover the actual port we bound to
my $sock = $sb->_server->read_handle;
my $port = $sock->sockport;
ok( $port, "bound to port $port" );
my $http = Net::Async::HTTP->new;
$loop->add($http);
# --- 1) non-streaming OpenAI roundtrip ---
{
my $body = $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
});
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port/v1/chat/completions" );
$req->header( 'Content-Type' => 'application/json' );
$req->content($body);
my $resp = $http->do_request( request => $req )->get;
is( $resp->code, 200, 'non-stream 200' );
my $data = $json->decode( $resp->decoded_content );
is( $data->{object}, 'chat.completion', 'non-stream object' );
is( $data->{choices}[0]{message}{content}, 'hello world', 'non-stream content' );
}
# --- 2) streaming OpenAI ---
{
my $body = $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
});
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port/v1/chat/completions" );
$req->header( 'Content-Type' => 'application/json' );
$req->content($body);
my $resp = $http->do_request( request => $req )->get;
is( $resp->code, 200, 'stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE content type' );
my $body_text = $resp->decoded_content;
my @data_lines = grep { /^data:/ } split /\n/, $body_text;
ok( scalar(@data_lines) >= 4, 'multiple SSE data lines emitted' );
like( $body_text, qr/hel/, 'first chunk present' );
like( $body_text, qr/ld/, 'last chunk present' );
}
# --- 3) streaming Ollama (NDJSON) ---
{
my $body = $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
});
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port/api/chat" );
$req->header( 'Content-Type' => 'application/json' );
$req->content($body);
my $resp = $http->do_request( request => $req )->get;
t/41_streaming_protocols_live.t view on Meta::CPAN
$loop->add($http);
sub post {
my ($path, $body) = @_;
my $req = HTTP::Request->new( POST => "http://127.0.0.1:$port$path" );
$req->header( 'Content-Type' => 'application/json' );
$req->content( $json->encode($body) );
return $http->do_request( request => $req )->get;
}
# --- Anthropic streaming: named SSE events ---
{
my $resp = post( '/v1/messages', {
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
});
is( $resp->code, 200, 'anthropic stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE type' );
my $body = $resp->decoded_content;
like( $body, qr/event: message_start/, 'message_start frame' );
t/41_streaming_protocols_live.t view on Meta::CPAN
like( $body, qr/event: content_block_stop/, 'block_stop frame' );
like( $body, qr/event: message_stop/, 'message_stop frame' );
like( $body, qr/"text":"hel"/, 'first chunk text in delta' );
like( $body, qr/"text":"ld"/, 'last chunk text in delta' );
# Count content_block_delta events — should equal number of stream chunks (4).
my $deltas = () = $body =~ /event: content_block_delta/g;
is( $deltas, 4, 'four content_block_delta events' );
}
# --- A2A streaming via JSON-RPC tasks/sendSubscribe ---
{
my $resp = post( '/', {
jsonrpc => '2.0',
id => 1,
method => 'tasks/sendSubscribe',
params => {
id => 'task-1',
message => { role => 'user', parts => [ { type => 'text', text => 'hi' } ] },
},
});
t/41_streaming_protocols_live.t view on Meta::CPAN
like( $body, qr/"text":"hel"/, 'first chunk text' );
like( $body, qr/"text":"ld"/, 'last chunk text' );
like( $body, qr/"final":true/, 'final flag set on close' );
# Each delta should be a JSON-RPC envelope with the original id.
my $first_delta = (grep { /artifact/ } @data)[0];
ok( $first_delta, 'has artifact frame' );
like( $first_delta, qr/"id":1/, 'jsonrpc id preserved' );
}
# --- ACP streaming ---
{
my $resp = post( '/runs', {
agent_name => 'm',
mode => 'stream',
input => [ { parts => [ { content_type => 'text/plain', content => 'hi' } ] } ],
});
is( $resp->code, 200, 'acp stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE type' );
my $body = $resp->decoded_content;
like( $body, qr/event: run\.created/, 'run.created' );
t/41_streaming_protocols_live.t view on Meta::CPAN
like( $body, qr/event: message\.part/, 'message.part frames' );
like( $body, qr/event: message\.completed/, 'message.completed' );
like( $body, qr/event: run\.completed/, 'run.completed' );
like( $body, qr/"content":"hel"/, 'first chunk content' );
like( $body, qr/"content":"ld"/, 'last chunk content' );
my $parts = () = $body =~ /event: message\.part/g;
is( $parts, 4, 'four message.part events' );
}
# --- AG-UI streaming ---
{
my $resp = post( '/awp', {
threadId => 'th-1',
runId => 'run-1',
messages => [ { role => 'user', content => 'hi' } ],
});
is( $resp->code, 200, 'agui stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE type' );
my $body = $resp->decoded_content;
like( $body, qr/"type":"RUN_STARTED"/, 'RUN_STARTED' );
t/60_psgi.t view on Meta::CPAN
handler => Langertha::Knarr::Handler::Code->new(
code => sub { 'psgi-said-hi' },
stream_code => sub { my @p = ('a','b','c'); sub { @p ? shift @p : undef } },
),
port => 0,
);
my $app = Langertha::Knarr::PSGI->new( knarr => $sb )->to_app;
my $test = Plack::Test->create($app);
# Non-streaming OpenAI
{
my $res = $test->request(
POST '/v1/chat/completions',
Content_Type => 'application/json',
Content => $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
}),
);
is( $res->code, 200, 'psgi non-stream 200' );
my $d = $json->decode($res->decoded_content);
is( $d->{choices}[0]{message}{content}, 'psgi-said-hi', 'psgi content' );
}
# Buffered streaming OpenAI
{
my $res = $test->request(
POST '/v1/chat/completions',
Content_Type => 'application/json',
Content => $json->encode({
model => 'm',
messages => [ { role => 'user', content => 'hi' } ],
stream => JSON::MaybeXS::true(),
}),
);
t/70_passthrough.t view on Meta::CPAN
use JSON::MaybeXS;
use Langertha::Knarr;
use Langertha::Knarr::Handler::Code;
use Langertha::Knarr::Handler::Passthrough;
my $json = JSON::MaybeXS->new( utf8 => 1, canonical => 1 );
my $loop = IO::Async::Loop->new;
# --- Backend: a Knarr that answers OpenAI/Anthropic/Ollama with a known string,
# plus token-by-token streaming so the passthrough's stream-extractor can
# unwrap the protocol-native chunks.
my $backend_handler = Langertha::Knarr::Handler::Code->new(
code => sub {
my ($s, $r) = @_;
my $u = $r->messages->[-1] // {};
return "BACKEND: " . ( $u->{content} // '' );
},
stream_code => sub {
my @parts = ('BACK', 'END:', ' hi');
return sub { @parts ? shift @parts : undef };
t/70_passthrough.t view on Meta::CPAN
{
my $resp = post_json(
"http://127.0.0.1:$fport/v1/messages",
{ model => 'm', messages => [ { role => 'user', content => 'sup' } ] },
);
is( $resp->code, 200, 'anthropic passthrough sync 200' );
my $d = $json->decode($resp->decoded_content);
is( $d->{content}[0]{text}, 'BACKEND: sup', 'anthropic content via passthrough' );
}
# --- 4) OpenAI streaming passthrough ---
{
my $resp = post_json(
"http://127.0.0.1:$fport/v1/chat/completions",
{ model => 'm', messages => [ { role => 'user', content => 'hi' } ], stream => JSON::MaybeXS::true() },
);
is( $resp->code, 200, 'openai passthrough stream 200' );
is( $resp->header('Content-Type'), 'text/event-stream', 'SSE content type' );
my $body = $resp->decoded_content;
my @data = grep { /^data:/ && !/\[DONE\]/ } split /\n/, $body;
ok( scalar(@data) >= 3, 'multiple SSE chunks forwarded' );
like( $body, qr/BACK/, 'first chunk text present' );
like( $body, qr/END:/, 'second chunk text present' );
}
# --- 5) Ollama streaming passthrough ---
{
my $resp = post_json(
"http://127.0.0.1:$fport/api/chat",
{ model => 'm', messages => [ { role => 'user', content => 'hi' } ], stream => JSON::MaybeXS::true() },
);
is( $resp->code, 200, 'ollama passthrough stream 200' );
is( $resp->header('Content-Type'), 'application/x-ndjson', 'NDJSON content type' );
my @lines = grep { length } split /\n/, $resp->decoded_content;
ok( scalar(@lines) >= 4, 'multiple NDJSON lines forwarded' );
my $last = $json->decode($lines[-1]);
t/77_raw_passthrough_tracing.t view on Meta::CPAN
is( $start->{kind}, 'start', 'first event is start' );
is( $start->{engine}, 'passthrough', 'engine tagged as passthrough' );
is( $start->{model}, 'gpt-mystery', 'model recorded' );
is( $start->{format}, 'openai', 'format recorded' );
my $end = $tracer->events->[1];
is( $end->{kind}, 'end', 'second event is end' );
is( $end->{id}, $start->{id}, 'end matches start id' );
}
# --- 3) Raw passthrough streaming ---
{
$tracer->events->@* = (); # reset
my $resp = post_json(
"http://127.0.0.1:$fport/v1/chat/completions",
{ model => 'gpt-unknown', messages => [ { role => 'user', content => 'stream' } ],
stream => JSON::MaybeXS::true() },
);
is( $resp->code, 200, 'raw passthrough stream 200' );
like( $resp->decoded_content, qr/UP/, 'stream chunk present' );
like( $resp->decoded_content, qr/STREAM/, 'stream chunk 2 present' );
# Trace was created for streaming too
ok( scalar @{ $tracer->events } >= 2, 'tracer has stream events' );
is( $tracer->events->[0]{kind}, 'start', 'stream trace start' );
is( $tracer->events->[0]{model}, 'gpt-unknown', 'stream model recorded' );
is( $tracer->events->[-1]{kind}, 'end', 'stream trace end' );
}
# --- 4) Anthropic raw passthrough with auth headers ---
{
$tracer->events->@* = ();
( run in 3.183 seconds using v1.01-cache-2.11-cpan-b16cb0d3907 )