diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 0000000..9fd0526
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,165 @@
+Nats.raku Agents Guide
+
+Purpose
+- This document standardizes how agentic coding tools (Cursor, Copilot, OpenCode, etc.) interact with this repository: how to build, lint, test, and how to write code consistent with the existing style.
+
+Project Overview
+- Language: Raku (Rakudo)
+- Distribution name: `Nats` (META6.json), a client library for NATS.
+- Library modules live under `lib/` (e.g., `lib/Nats.rakumod`, `lib/Nats/Message.rakumod`).
+- Tests live in `t/*.rakutest` and use `Test` and `Test::Mock`.
+- Examples live in `examples/`.
+- Integration tests live in `integration-tests/1/` using Docker Compose.
+- CI: `.github/workflows/test.yml` uses `JJ/raku-test-action@v2` with coverage.
+
+Environment Variables
+- `NATS_URL`: default server URL used by `Nats.default-url` if set; otherwise `nats://127.0.0.1:4222`.
+- `NATS_DEBUG`: when truthy, enables debug logging via `note` in `Nats!debug`.
+- `NATS_SERVERS`: comma-separated list of URLs for integration tests.
+
+Build, Install, Lint, Test
+- Dependencies
+  - Install only dependencies: `zef install --depsonly .`
+  - Install the distribution locally: `zef install .`
+- Build
+  - Raku modules are interpreted; there is no compile step beyond syntax checks.
+  - Packaging metadata is in `META6.json`; release process uses `dist.ini` (ReadmeFromPod, UploadToZef, Badges).
+- Lint / Syntax Check
+  - Per-file syntax check: `raku -c lib/Nats.rakumod`
+  - Batch syntax check (examples/tests): `raku -c examples/request.raku` and `raku -c t/nats.rakutest`
+  - Optional formatter: if you use `rakufmt`, keep its output consistent with the style rules below; do not auto-format CI unless agreed.
+- Run All Tests
+  - Using prove6: `prove6 -Ilib -v t/*.rakutest`
+  - Using zef: `zef test .`
+- Run a Single Test File
+  - With prove6: `prove6 -Ilib -v t/message.rakutest`
+  - Directly with raku: `raku -Ilib t/message.rakutest`
+- Coverage (CI)
+  - CI runs `JJ/raku-test-action@v2` with `coverage: true`. Locally, prefer the same test commands; coverage tooling is provided by the action in CI.
+- Integration Tests
+  - From `integration-tests/1/`: `docker compose up --build`
+  - Container `test` uses `ENTRYPOINT ["raku", "/test.raku"]` and requires `nats` service reachable with `NATS_SERVERS`.
+
+Runtime and Concurrency Notes
+- Starts use asynchronous sockets (`IO::Socket::Async`) and event supplies (`Supply`, `Supplier`).
+- `Nats.start` returns a `Promise`; await it when chaining work that must start after connection.
+- Messages are emitted on `Nats.supply`; tap the supply to process messages.
+- JetStream helpers live in `lib/Nats/JetStream.rakumod` and use `request`/`reply` patterns.
+
+Code Style Guidelines
+
+Imports and Module Organization
+- Place `use` statements at the top of the file, after the `unit class` or `unit grammar` line if present.
+- Use fully qualified module names (e.g., `use Nats::Message;`, `use JSON::Fast;`).
+- In tests, include library path explicitly: `use lib 'lib';` followed by `use Nats;`.
+- Keep module boundaries clear: classes under `Nats::*` go in matching `lib/Nats/*.rakumod` paths and are declared with `unit class Nats::Name;`.
+
+Formatting
+- Indentation: 4 spaces; no tabs.
+- Line length: target <= 100 chars; wrap thoughtfully without breaking readability.
+- Braces: opening brace on the same line; closing brace aligned with the start of the block.
+- Spacing: around operators and after commas; avoid trailing whitespace.
+- Blank lines: use to separate logical sections (attributes, methods, private helpers).
+- Comments: write concise, purposeful comments only when the intent is non-obvious (e.g., protocol framing or parsing assumptions).
+
+Types and Signatures
+- Prefer typed attributes and parameters: `has Str $.subject;`, `method publish(Str $subject, Str() $payload = "") { ... }`.
+- Use type constraints for optional values: `Str() $payload?` and named parameters for flags and options.
+- For numeric counters and IDs, use `UInt` where appropriate (e.g., SIDs, counts).
+- Enforce interface capabilities with `where` when needed (e.g., `has $.nats where { .^can('publish') }`).
+- Use multi methods when overloading by type or arity makes intent clearer (see `unsubscribe` multis in `lib/Nats.rakumod`).
+
+Naming Conventions
+- Modules and classes: `Nats` and `Nats::*` (PascalCase after the top-level namespace).
+- Attributes and methods: lower-case with hyphens only when idiomatic to Raku (e.g., `reply-json`), otherwise lower-case with dashes avoided in general-purpose names.
+- Constants: UPPERCASE with hyphens or underscores as in `JS-API`, `STREAM-CREATE`; keep consistency with existing JetStream constants.
+- Private helpers: prefix with `!` (e.g., `method !print`, `method !debug`); do not expose them in public APIs.
+- Test names: human-readable strings in `pass/is/ok` messages that state behavior succinctly.
+
+Error Handling
+- Use `Nats::Error is Exception` for domain-specific exceptions when throwing from library code; construct with `:message`.
+- In protocol handlers, map NATS `-ERR` messages to exceptions (current code uses `die $cmd.data`). Prefer explicit `die Nats::Error.new(:message($cmd.data))` when enhancing error semantics.
+- Do not swallow exceptions silently; emit them or fail the promise/supply appropriately.
+- For recoverable states (e.g., `PING`/`PONG` flow), keep logic non-throwing and side-effectful as implemented.
+
+Logging and Debugging
+- Use `self!debug(*@msg)` for structured debug output; it checks `NATS_DEBUG` and writes with `note`.
+- Do not leave `say/diag` calls in library code unless behind debug flags; tests can use `diag`.
+
+Protocol and Parsing
+- Grammar: define parsing in `unit grammar` (`lib/Nats/Grammar.rakumod`); keep tokens small and purposeful.
+- Actions: construct domain objects in `Nats::Actions` with `make` and typed values (e.g., `:+$<sid>` to `UInt`).
+- Keep message framing rules explicit: size-limited payload, CRLF boundaries, and optional `reply-to` subjects.
+
+Concurrency and Supplies
+- Subscribe flow: create `Nats::Subscription`, attach a filtered `Supply` using `messages-from-supply`, tap and dispatch to user blocks.
+- Unsubscribe by signaling via `UNSUB` with optional `:max-messages` and deleting SID from registry.
+- When writing new reactive flows, prefer `react/whenever` or `Supply.tap` consistent with existing patterns.
+
+JSON Handling
+- Use `JSON::Fast` exclusively for JSON encode/decode; prefer `to-json` for emitting and `from-json` for parsing.
+- Keep JSON payloads as `Str` in `Nats::Message`; expose `.json` to parse lazily and throw on invalid JSON.
+
+JetStream Helpers
+- Use `sprintf` formatting for subject templates (`method subject`) to avoid manual string assembly.
+- Stream and Consumer configuration should return maps convertible with `to-json` and keep defaults aligned with NATS expectations.
+
+Testing Practices
+- Tests live under `t/` and use `use lib 'lib';` at the top.
+- Mock IO and dependencies with `Test::Mock` as in current tests (`mocked IO::Socket::Async`, `mocked Nats`).
+- Prefer `use-ok`, `can-ok`, `isa-ok`, `lives-ok`, `dies-ok`, `check-mock` idioms already present.
+- Keep tests deterministic; use `Supplier` to emit messages and verify behaviors.
+- For a quick ad-hoc run of a test file: `raku -Ilib t/nats.rakutest`.
+
+Examples and Demos
+- Examples under `examples/` demonstrate basic usage with `react`/`whenever` and subscription DSL.
+- Run an example: `raku -Ilib examples/request.raku` (ensure NATS server running and `NATS_URL` set if needed).
+
+Repository Conventions
+- Do not introduce non-ASCII unless necessary (e.g., literal protocol examples); keep source ASCII by default.
+- Keep public API surface stable; add new features under `Nats::*` modules with clear responsibilities.
+- Avoid global mutable state except the per-process `@*SUBSCRIPTIONS` in the subscription DSL; reset it in `subscriptions(&block)` as implemented.
+
+CI and Automation
+- CI uses `.github/workflows/test.yml`:
+  - `JJ/raku-test-action@v2` with `coverage: true`.
+  - Ensure tests pass locally before pushing.
+- Badges and metadata defined via `dist.ini`; leave publishing steps to maintainers.
+
+Cursor / Copilot Rules
+- No Cursor rules found in `.cursor/rules/` or `.cursorrules`.
+- No Copilot instructions found in `.github/copilot-instructions.md`.
+- Agents should follow this AGENTS.md for guidance in the absence of tool-specific rule files.
+
+Common Command Cheat Sheet
+
+```sh
+# Install dependencies only
+zef install --depsonly .
+
+# Install locally
+zef install .
+
+# Syntax-check core modules
+raku -c lib/Nats.rakumod
+raku -c lib/Nats/Message.rakumod
+
+# Run all tests with verbose output
+prove6 -Ilib -v t/*.rakutest
+
+# Run a single test file
+prove6 -Ilib -v t/message.rakutest
+# or
+raku -Ilib t/message.rakutest
+
+# Run an example (requires NATS server)
+
+# Integration test (docker compose)
+cd integration-tests/1
+docker compose up --build
+```
+
+When In Doubt
+- Mirror existing patterns; do not invent new frameworks or paradigms.
+- Prefer explicit types, small methods, and clear protocol boundaries.
+- Keep behavior changes minimal; add tests for new features.
diff --git a/META6.json b/META6.json
index 509c527..21643bc 100644
--- a/META6.json
+++ b/META6.json
@@ -11,7 +11,7 @@
     "JSON::Fast",
     "URL"
   ],
