view release on metacpan or  search on metacpan

lib/Claude/Agent/Query.pm  view on Meta::CPAN

                while ($$buffref =~ s/^([^\n]+)\n//) {
                    my $line = $1;
                    $self->_handle_line($line);
                }
                return 0;
            },
        },
        stderr => {
            on_read => sub {
                my ($stream, $buffref) = @_;
                # Log stderr but don't treat as fatal
                while ($$buffref =~ s/^([^\n]+)\n//) {
                    $log->debug(sprintf("Claude CLI stderr: %s", $1));
                }
                return 0;
            },
        },
        on_finish => sub {
            my ($proc, $exitcode) = @_;
            $self->_finished(1);
            # Extract actual exit status (WEXITSTATUS equivalent)
            my $exit_status = $exitcode >> 8;
            $log->debug(sprintf("Query: Process finished with exit code %d", $exit_status));
            if ($exit_status != 0) {
                $self->_error("Claude CLI exited with code $exit_status");
            }
            # Resolve any pending async futures
            $self->_resolve_pending_futures_on_finish();
        },
        on_exception => sub {
            my ($proc, $exception, $errno, $exitcode) = @_;
            $self->_finished(1);
            $log->debug(sprintf("Query: Process exception: %s", $exception));
            $self->_error("Claude CLI exception: $exception");
            # Resolve any pending async futures
            $self->_resolve_pending_futures_on_finish();
        },
    );

    # Store references before adding to loop to avoid race conditions
    $self->_process($process);
    $self->_stdin($process->stdin);
    $self->_loop->add($process);

    # For --print mode, prompt is in argv, stdin can be closed
    if (!ref($self->prompt)) {
        $self->_loop->later(sub {
            $self->_stdin->close_when_empty if $self->_stdin;
        });
    }
    # For ref prompts (streaming input), caller will send messages via send_user_message
    return;
}

sub _handle_line {
    my ($self, $line) = @_;

    return unless defined $line && length $line;

    # Use JSON::Lines decode method for single line
    my @decoded = $self->_jsonl->decode($line);
    if ($log->is_trace) {
        $log->trace(sprintf("Raw line length: %d", length($line)));
        $log->trace(sprintf("Raw line: %s", $line));
        $log->trace(sprintf("Decoded %d objects", scalar(@decoded)));
        $log->trace(sprintf("JSON::Lines buffer remaining: %d chars", length($self->_jsonl->remaining)));
        if ($self->_jsonl->remaining) {
            $log->trace(sprintf("Buffer content (first 200): %s", substr($self->_jsonl->remaining, 0, 200)));
        }
    }

    # Guard against buffer overflow from accumulated malformed data
    # Configurable threshold via environment variable, default 1MB, max 10MB
    # Note: Typical JSON messages from Claude CLI are 1-50KB; tool results (especially
    # large file reads or grep outputs) may be 100KB-1MB. The minimum of 100KB ensures
    # legitimate large tool results are not prematurely truncated.
    # Set CLAUDE_AGENT_JSONL_BUFFER_MAX to at least 2x your largest expected message size.
    my $buffer_threshold = $ENV{CLAUDE_AGENT_JSONL_BUFFER_MAX} // 1_000_000;
    my $min_threshold = 100_000;  # Minimum 100KB to accommodate large tool results
    my $max_threshold = 10_000_000;  # Hard cap at 10MB
    if (!defined $buffer_threshold || $buffer_threshold !~ /^\d+$/) {
        if (defined $ENV{CLAUDE_AGENT_JSONL_BUFFER_MAX}) {
            # Sanitize env var value to prevent log injection (remove control characters)
            my $safe_val = $ENV{CLAUDE_AGENT_JSONL_BUFFER_MAX} // '';
            $safe_val =~ s/[[:cntrl:]]//g;
            $log->warning(sprintf("CLAUDE_AGENT_JSONL_BUFFER_MAX='%s' is not a valid integer, using default 1000000", $safe_val));
        }
        $buffer_threshold = 1_000_000;
    }
    elsif ($buffer_threshold < $min_threshold) {
        $log->warning(sprintf("CLAUDE_AGENT_JSONL_BUFFER_MAX=%d is below minimum (%d), using %d. "
            . "Very small values may truncate legitimate large tool results.",
            $buffer_threshold, $min_threshold, $min_threshold));
        $buffer_threshold = $min_threshold;
    }
    elsif ($buffer_threshold > $max_threshold) {
        $log->debug(sprintf("CLAUDE_AGENT_JSONL_BUFFER_MAX=%d exceeds maximum (%d), using %d",
            $buffer_threshold, $max_threshold, $max_threshold));
        $buffer_threshold = $max_threshold;
    }

    # Check buffer size regardless of decode success to prevent unbounded growth
    if ($self->_jsonl->remaining && length($self->_jsonl->remaining) > $buffer_threshold) {
        $log->debug(sprintf("JSON::Lines buffer overflow detected (size: %d, threshold: %d), reinitializing",
            length($self->_jsonl->remaining), $buffer_threshold));
        $self->_jsonl(JSON::Lines->new(
            utf8     => 1,
            error_cb => sub {
                my ($action, $error, $data) = @_;
                $log->trace(sprintf("JSON::Lines %s error: %s", $action, $error));
                return;
            },
        ));
    }

    return unless @decoded;

    for my $data (@decoded) {
        if ($log->is_trace) {
            $log->trace(sprintf("Decoded item ref type: %s", ref($data) // "not a ref"));
            if (ref $data eq 'HASH') {
                $log->trace(sprintf("Hash keys: %s", join(", ", keys %$data)));
            }
        }
        next unless defined $data && ref $data eq 'HASH';
        $log->trace(sprintf("Message type in data: %s", $data->{type} // "undef"))
            if $log->is_trace;
        next unless exists $data->{type};  # Skip malformed/partial JSON data

        $log->debug(sprintf("Query: Received message type=%s", $data->{type}));

        my $msg = Claude::Agent::Message->from_json($data);

        # Log message subtype for more detail
        if ($msg->can('subtype') && $msg->subtype) {
            $log->debug(sprintf("Query: Message subtype=%s", $msg->subtype));
        }

        # Capture session_id from init message
        if ($msg->isa('Claude::Agent::Message::System')
            && $msg->subtype eq 'init') {
            $self->_session_id($msg->get_session_id);
            $log->debug(sprintf("Query: Session ID captured: %s", $self->_session_id // 'none'));
            # Update hook executor with session_id
            if ($self->_hook_executor) {
                $self->_hook_executor->session_id($self->_session_id);
            }
        }

        # Execute Perl hooks for tool use messages
        $msg = $self->_execute_hooks_for_message($msg);

        # Skip if message was blocked by hooks
        if (!defined $msg) {
            $log->debug(sprintf("Query: Message type=%s blocked by hooks", $data->{type}));
            next;
        }

        # Re-entrancy guard: if already processing, queue for later
        if ($self->_processing_message) {
            push @{$self->_messages}, $msg;
            next;
        }
        $self->_processing_message(1);

        if (@{$self->_pending_futures}) {
            my $future = shift @{$self->_pending_futures};
            $log->debug(sprintf("Query: Delivering message type=%s to waiting future", $data->{type}));
            $future->done($msg);
        }
        else {
            $log->debug(sprintf("Query: Queuing message type=%s (queue size=%d)", $data->{type}, scalar(@{$self->_messages}) + 1));
            push @{$self->_messages}, $msg;
        }

        $self->_processing_message(0);
    }
    return;