-  "description": "NATS client for Raku",
+  "description": "NATS client for Raku with JetStream support",
   "license": "Artistic-2.0",
   "name": "Nats",
   "perl": "6.d",
@@ -23,6 +23,7 @@
     "Nats::Error": "lib/Nats/Error.rakumod",
     "Nats::Grammar": "lib/Nats/Grammar.rakumod",
     "Nats::JetStream": "lib/Nats/JetStream.rakumod",
+    "Nats::JetStream::Ackable": "lib/Nats/JetStream/Ackable.rakumod",
     "Nats::Message": "lib/Nats/Message.rakumod",
     "Nats::Replyable": "lib/Nats/Replyable.rakumod",
     "Nats::Subscription": "lib/Nats/Subscription.rakumod",
@@ -30,10 +31,13 @@
   },
   "resources": [
   ],
-  "source-url": "https://github.com/FCO/nats.git",
+  "source-url": "https://github.com/FCO/nats.raku.git",
   "tags": [
+    "nats",
+    "jetstream",
+    "messaging"
   ],
   "test-depends": [
   ],
-  "version": "0.0.1"
+  "version": "0.1.0"
 }
diff --git a/examples/js-consume-pull.raku b/examples/js-consume-pull.raku
new file mode 100644
index 0000000..c913d57
--- /dev/null
+++ b/examples/js-consume-pull.raku
@@ -0,0 +1,40 @@
+#!/usr/bin/env raku
+
+use Nats;
+use Nats::JetStream;
+use JSON::Fast;
+
+my $url = %*ENV<NATS_URL> // 'nats://127.0.0.1:4222';
+
+my $n = Nats.new: :servers[$url];
+await $n.start;
+$n.connect;
+
+say "creating stream";
+my $s = $n.stream: 'TEST', :subjects['js.test'];
+say "awaiting stream create response";
+my $resp = await $s.create;
+say "created stream: { $resp.payload }";
+
+my $c = $s.consumer('dur', filter-subject => 'js.test');
+say "awaiting consumer create response";
+my $cres = await $c.create-named;
+say "consumer created: { $cres.payload }";
+say "pull consumer created via named endpoint (wrapped config)";
+say "payload: { to-json { :config($c.config(:include-durable(False))) } }";
+
+say "starting pull loop";
+
+# Pull and print messages in repeated batches until we see at least 15
+my $seen = 0;
+react {
+    # Pull first 15 messages; skip status/no-message frames
+    whenever $c.msgs: :15batch, :no-wait -> $msg {
+        next unless $msg.payload.defined && $msg.payload.chars;
+        say "PAYLOAD: ", $msg.payload;
+        if $msg.^can('ack') { $msg.ack }
+        done if ++$seen >= 15
+    }
+}
+
+$n.stop;
diff --git a/examples/js-produce.raku b/examples/js-produce.raku
new file mode 100644
index 0000000..14ace1d
--- /dev/null
+++ b/examples/js-produce.raku
@@ -0,0 +1,28 @@
+#!/usr/bin/env raku
+
+use Nats;
+use Nats::JetStream;
+
+my $url = %*ENV<NATS_URL> // 'nats://127.0.0.1:4222';
+
+my $n = Nats.new: :servers[$url];
+
+say "starting client";
+await $n.start;
+$n.connect;
+say "client started";
+
+say "creating stream";
+my $s = $n.stream: 'TEST', :subjects['js.test'];
+say "awaiting stream create response";
+my $resp = await $s.create;
+say "created stream: { $resp.payload }";
+
+# publish a few messages
+for 1..15 -> $i {
+    say qq<pushing: 'js.test', "msg $i">;
+    $n.publish('js.test', "msg $i");
+}
+
+say 'Produced 15 messages to js.test';
+$n.stop;
diff --git a/lib/Nats.rakumod b/lib/Nats.rakumod
index 37af462..e36581e 100644
--- a/lib/Nats.rakumod
+++ b/lib/Nats.rakumod
@@ -15,6 +15,8 @@ has URL()    @.servers   = self.default-url;
 has Promise  $!conn     .= new;
 has Supplier $!supplier .= new;
 has Supply   $.supply    = $!supplier.Supply;
+has Bool()   $.headers-supported = False;
+has Str      $!buffer    = '';
 
 has Bool() $!DEBUG = %*ENV<NATS_DEBUG>;
 
@@ -50,8 +52,22 @@ method stop {
 
 method handle-input {
     $!conn.result.Supply.tap: -> $line {
-        self!in($line);
-        my @cmds = Nats::Grammar.parse($line, :actions(Nats::Actions.new: :nats(self))).ast;
+        $!buffer ~= $line;
+        self!process-buffer;
+    }
+}
+
+method !process-buffer {
+    loop {
+        my $before = $!buffer;
+        my $match = Nats::Grammar.parse($!buffer, :actions(Nats::Actions.new: :nats(self)));
+        last unless $match;
+
+        my $consumed = $match.to;
+        $!buffer = $!buffer.substr($consumed);
+
+        self!in($match.Str);
+        my @cmds = $match.ast;
         for @cmds -> $cmd {
             given $cmd {
                 when Nats::Data {
@@ -60,7 +76,19 @@ method handle-input {
                         when "err"  { die $cmd.data      }
                         when "ping" { self!print: "PONG" }
                         when "pong" {                    }
-                        when "info" {                    }
+                        when "info" {
+                            my %info = $cmd.data;
+                            $!DEBUG && self!debug("INFO", to-json %info);
+                            if %info<headers>:exists { $!headers-supported = %info<headers> ?? True !! False }
+                            if %info<connect_urls>:exists {
+                                my @urls = %info<connect_urls>.
+                                    map({ $_ ~~ /':'/ && $_ !~~ /^'nats://' /
+                                        ?? "nats://$_"
+                                        !! $_ }).
+                                    map({ URL.new: .Str });
+                                @!servers = @urls if @urls.elems;
+                            }
+                        }
                     }
                 }
                 when Nats::Message { $!supplier.emit: $_ }
@@ -70,7 +98,7 @@ method handle-input {
 }
 
 method connect {
-    self!print: "CONNECT", to-json :!pretty, %();
+    self!print: "CONNECT", to-json :!pretty, { :headers };
 }
 
 method ping {
@@ -98,16 +126,20 @@ method !gen-inbox {
     $inbox
 }
 
-method request(
-    Str   $subject,
-    Str() $payload?,
-    Str   :$reply-to     = self!gen-inbox,
-    UInt  :$max-messages = 1,
-) {
-    my $sub = self.subscribe: $reply-to, :$max-messages;
-    self.publish: $subject, |(.Str with $payload), :$reply-to;
-    $sub.supply.head: $max-messages;
-}
+ method request(
+     Str   $subject,
+     Str() $payload?,
+     Str   :$reply-to     = self!gen-inbox,
+     UInt  :$max-messages = 1,
+     :headers(%headers),
+ ) {
+     my $sub = self.subscribe: $reply-to, |($max-messages ?? :$max-messages !! Empty);
+     return $sub.supply unless $max-messages;
+     my $p = $sub.supply.head($max-messages);
+     self.publish: $subject, |(.Str with $payload), :$reply-to,
+         |( %headers.elems ?? :headers(%headers) !! Empty );
+     $p
+ }
 
 multi method unsubscribe(Nats::Subscription $sub, UInt :$max-messages) {
     self.unsubscribe: $sub.sid, |(:$max-messages with $max-messages)
@@ -118,8 +150,65 @@ multi method unsubscribe(UInt $sid, UInt :$max-messages) {
     %!subs{$sid}:delete;
 }
 
-method publish(Str $subject, Str() $payload = "", Str :$reply-to) {
-    self!print: "PUB", $subject, $reply-to // Empty, "{ $payload.chars }\r\n$payload";
+method publish(
+    Str   $subject,
+    Str() $payload = "",
+    Str   :$reply-to,
+          :headers(%headers),
+    Bool  :$ack = False,
+    Str   :$msg-id,
+    UInt  :$timeout = 5,
+) {
+    return self!publish-with-ack: $subject, $payload, :$msg-id, :$timeout if $ack;
+    %headers && %headers.elems
+        ?? self!hpub($subject, $payload, :%headers, :$reply-to)
+        !! self!pub($subject, $payload, :$reply-to)
+}
+
+method !publish-with-ack(
+    Str   $subject,
+    Str() $payload = "",
+    Str   :$msg-id,
+    UInt  :$timeout = 5,
+) {
+    my %headers;
+    %headers<Nats-Msg-Id> = $msg-id if $msg-id.defined && $msg-id.chars;
+
+    my $reply-to = self!gen-inbox;
+    my $sub      = self.subscribe: $reply-to, :max-messages(1);
+
+    # Tap BEFORE publish — avoids race where PubAck arrives before we listen
+    my $p = start await $sub.supply.head.Promise;
+
+    self.publish: $subject, $payload, :$reply-to,
+        |( %headers.elems ?? :headers(%headers) !! Empty );
+
+    await Promise.anyof: $p, Promise.in($timeout);
+    $p.so ?? $p.result !! Nil
+}
+
+method !pub(Str $subject, Str() $payload = "", Str :$reply-to) {
+    self!print: "PUB", $subject, $reply-to // Empty, "{ $payload.encode('utf8').bytes }\r\n$payload";
+}
+
+method !hpub(
+    Str   $subject,
+    Str() $payload = "",
+    :%headers,
+    Str   :$reply-to,
+) {
+    my @lines = ("NATS/1.0", |(%headers.kv.map: -> $k, $v {
+        $v ~~ Positional
+            ?? $v.map({ "{ $k }: { $_ }" })
+            !! "{ $k }: { $v }"
+    }).flat);
+    my $headers-lines = @lines.join("\r\n");
+    my $headers-block = $headers-lines ~ "\r\n\r\n"; # includes CRLFCRLF
+    my $payload-str   = $payload // "";
+    my UInt $hsize    = $headers-block.encode('utf8').bytes;
+    my UInt $tsize    = $hsize + $payload-str.encode('utf8').bytes;
+    # NB: !print adds trailing \r\n, so the payload CRLF serves as the HPUB terminator
+    self!print: "HPUB", $subject, $reply-to // Empty, $hsize, "$tsize\r\n$headers-block$payload-str";
 }
 
 method stream($name, *@subjects, |c) {
diff --git a/lib/Nats/Actions.rakumod b/lib/Nats/Actions.rakumod
index 10b275d..6308ea2 100644
--- a/lib/Nats/Actions.rakumod
+++ b/lib/Nats/Actions.rakumod
@@ -1,6 +1,7 @@
 use JSON::Fast;
 use Nats::Message;
 use Nats::Data;
+use Nats::JetStream::Ackable;
 unit class Nats::Actions;
 
 has $.nats;
@@ -13,11 +14,27 @@ method msg-option:sym<PING>($/) {make Nats::Data.new: :type<ping>}
 method msg-option:sym<PONG>($/) {make Nats::Data.new: :type<pong>}
 method msg-option:sym<INFO>($/) {make Nats::Data.new: :type<info>, :data(from-json ~$<info>)}
 method msg-option:sym<MSG>($/) {
-    make Nats::Message.new:
+    my $msg = Nats::Message.new:
         :subject($<subject>.made),
         :sid(+$<sid>),
         |(:reply-to(.Str) with $<reply-to>),
         :payload(~$<payload>),
         :$!nats,
     ;
+    if $<reply-to> && $<reply-to>.Str.starts-with('$JS.ACK') {
+        $msg does Nats::JetStream::Ackable;
+    }
+    make $msg;
+}
+method msg-option:sym<HMSG>($/) {
+    # For now, expose raw header+payload block; future enhancement can parse headers
+    my $msg = Nats::Message.new:
+        :subject($<subject>.made),
+        :sid(+$<sid>),
+        |(:reply-to(.Str) with $<reply-to>),
+        :payload(~$<hpayload>),
+        :$!nats,
+    ;
+    $msg does Nats::JetStream::Ackable if $<reply-to>;
+    make $msg;
 }
diff --git a/lib/Nats/Grammar.rakumod b/lib/Nats/Grammar.rakumod
index 85be77f..cb9d66c 100644
--- a/lib/Nats/Grammar.rakumod
+++ b/lib/Nats/Grammar.rakumod
@@ -2,10 +2,12 @@
 unit grammar Nats::Grammar;
 
 token subject {
-    [ \w+ ]+ %% '.'
+    # Allow standard NATS subject charset including '$' for JetStream ack subjects,
+    # alphanumerics, underscore, star and '>' for wildcards; literal '-' included.
+    [ <[ A..Z a..z 0..9 _ $ * > ]>+ '-'* ]+ %% '.'
 }
 token TOP {
-    <msg-option>+ %% \n
+    [<msg-option> \n*]+
 }
 token sid { \d+ }
 token size { \d+ }
@@ -13,7 +15,16 @@ token payload(UInt $size) {
     <(
         . ** { $size }
     )>
-    <?before \n [\n | $]>
+    <?before \n | $>
+    \n
+}
+token hsize { \d+ }
+token tsize { \d+ }
+token hpayload(UInt $hsize, UInt $tsize) {
+    <(
+        . ** { $tsize }
+    )>
+    <?before \n | $>
     \n
 }
 proto token msg-option           { * }
@@ -33,3 +44,15 @@ token msg-option:sym<MSG>  {
     {}
     <payload(+$<size>)>
 }
+token msg-option:sym<HMSG>  {
+    <.sym>    \s+
+    <subject> \s+
+    <sid>     \s+
+    [
+        <reply-to=.subject> \s+
+    ]??
+    <hsize>   \s+
+    <tsize>   \n
+    {}
+    <hpayload(+$<hsize>, +$<tsize>)>
+}
diff --git a/lib/Nats/JetStream.rakumod b/lib/Nats/JetStream.rakumod
index 0a32808..40681ba 100644
--- a/lib/Nats/JetStream.rakumod
+++ b/lib/Nats/JetStream.rakumod
@@ -1,4 +1,5 @@
 use JSON::Fast;
+use Nats::Message;
 
 # Constantes principais
 constant JS = '$JS';
@@ -7,116 +8,270 @@ constant JS-ACK = JS ~ '.ACK';
 
 # Stream Subjects
 constant STREAM-CREATE     = JS-API ~ '.STREAM.CREATE.%s';
+constant STREAM-UPDATE     = JS-API ~ '.STREAM.UPDATE.%s';
 constant STREAM-INFO       = JS-API ~ '.STREAM.INFO.%s';
 constant STREAM-DELETE     = JS-API ~ '.STREAM.DELETE.%s';
 constant STREAM-LIST       = JS-API ~ '.STREAM.LIST';
+constant STREAM-NAMES      = JS-API ~ '.STREAM.NAMES';
+constant STREAM-PURGE       = JS-API ~ '.STREAM.PURGE.%s';
+
+# Direct Message Subjects
+constant DIRECT-GET         = JS-API ~ '.DIRECT.GET.%s';
+constant DIRECT-GET-LAST    = JS-API ~ '.DIRECT.GET.%s.%s';
 
 # Consumer Subjects
 constant CONSUMER-CREATE   = JS-API ~ '.CONSUMER.CREATE.%s.%s';
 constant CONSUMER-INFO     = JS-API ~ '.CONSUMER.INFO.%s.%s';
 constant CONSUMER-DELETE   = JS-API ~ '.CONSUMER.DELETE.%s.%s';
+constant CONSUMER-LIST     = JS-API ~ '.CONSUMER.LIST.%s';
 constant CONSUMER-MSG-NEXT = JS-API ~ '.CONSUMER.MSG.NEXT.%s.%s';
 
+# Convert object attributes to a JetStream-compatible Map (kebab→snake_case)
 sub to-map($obj, *%pars --> Map()) {
     $obj.^attributes.map: -> $attr {
-        my $name = $attr.name.substr(2).subst: /_/, "-", :g;
-        next if %pars{$name}:e &&!%pars{$name};
+        my $name = $attr.name.substr(2).subst: /'-'/, "_", :g;
+        next if %pars{$name}:e && !%pars{$name};
         my $val = $attr.get_value: $obj;
-        next unless $val ~~ Str | Int | Positional | Associative | Nil;
+        next unless $val.defined && $val ~~ Str | Int | Positional | Associative;
+        next if $val ~~ Associative && $val.elems == 0;
+        next if $val ~~ Positional && $val.elems == 0;
         $name => $val
     }
 }
 
+# Parse JetStream API response into a Map (snake_case→kebab for Raku attrs)
+sub from-map(%data, $obj) is export {
+    for $obj.^attributes -> $attr {
+        my $name = $attr.name.substr(2);
+        my $js-name = $name.subst: /'-'/, "_", :g;
+        next unless %data{$js-name}:exists;
+        given $attr.type {
+            when Str  { $attr.set_value: $obj, %data{$js-name}.Str }
+            when Int  { $attr.set_value: $obj, %data{$js-name}.Int }
+            when Bool { $attr.set_value: $obj, ?%data{$js-name}    }
+            default   { $attr.set_value: $obj, %data{$js-name}     }
+        }
+    }
+    $obj
+}
+
 class Nats::Consumer {...}
 
 class Nats::Stream {
 
     has       $.nats is required;
     has Str() $.name is required;
-    has Str() @.subjects,
-    has Str() $.retention = 'limits',
-    has Str() $.storage = 'file',
-    has Int() $.max-msgs = -1,
-    has Int() $.max-bytes = -1,
-    has Int() $.max-age = 0,
-
-    #enum RetentionPolicy <limits interest work_queue>;
-    #enum DiscardPolicy <old new>;
-    #enum StorageType <file memory any>;
-    ##enum Placement <>;
-    #enum StoreCompression <none s2>;
-    #
-    #has $.nats;
-    #
-    #has Str              $!name                     is required;
-    #has Str              @!subjects                 is required;
-    #has Str              $!description;
-    #has RetentionPolicy  $!retention;
-    #has Int              $!max-consumers;
-    #has Int              $!max-msgs;
-    #has Int              $!max-bytes;
-    #has Int              $!max-age;
-    #has Int              $!max-msgs-per-subject;
-    #has Int              $!max-msg-size;
-    #has DiscardPolicy    $!discard;
-    #has StorageType      $!storage;
-    #has Int              $!num-replicas;
-    #has Bool             $!no-ack;
-    #has Str              $!template-owner;
-    #has Int              $!duplicate-window;
-    ##has Placement        $!placement;
-    #has                  %!mirror;
-    #has Associative      @!sources;
-    #has StoreCompression $!compression;
-    #has UInt             $!first-seq;
+    has Str() @.subjects;
+    has Str() $.description;
+    has Str() $.retention       = 'limits';
+    has Str() $.storage         = 'file';
+    has Str() $.discard         = 'old';
+    has Int() $.max-msgs        = -1;
+    has Int() $.max-bytes       = -1;
+    has Int() $.max-age         = 0;
+    has Int() $.max-msg-size    = -1;
+    has Int() $.max-msgs-per-subject = -1;
+    has Int() $.max-consumers   = -1;
+    has Int() $.num-replicas    = 1;
+    has Int() $.duplicate-window;
+    has Bool() $.no-ack         = False;
+    has Str() $.template-owner;
+    has Str() $.compression     = 'none';
+    has UInt() $.first-seq;
+    has        %.mirror;
+    has        @.sources;
 
     method subject(Str $template, Str $stream? --> Str) {
         sprintf $template, |(.Str with $stream)
     }
 
-    method create   { $!nats.request: $.subject(STREAM-CREATE, $!name), to-json self.&to-map }
-    method info     { $!nats.request: $.subject(STREAM-INFO, $!name)   }
-    method delete   { $!nats.request: $.subject(STREAM-DELETE, $!name) }
-    method list     { $!nats.request: $.subject(STREAM-LIST)           }
-    method consumer(Str $name, |c) { Nats::Consumer.new: |c, :$!nats, :$name, :stream($!name) }
+    method create  { $!nats.request: $.subject(STREAM-CREATE, $!name), to-json self.&to-map }
+    method update  { $!nats.request: $.subject(STREAM-UPDATE, $!name), to-json self.&to-map }
+    method info    { $!nats.request: $.subject(STREAM-INFO,   $!name) }
+    method delete  { $!nats.request: $.subject(STREAM-DELETE, $!name) }
+    method list    { $!nats.request: $.subject(STREAM-LIST)           }
+    method names   { $!nats.request: $.subject(STREAM-NAMES)          }
+    method purge   { $!nats.request: $.subject(STREAM-PURGE, $!name)  }
+
+    # Direct message get by sequence number
+    method get-msg(UInt $seq, Str :$subject) {
+        my $api-subject = $subject
+            ?? sprintf(DIRECT-GET-LAST, $!name, $subject)
+            !! sprintf(DIRECT-GET, $!name);
+        my %payload = :last_by_subj($seq);
+        $!nats.request: $api-subject, to-json %payload
+    }
+
+    # Direct get last message for a subject
+    method get-last-msg(Str $subject) {
+        $!nats.request: sprintf(DIRECT-GET-LAST, $!name, $subject), to-json { :last_by_subj($subject) }
+    }
+
+    method consumer(Str $name, |c) {
+        Nats::Consumer.new: |c, :$!nats, :$name, :stream($!name)
+    }
+
+    method consumers {
+        $!nats.request: sprintf(CONSUMER-LIST, $!name)
+    }
 }
 
 class Nats::Consumer {
     has     $.nats is required;
     has Str $.name is required;
     has Str $.stream is required;
-    has Str $.durable-name = $!name,
-    has Str $.deliver-policy  = 'all',
-    has Str $.ack-policy      = 'explicit',
-    has Str $.filter-subject,
-    has Int $.ack-wait        = 30,
-    has Int $.max-deliver     = -1,
-    has Int $.max-ack-pending = 100,
-    has Str $.replay-policy   = "instant",
-    has Int $.num-replicas    = 0,
-
-    method config(--> Map()) {
-        :stream_name($!stream),
-        :config{
+    has Str $.durable-name    = $!name;
+    has Str $.deliver-policy  = 'all';
+    has Str $.ack-policy      = 'explicit';
+    has Str $.filter-subject;
+    has Str $.deliver-subject;
+    has Str $.description;
+    has Int $.ack-wait        = 30;
+    has Int $.max-deliver     = -1;
+    has Int $.max-ack-pending = 100;
+    has Str $.replay-policy   = "instant";
+    has Int $.num-replicas    = 0;
+    has Int $.inactive-threshold;
+    has Int $.max-batch;
+    has Int $.max-expires;
+    has Int $.max-bytes;
+
+    method config(Bool :$include-durable = True --> Map()) {
+        my %cfg = %(
             :ack_policy($!ack-policy),
             :deliver_policy($!deliver-policy),
-            :durable_name($!durable-name),
-            :$!name,
-            :max_ack_pending($!max-ack-pending),
-            :max_deliver($!max-deliver),
             :replay_policy($!replay-policy),
-            :num_replicas($!num-replicas),
-        },
-        :action(""),
+        );
+        %cfg<durable_name>    = $!durable-name    if $include-durable && $!durable-name.defined;
+        %cfg<filter_subject>  = $!filter-subject  if $!filter-subject.defined;
+        %cfg<deliver_subject> = $!deliver-subject if $!deliver-subject.defined;
+        %cfg<description>     = $!description     if $!description.defined;
+        %cfg<max_ack_pending> = $!max-ack-pending if $!max-ack-pending.defined && $!max-ack-pending >= 0;
+        %cfg<max_deliver>     = $!max-deliver     if $!max-deliver.defined && $!max-deliver > 0;
+        %cfg<max_waiting>     = $!max-deliver     if $!max-deliver.defined && $!max-deliver > 0;
+        %cfg<num_replicas>    = $!num-replicas    if $!num-replicas.defined && $!num-replicas > 0;
+        %cfg<inactive_threshold> = $!inactive-threshold * 1_000_000_000
+            if $!inactive-threshold.defined && $!inactive-threshold > 0;
+        %cfg<max_batch>   = $!max-batch   if $!max-batch.defined && $!max-batch > 0;
+        %cfg<max_expires>  = $!max-expires * 1_000_000_000
+            if $!max-expires.defined && $!max-expires > 0;
+        %cfg<max_bytes>   = $!max-bytes   if $!max-bytes.defined && $!max-bytes > 0;
+        %cfg<ack_wait>    = ($!ack-wait * 1_000_000_000) if $!ack-wait.defined && $!ack-wait > 0;
+        %cfg.Map
     }
 
     method subject(Str $template, Str $stream, Str $consumer? --> Str) {
         sprintf $template, $stream, |(.Str with $consumer)
     }
 
-    method create { $!nats.request: $.subject(CONSUMER-CREATE, $!stream, $!name), to-json self.config }
-    method next   { $!nats.request: $.subject(CONSUMER-MSG-NEXT, $!stream, $!name) }
+    method create {
+        my $subject = JS-API ~ ".CONSUMER.CREATE." ~ $!stream;
+        my %req = %(
+            :stream_name($!stream),
+            :config(self.config),
+        );
+        $!nats.request: $subject, to-json %req.Map
+    }
 
-    #sub consumer-info     { $.subject(CONSUMER-INFO, $!stream)      }
-    #sub consumer-delete   { $.subject(CONSUMER-DELETE,)             }
+    method create-named {
+        my $subject = $.subject(CONSUMER-CREATE, $!stream, $!name);
+        my %req = %(
+            :stream_name($!stream),
+            :config(self.config(:!include-durable)),
+        );
+        $!nats.request: $subject, to-json %req.Map
+    }
+
+    method info {
+        $!nats.request: $.subject(CONSUMER-INFO, $!stream, $!name)
+    }
+
+    method delete {
+        $!nats.request: $.subject(CONSUMER-DELETE, $!stream, $!name)
+    }
+
+    method update {
+        my $subject = $.subject(CONSUMER-CREATE, $!stream, $!name);
+        my %req = %(
+            :stream_name($!stream),
+            :config(self.config(:!include-durable)),
+        );
+        $!nats.request: $subject, to-json %req.Map
+    }
+
+    method next(UInt :$batch, UInt :$expires, Bool :$no-wait) {
+        my %payload;
+        %payload<batch>   = $batch   if $batch && $batch > 0;
+        %payload<expires> = $expires * 1_000_000_000 if $expires;
+        %payload<no_wait> = True if $no-wait;
+        $!nats.request:
+            $.subject(CONSUMER-MSG-NEXT, $!stream, $!name),
+            to-json(%payload.elems ?? %payload !! {})
+    }
+
+    method msgs(UInt :$expires, Bool :$no-wait, UInt :$batch) {
+        my %payload;
+        %payload<no_wait> = True if $no-wait;
+        %payload<expires> = $expires * 1_000_000_000 if $expires;
+        %payload<batch>   = $batch if $batch && $batch > 0;
+
+        my $subj = $.subject: CONSUMER-MSG-NEXT, $!stream, $!name;
+
+        supply {
+            if $expires {
+                whenever Promise.in($expires) { done }
+            }
+            loop {
+                my $response = $!nats.request:
+                    $subj,
+                    to-json(%payload.elems ?? %payload !! {});
+                # Await the response; if it's a Supply, take the first emission
+                my $msg = $response ~~ Supply
+                    ?? await $response.head.Promise
+                    !! await $response;
+                # Check for JetStream 404/408/409 errors in the message
+                if $msg.payload && $msg.payload.starts-with('-ERR') {
+                    die Nats::Error.new: :message($msg.payload)
+                }
+                emit $msg;
+                CATCH {
+                    when X::AdHoc { note "JetStream fetch error: $_"; done }
+                    default       { die $_ }
+                }
+            }
+        }
+    }
+
+    # Ack helper: explicit ack to the message reply subject
+    method ack(Nats::Message $msg) {
+        return unless $msg.?reply-to;
+        $!nats.publish: $msg.reply-to, "+ACK";
+    }
+
+    # NAK: negative acknowledge
+    method nak(Nats::Message $msg) {
+        return unless $msg.?reply-to;
+        $!nats.publish: $msg.reply-to, "-NAK";
+    }
+
+    # Ack with server confirmation (double-ack / ack-sync)
+    method ack-sync(Nats::Message $msg) {
+        return unless $msg.?reply-to;
+        $!nats.request: $msg.reply-to, "+ACK";
+    }
+
+    # AckNext: request next messages on pull consumer via the message reply subject
+    method ack-next(Nats::Message $msg, UInt :$batch = 1, Bool :$no-wait) {
+        return unless $msg.?reply-to;
+        my Str $payload = $no-wait
+            ?? "+NXT " ~ to-json { :no_wait }
+            !! "+NXT " ~ $batch;
+        $!nats.publish: $msg.reply-to, $payload;
+    }
+
+    # Term: signal the server to stop redelivery
+    method term(Nats::Message $msg) {
+        return unless $msg.?reply-to;
+        $!nats.publish: $msg.reply-to, "+TERM";
+    }
 }
diff --git a/lib/Nats/JetStream/Ackable.rakumod b/lib/Nats/JetStream/Ackable.rakumod
new file mode 100644
index 0000000..6caa835
--- /dev/null
+++ b/lib/Nats/JetStream/Ackable.rakumod
@@ -0,0 +1,24 @@
+unit role Nats::JetStream::Ackable;
+
+# Basic JetStream acknowledgement helpers.
+# These publish control messages to the message reply subject.
+
+method ack() {
+    return unless $.^can('nats') && $.^can('reply-to');
+    $.nats.publish: $.reply-to, "+ACK";
+}
+
+method nak() {
+    return unless $.^can('nats') && $.^can('reply-to');
+    $.nats.publish: $.reply-to, "-NAK";
+}
+
+method in-progress() {
+    return unless $.^can('nats') && $.^can('reply-to');
+    $.nats.publish: $.reply-to, "+WPI";
+}
+
+method term() {
+    return unless $.^can('nats') && $.^can('reply-to');
+    $.nats.publish: $.reply-to, "+TERM";
+}
diff --git a/lib/Nats/Message.rakumod b/lib/Nats/Message.rakumod
index 2faff5a..b9f503c 100644
--- a/lib/Nats/Message.rakumod
+++ b/lib/Nats/Message.rakumod
@@ -1,14 +1,34 @@
 use JSON::Fast;
 use Nats::Replyable;
+use Nats::JetStream::Ackable;
 unit class Nats::Message;
 
 has Str  $.subject;
 has UInt $.sid;
 has Str  $.payload;
+has      %.headers;
 has      $.nats where { .^can('publish') }
 
-method TWEAK(Str :$reply-to) {
-    self does Nats::Replyable($reply-to) if $reply-to && self !~~ Nats::Replyable;
+method TWEAK(:$reply-to) {
+    # Add reply and JetStream ack helpers when we have a reply subject
+    if $reply-to {
+        self does Nats::Replyable($reply-to) if self !~~ Nats::Replyable;
+        self does Nats::JetStream::Ackable   if self !~~ Nats::JetStream::Ackable;
+    }
+    # Try to parse headers if payload includes NATS/1.0 header block
+    if $!payload.starts-with('NATS/1.0') {
+        my ($head, $body) = $!payload.split(/\n\n/, 2);
+        my %h;
+        for $head.lines.skip -> $line {
+            next unless $line.chars;
+            my ($k, $v) = $line.split(':', 2);
+            next unless $v.defined;
+            %h{$k.trim} //= [];
+            %h{$k.trim}.push: $v.trim;
+        }
+        %.headers = %h;
+        $!payload = $body // $!payload;
+    }
 }
 
 method json() {
diff --git a/t/headers.rakutest b/t/headers.rakutest
new file mode 100644
index 0000000..d2d72fd
--- /dev/null
+++ b/t/headers.rakutest
@@ -0,0 +1,57 @@
+#!/usr/bin/env raku
+
+use Test;
+use Test::Mock;
+
+use lib 'lib';
+
+use Nats;
+use Nats::Message;
+use Nats::Grammar;
+use Nats::Actions;
+
+use-ok 'Nats';
+
+my $printed-global;
+my $payload-global;
+
+# HPUB publish with headers should format sizes and block correctly
+{
+    my Supplier $supplier .= new;
+    my $printed;
+    my $conn         = mocked IO::Socket::Async, overriding => {
+        Supply => -> { $supplier.Supply },
+        print  => -> $s { $printed = $s },
+    };
+    my $socket-class = mocked IO::Socket::Async, returning => { connect => Promise.kept: $conn };
+    my $nats = Nats.new: :$socket-class;
+    await $nats.start;
+
+    my %headers = Bar => 'Baz';
+    my $payload = 'Hello';
+
+    $nats.publish: 'foo', $payload, :headers(%headers);
+    $printed-global = $printed;
+    $payload-global = $payload;
+
+    ok $printed ~~ /^ 'HPUB ' /, 'printed HPUB control line';
+
+    # compute expected header block and sizes with same formatting
+    like $printed, /^ 'HPUB foo ' \d+ ' ' \d+ "\r\n" 'NATS/1.0' "\r\n" 'Bar: Baz' "\r\n\r\n" 'Hello' "\r\n" ("\r\n"?) $/, 'HPUB matches regex formatting';
+}
+
+# HMSG parsing should split headers and payload (via direct Message construction)
+{
+    my @lines = ('NATS/1.0', 'Bar: Baz');
+    my $headers-block = @lines.join("\r\n") ~ "\r\n\r\n";
+    my $msg = Nats::Message.new(
+        subject => 'foo',
+        sid     => 1,
+        payload => $headers-block ~ 'Hello',
+    );
+    isa-ok $msg, Nats::Message, 'constructed Nats::Message';
+    is-deeply $msg.headers<Bar>, ['Baz'], 'parsed header Bar: Baz';
+    is $msg.payload, 'Hello', 'parsed payload';
+}
+
+done-testing;
diff --git a/t/hmsg.rakutest b/t/hmsg.rakutest
new file mode 100644
index 0000000..5e4bb76
--- /dev/null
+++ b/t/hmsg.rakutest
@@ -0,0 +1,47 @@
+#!/usr/bin/env raku
+
+use Test;
+use Test::Mock;
+
+use lib 'lib';
+
+use Nats;
+use Nats::Message;
+
+use-ok 'Nats';
+
+{
+    my Supplier $supplier .= new;
+    my $conn         = mocked IO::Socket::Async, returning => { Supply => $supplier.Supply };
+    my $socket-class = mocked IO::Socket::Async, returning => { connect => Promise.kept: $conn };
+
+    my $nats = Nats.new: :$socket-class;
+    await $nats.start;
+
+    my $subject = 'js.test';
+    my $sid     = 42;
+
+    # Build NATS headers block with body using \n line endings
+    my @lines = ('NATS/1.0', 'Content-Type: text/plain', 'X-Req: abc');
+    my $hdrs  = @lines.join("\n") ~ "\n\n";
+    my $body1 = 'hello';
+    my $body2 = 'world';
+    my $hsize = $hdrs.encode('utf8').bytes;
+    my $tsize1 = $hsize + $body1.encode('utf8').bytes;
+    my $tsize2 = $hsize + $body2.encode('utf8').bytes;
+
+    my $frame1 = "HMSG $subject $sid $hsize $tsize1\n$hdrs$body1\n";
+    my $frame2 = "HMSG $subject $sid $hsize $tsize2\n$hdrs$body2\n";
+
+    my @msgs;
+    $nats.supply.tap: -> $m { @msgs.push: $m if $m ~~ Nats::Message };
+
+    $supplier.emit: $frame1;
+    $supplier.emit: $frame2;
+
+    is @msgs.elems, 2, 'received two HMSG messages';
+    is @msgs[0].payload, $body1, 'first HMSG body extracted' if @msgs.elems == 2;
+    is @msgs[1].payload, $body2, 'second HMSG body extracted' if @msgs.elems == 2;
+}
+
+done-testing;
diff --git a/t/jetstream.rakutest b/t/jetstream.rakutest
new file mode 100644
index 0000000..c161d94
--- /dev/null
+++ b/t/jetstream.rakutest
@@ -0,0 +1,403 @@
+#!/usr/bin/env raku
+
+use Test;
+use Test::Mock;
+use JSON::Fast;
+
+use lib 'lib';
+
+use Nats::JetStream;
+use Nats;
+
+use-ok 'Nats::JetStream';
+
+# Stream create
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.STREAM.CREATE.MY', 'stream create subject';
+            my %cfg = from-json($payload);
+            is %cfg<name>, 'MY', 'stream config name';
+            is %cfg<subjects>[0], 'foo', 'stream config subjects';
+            is %cfg<retention>, 'limits', 'stream config retention';
+            is %cfg<storage>, 'file', 'stream config storage';
+        }
+    };
+
+    my $stream = Nats::Stream.new(
+        :nats($nats),
+        name      => 'MY',
+        subjects  => ['foo'],
+        retention => 'limits',
+        storage   => 'file',
+        max-msgs  => -1,
+        max-bytes => -1,
+        max-age   => 0,
+    );
+
+    $stream.create;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream update
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.STREAM.UPDATE.MY', 'stream update subject';
+            my %cfg = from-json($payload);
+            is %cfg<name>, 'MY', 'stream update config name';
+            is %cfg<subjects>[0], 'bar', 'stream update config subjects';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY', subjects => ['bar']).update;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream info
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.STREAM.INFO.MY', 'stream info subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').info;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream delete
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.STREAM.DELETE.MY', 'stream delete subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').delete;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream list
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.STREAM.LIST', 'stream list subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'ANY').list;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream names
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.STREAM.NAMES', 'stream names subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'ANY').names;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream consumers list
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.CONSUMER.LIST.MY', 'consumer list subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').consumers;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream with extended attributes
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            my %cfg = from-json($payload);
+            is %cfg<discard>, 'new', 'stream config discard';
+            is %cfg<max_msg_size>, 1048576, 'stream config max-msg-size';
+            is %cfg<max_msgs_per_subject>, 1000, 'stream config max-msgs-per-subject';
+            is %cfg<max_consumers>, 10, 'stream config max-consumers';
+            is %cfg<duplicate_window>, 120, 'stream config duplicate-window';
+            is %cfg<compression>, 's2', 'stream config compression';
+        }
+    };
+
+    Nats::Stream.new(
+        :nats($nats),
+        name               => 'EXT',
+        discard            => 'new',
+        max-msg-size       => 1048576,
+        max-msgs-per-subject => 1000,
+        max-consumers      => 10,
+        duplicate-window   => 120,
+        compression        => 's2',
+    ).create;
+}
+
+# Consumer create (stream-only endpoint with config payload)
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.CONSUMER.CREATE.MY', 'consumer create subject';
+            my %req = from-json($payload);
+            my %cfg = %req<config>;
+            is %cfg<ack_policy>, 'explicit', 'consumer ack policy';
+            is %cfg<deliver_policy>, 'all', 'consumer deliver policy';
+            is %cfg<durable_name>, 'C', 'consumer durable name';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    $c.create;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer create named (subject includes durable, payload omits durable_name)
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.CONSUMER.CREATE.MY.C', 'consumer create named subject';
+            my %req = from-json($payload);
+            is %req<stream_name>, 'MY', 'stream_name present';
+            my %cfg = %req<config>;
+            ok !(%cfg<durable_name>:exists), 'durable_name omitted in payload';
+            is %cfg<ack_policy>, 'explicit', 'consumer ack policy';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    $c.create-named;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer with extended config
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            my %req = from-json($payload);
+            my %cfg = %req<config>;
+            is %cfg<inactive_threshold>, 60_000_000_000, 'consumer inactive threshold in nanos';
+            is %cfg<max_batch>, 256, 'consumer max batch';
+            is %cfg<max_bytes>, 1048576, 'consumer max bytes';
+            is %cfg<description>, 'my-consumer', 'consumer description';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C',
+        :inactive-threshold(60), :max-batch(256), :max-bytes(1048576), :description('my-consumer')
+    );
+    $c.create;
+}
+
+# Consumer info
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.CONSUMER.INFO.MY.C', 'consumer info subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').consumer('C').info;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer delete
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.CONSUMER.DELETE.MY.C', 'consumer delete subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').consumer('C').delete;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer next
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, |c {
+            is $subject, '$JS.API.CONSUMER.MSG.NEXT.MY.C', 'consumer next subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').consumer('C').next;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer next with payload options
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload, |c {
+            is $subject, '$JS.API.CONSUMER.MSG.NEXT.MY.C', 'consumer next subject with payload';
+            my %p = from-json($payload);
+            is %p<batch>, 5, 'next payload batch';
+            is %p<expires>, 2_000_000_000, 'next payload expires nanos';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').consumer('C').next(:batch(5), :expires(2));
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer ack helper
+{
+    my $nats = mocked Nats, overriding => {
+        publish => -> $subject, $payload {
+            is $subject, '$JS.ACK.MY.C', 'ack subject';
+            is $payload, '+ACK', 'ack payload';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    my $msg = Nats::Message.new(
+        subject  => 's',
+        sid      => 1,
+        payload  => 'p',
+        reply-to => '$JS.ACK.MY.C',
+        nats     => $nats,
+    );
+    $c.ack($msg);
+    check-mock $nats, *.called('publish', :once);
+}
+
+# Consumer NAK
+{
+    my $nats = mocked Nats, overriding => {
+        publish => -> $subject, $payload {
+            is $subject, '$JS.ACK.MY.C', 'nak subject';
+            is $payload, '-NAK', 'nak payload';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    my $msg = Nats::Message.new(
+        subject  => 's', sid => 1, payload => 'p',
+        reply-to => '$JS.ACK.MY.C', nats => $nats,
+    );
+    $c.nak($msg);
+    check-mock $nats, *.called('publish', :once);
+}
+
+# Consumer term
+{
+    my $nats = mocked Nats, overriding => {
+        publish => -> $subject, $payload {
+            is $subject, '$JS.ACK.MY.C', 'term subject';
+            is $payload, '+TERM', 'term payload';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    my $msg = Nats::Message.new(
+        subject  => 's', sid => 1, payload => 'p',
+        reply-to => '$JS.ACK.MY.C', nats => $nats,
+    );
+    $c.term($msg);
+    check-mock $nats, *.called('publish', :once);
+}
+
+# Consumer ack-sync
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.ACK.MY.C', 'ack-sync subject';
+            is $payload, '+ACK', 'ack-sync payload';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    my $msg = Nats::Message.new(
+        subject  => 's', sid => 1, payload => 'p',
+        reply-to => '$JS.ACK.MY.C', nats => $nats,
+    );
+    $c.ack-sync($msg);
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer ack-next with batch
+{
+    my $nats = mocked Nats, overriding => {
+        publish => -> $subject, $payload {
+            is $subject, '$JS.ACK.MY.C', 'ack-next batch subject';
+            is $payload, '+NXT 5', 'ack-next batch payload';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    my $msg = Nats::Message.new(
+        subject  => 's', sid => 1, payload => 'p',
+        reply-to => '$JS.ACK.MY.C', nats => $nats,
+    );
+    $c.ack-next($msg, :batch(5));
+    check-mock $nats, *.called('publish', :once);
+}
+
+# Consumer ack-next with no-wait
+{
+    my $nats = mocked Nats, overriding => {
+        publish => -> $subject, $payload {
+            is $subject, '$JS.ACK.MY.C', 'ack-next no-wait subject';
+            like $payload, /'no_wait'/, 'ack-next no-wait payload has no_wait';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    my $msg = Nats::Message.new(
+        subject  => 's', sid => 1, payload => 'p',
+        reply-to => '$JS.ACK.MY.C', nats => $nats,
+    );
+    $c.ack-next($msg, :no-wait);
+    check-mock $nats, *.called('publish', :once);
+}
+
+# from-map helper
+{
+    my $stream = Nats::Stream.new(name => 'TEST', nats => Nil);
+    from-map({ name => 'UPDATED', subjects => ['x', 'y'], max_msgs => 1000 }, $stream);
+    is $stream.name, 'UPDATED', 'from-map updates name';
+    is $stream.subjects[0], 'x', 'from-map updates subjects';
+    is $stream.max-msgs, 1000, 'from-map updates max-msgs';
+}
+
+# Stream purge
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject {
+            is $subject, '$JS.API.STREAM.PURGE.MY', 'stream purge subject';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').purge;
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream direct get by sequence
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.DIRECT.GET.MY', 'direct get subject';
+            my %p = from-json($payload);
+            is %p<last_by_subj>, 5, 'sequence number in payload';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').get-msg(5);
+    check-mock $nats, *.called('request', :once);
+}
+
+# Stream direct get last per subject
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.DIRECT.GET.MY.foo', 'direct get last subject';
+            my %p = from-json($payload);
+            is %p<last_by_subj>, 'foo', 'subject in payload';
+        }
+    };
+    Nats::Stream.new(:nats($nats), name => 'MY').get-last-msg('foo');
+    check-mock $nats, *.called('request', :once);
+}
+
+# Consumer update
+{
+    my $nats = mocked Nats, overriding => {
+        request => -> $subject, $payload {
+            is $subject, '$JS.API.CONSUMER.CREATE.MY.C', 'consumer update subject';
+            my %req = from-json($payload);
+            is %req<stream_name>, 'MY', 'stream_name present';
+            my %cfg = %req<config>;
+            ok !(%cfg<durable_name>:exists), 'durable_name omitted in update';
+        }
+    };
+    my $c = Nats::Stream.new(:nats($nats), name => 'MY').consumer('C');
+    $c.update;
+    check-mock $nats, *.called('request', :once);
+}
+
+done-testing;
diff --git a/t/nats.rakutest b/t/nats.rakutest
index 745590e..52fb135 100644
--- a/t/nats.rakutest
+++ b/t/nats.rakutest
@@ -60,7 +60,7 @@ $nats.publish:   "foo", "hello world";
 $nats.publish:   "bar", "hello world", :reply-to<qux>;
 
 check-mock $conn,
-    *.called("print", :once, with => :("CONNECT \{}\r\n")),
+    *.called("print", :once, with => :("CONNECT \{\"headers\":true\}\r\n")),
     *.called("print", :once, with => :("PING\r\n")),
     *.called("print", :once, with => :("SUB foo 0\r\n")),
     *.called("print", :once, with => :("SUB bar baz 1\r\n")),
diff --git a/t/pull-consumer.rakutest b/t/pull-consumer.rakutest
new file mode 100644
index 0000000..f46c99f
--- /dev/null
+++ b/t/pull-consumer.rakutest
@@ -0,0 +1,47 @@
+#!/usr/bin/env raku
+
+use Test;
+use Test::Mock;
+
+use lib 'lib';
+
+use Nats;
+use Nats::JetStream;
+
+use-ok 'Nats';
+
+# This test simulates producer and pull-consumer interaction.
+# It feeds a realistic sequence of frames into the mocked socket Supply
+# including partial chunks to exercise the input buffer reassembly.
+
+{
+    my Supplier $supplier .= new;
+    my $conn         = mocked IO::Socket::Async, returning => { Supply => $supplier.Supply };
+    my $socket-class = mocked IO::Socket::Async, returning => { connect => Promise.kept: $conn };
+
+    my $nats = Nats.new: :$socket-class;
+    await $nats.start;
+
+    # Prepare two MSG frames as NATS server would send to the consumer inbox
+    my $inbox = '_INBOX.pulltest________________________';
+    my $payload1 = 'msg 1';
+    my $payload2 = 'msg 2';
+    my $frame1 = "MSG $inbox 0 " ~ $payload1.chars ~ "\r\n$payload1\r\n";
+    my $frame2 = "MSG $inbox 0 " ~ $payload2.chars ~ "\r\n$payload2\r\n";
+
+    my @received;
+    $nats.supply.tap: -> $msg {
+        @received.push: $msg if $msg ~~ Nats::Message;
+    };
+
+    # Emit both frames
+    $supplier.emit: $frame1;
+    $supplier.emit: $frame2;
+
+    sleep 0.01;
+    is @received.elems, 2, 'consumer received two messages';
+    is @received[0].payload, $payload1, 'first payload intact';
+    is @received[1].payload, $payload2, 'second payload intact';
+}
+
+done-testing;
diff --git a/t/split-msg.rakutest b/t/split-msg.rakutest
new file mode 100644
index 0000000..b62c58f
--- /dev/null
+++ b/t/split-msg.rakutest
@@ -0,0 +1,61 @@
+#!/usr/bin/env raku
+
+use Test;
+use Test::Mock;
+
+use lib 'lib';
+
+use Nats;
+use Nats::Message;
+
+use-ok 'Nats';
+
+# Test frame reassembly: when TCP splits a frame across chunks,
+# the parser should buffer and reassemble before parsing.
+{
+    my Supplier $supplier .= new;
+    my $conn         = mocked IO::Socket::Async, returning => { Supply => $supplier.Supply };
+    my $socket-class = mocked IO::Socket::Async, returning => { connect => Promise.kept: $conn };
+
+    my $nats = Nats.new: :$socket-class;
+    await $nats.start;
+
+    my @msgs;
+    $nats.supply.tap: -> $m { @msgs.push: $m if $m ~~ Nats::Message };
+
+    # Split a MSG frame across two chunks
+    my $hdr  = "MSG foo 1 11\r\nhello ";
+    my $tail = "world\r\n";
+
+    $supplier.emit: $hdr;
+    $supplier.emit: $tail;
+    sleep 0.01;
+
+    is @msgs.elems, 1, 'parsed one split MSG frame';
+    is @msgs[0].subject, 'foo', 'subject parsed';
+    is @msgs[0].sid, 1, 'sid parsed';
+    is @msgs[0].payload, 'hello world', 'payload reconstructed correctly';
+}
+
+# Test multiple complete frames in one chunk
+{
+    my Supplier $supplier .= new;
+    my $conn         = mocked IO::Socket::Async, returning => { Supply => $supplier.Supply };
+    my $socket-class = mocked IO::Socket::Async, returning => { connect => Promise.kept: $conn };
+
+    my $nats = Nats.new: :$socket-class;
+    await $nats.start;
+
+    my @msgs;
+    $nats.supply.tap: -> $m { @msgs.push: $m if $m ~~ Nats::Message };
+
+    my $chunk = "MSG a 1 3\r\nabc\r\nMSG b 2 3\r\ndef\r\n";
+    $supplier.emit: $chunk;
+    sleep 0.01;
+
+    is @msgs.elems, 2, 'parsed two frames in one chunk';
+    is @msgs[0].subject, 'a', 'first subject';
+    is @msgs[1].subject, 'b', 'second subject';
+}
+
+done-testing;
diff --git a/t/utf8-publish.rakutest b/t/utf8-publish.rakutest
new file mode 100644
index 0000000..c8922a1
--- /dev/null
+++ b/t/utf8-publish.rakutest
@@ -0,0 +1,39 @@
+#!/usr/bin/env raku
+
+use Test;
+use Test::Mock;
+
+use lib 'lib';
+
+use Nats;
+
+# ── Setup: mock socket (same pattern as t/nats.rakutest) ─────────
+my Supplier $supplier .= new;
+my $conn         = mocked IO::Socket::Async, returning => { Supply => $supplier.Supply };
+my $socket-class = mocked IO::Socket::Async, returning => { connect => Promise.kept: $conn };
+
+my $nats = Nats.new: :$socket-class;
+$nats.start;
+
+# ── Test 1: ASCII-only — chars == bytes ──────────────────────────
+# "hello" = 5 chars = 5 bytes
+$nats.publish: "foo", "hello";
+
+# ── Test 2: UTF-8 multi-byte — chars != bytes ────────────────────
+# "coração" = 7 chars, 9 bytes (ç=2, ã=2 in UTF-8)
+# BEFORE fix: !pub used .chars → PUB with 7 → NATS reads 7 of 9 bytes
+#   → 2 orphan bytes parsed as new command → 'Unknown Protocol Operation'
+# AFTER fix:  !pub uses .encode('utf8').bytes → PUB with 9 → correct
+$nats.publish: "bar", "coração";
+
+$nats.stop;
+
+# ── Verify PUB wire format uses byte counts, not char counts ─────
+# If the fix is correct, "coração" (7 chars, 9 bytes) → PUB uses 9
+# If the bug is present, "coração" (7 chars) → PUB uses 7 (WRONG)
+check-mock $conn,
+    *.called("print", :once, with => :("PUB foo 5\r\nhello\r\n")),
+    *.called("print", :once, with => :("PUB bar 9\r\ncoração\r\n")),
+;
+
+done-testing;