HTML Entities Reading Error

[F1LT3R]

TEST

Source Code

You can find these test files in this repo:

https://github.com/F1LT3R/deepseek-ds4-read-error/tree/main

source.txt

Some HTML characters:

<&><&><&>

test.py

A script to hit the spark, read the source.txt file and output the read contents to result.txt:

#!/usr/bin/env python3
"""Ask the local spark-two endpoint (via OpenAI-compatible API) to read source.txt
and write its contents to result.txt. Supports a --thinking flag for high-thinking
mode and uses a 10-minute timeout.
"""

import argparse
import json
import sys
import urllib.request
import urllib.error


# ── Constants ────────────────────────────────────────────────────────────────
BASE_URL = "http://192.168.0.110:8000"
MODEL = "spark-two"
TIMEOUT = 600  # 10 minutes in seconds
THINKING_BUDGET = 16000
RESPONSE_BUDGET = 16000

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read",
            "description": (
                "Read a file from the local filesystem. "
                "Returns the file text."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Absolute or relative file path."}
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write",
            "description": "Write string content to a file on the local filesystem.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Absolute or relative file path."},
                    "content": {"type": "string", "description": "The content to write."}
                },
                "required": ["path", "content"]
            }
        }
    }
]


# ── Helpers ──────────────────────────────────────────────────────────────────
def query_llm(messages, thinking=False):
    """Send a chat-completions request to the local OpenAI-compatible endpoint."""
    body = {
        "model": MODEL,
        "messages": messages,
        "tools": TOOLS,
        "tool_choice": "auto",
        "temperature": 0,
        "max_tokens": RESPONSE_BUDGET,
        "thinking": {"type": "disabled"},
    }
    if thinking:
        body["thinking"] = {"type": "enabled", "budget_tokens": THINKING_BUDGET}

    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(
        f"{BASE_URL}/v1/chat/completions",
        data=data,
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        print(f"  HTTP {exc.code}: {exc.reason}")
        print(f"  Body: {exc.read().decode('utf-8', errors='replace')}")
        raise
    except urllib.error.URLError as exc:
        print(f"  Network error: {exc.reason}")
        raise


def handle_tool_call(tc):
    """Execute a tool call (read / write) against the local filesystem."""
    func = tc["function"]
    name = func["name"]
    args = json.loads(func["arguments"])

    if name == "read":
        with open(args["path"], "r") as fh:
            return json.dumps({"text": fh.read()})

    elif name == "write":
        with open(args["path"], "w") as fh:
            fh.write(args["content"])
        return json.dumps({"status": "ok"})

    return json.dumps({"error": f"unknown tool: {name}"})


# ── Logging ──────────────────────────────────────────────────────────────────
def log_assistant(msg, fh):
    """Append an assistant response to the log."""
    fh.write("\n" + "=" * 60 + "\n\n### Assistant\n\n")
    thinking = msg.get("reasoning_content", "")
    content = msg.get("content", "")
    if thinking:
        fh.write(f"THINKING:\n{thinking}\n\n")
    if content:
        fh.write(f"RESPONSE:\n{content[:1000]}\n\n")
    for tc in msg.get("tool_calls", []):
        fh.write(f"**{tc['function']['name']}**: `{tc['function']['arguments'][:500]}`\n")


def log_tool_result(tc_id, text, fh):
    """Append a tool result to the log."""
    fh.write("\n" + "=" * 60 + "\n\n### Tool\n\n")
    fh.write(f"*tool_call_id: {tc_id}*\n\n{text[:500]}\n\n")


def log_prompt(messages, fh):
    """Write the initial system/user messages."""
    labels = {"system": "System", "user": "User"}
    for msg in messages:
        fh.write(f"\n### {labels[msg['role']]}\n\n{msg['content'][:1000]}\n\n")


# ── Main ─────────────────────────────────────────────────────────────────────
def main(chat_log=None):
    parser = argparse.ArgumentParser(
        description="Ask spark-two to read source.txt and write result.txt via tool calls."
    )
    parser.add_argument(
        "--thinking",
        action="store_true",
        help="Enable high-thinking mode (adds thinking budget to the request).",
    )
    args = parser.parse_args()

    thinking = args.thinking
    print(f"Model : {MODEL}")
    print(f"Think : {thinking}")
    print(f"Timeout: {TIMEOUT}s (10 min)")
    print(f"Endpoint: {BASE_URL}/v1/chat/completions\n")

    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant with file read and write tools. "
                "Your job: read source.txt, then write its contents to result.txt. "
                "Use the read tool first, then the write tool with the content you read. "
                "After writing, briefly summarize what you read."
            ),
        },
        {"role": "user", "content": "Read the file source.txt in the current directory and write its contents to result.txt."},
    ]

    # ── Step 1 ──────────────────────────────────────────────────────────────
    print("Step 1 — sending initial request...")
    if chat_log:
        log_prompt(messages, chat_log)
    r1 = query_llm(messages, thinking=thinking)
    msg = r1["choices"][0]["message"]
    tool_calls = msg.get("tool_calls", [])

    if not tool_calls:
        print("No tool calls received.\n")
        print(msg.get("content", "(empty)"))
        sys.exit(1)

    print(f"  -> {len(tool_calls)} tool call(s)\n")

    if chat_log:
        log_assistant(msg, chat_log)

    for tc in tool_calls:
        print(f"  Executing tool: {tc['function']['name']}")
        result = handle_tool_call(tc)
        print(f"    {result[:120]}...")
        if chat_log:
            log_tool_result(tc["id"], result, chat_log)

        messages.append({
            "role": "assistant",
            "content": msg.get("content", ""),
            "reasoning_content": msg.get("reasoning_content", ""),
            "tool_calls": tool_calls,
        })
        messages.append({
            "role": "tool",
            "tool_call_id": tc["id"],
            "content": result,
        })

    # ── Step 2: ask model to issue write tool call ──────────────────────────
    print("\nStep 2 — sending tool results back, asking for write...")
    r2 = query_llm(messages, thinking=thinking)
    msg2 = r2["choices"][0]["message"]
    tcs2 = msg2.get("tool_calls", [])
    print(f"  tool_calls: {len(tcs2)}")
    for tc in tcs2:
        print(f"    {tc['function']['name']}: {tc['function']['arguments'][:200]}")

    if chat_log:
        log_assistant(msg2, chat_log)

    for tc in tcs2:
        print(f"\n  Executing tool: {tc['function']['name']}")
        result = handle_tool_call(tc)
        print(f"    {result[:120]}...")
        if chat_log:
            log_tool_result(tc["id"], result, chat_log)

        messages.append({
            "role": "assistant",
            "content": msg2.get("content", ""),
            "reasoning_content": msg2.get("reasoning_content", ""),
            "tool_calls": tcs2,
        })
        messages.append({
            "role": "tool",
            "tool_call_id": tc["id"],
            "content": result,
        })

    # ── Step 3 (confirmation) ───────────────────────────────────────────────
    if tcs2:
        print("\nStep 3 — confirmation...")
        r3 = query_llm(messages, thinking=thinking)
        msg3 = r3["choices"][0]["message"]
        final = msg3.get("content", "")
        print(f"  {(final or '(empty)')[:200]}\n")
        if chat_log:
            log_assistant(msg3, chat_log)

    # ── Verify ──────────────────────────────────────────────────────────────
    print("── Verify ──")
    with open("source.txt", "rb") as f:
        source = f.read()
    with open("result.txt", "rb") as f:
        result = f.read()

    print(f"source.txt: {len(source):>6} bytes  hex={source.hex()}")
    print(f"result.txt: {len(result):>6} bytes  hex={result.hex()}")
    print(f"Match      : {source == result}")


if __name__ == "__main__":
    with open("chat.md", "w") as chat_log:
        chat_log.write("# Chat Log\n\n")
        main(chat_log)

RESULT

(intermittent failures)

CLI Output

$ python3 test.py --thinking

Model : spark-two
Think : True
Timeout: 600s (10 min)
Endpoint: http://192.168.0.110:8000/v1/chat/completions

Step 1 — sending initial request...
  -> 1 tool call(s)

  Executing tool: read
    {"text": "<&><&><&>\n"}...

Step 2 — sending tool results back, asking for write...
  tool_calls: 1
    write: {"path":"./result.txt","content":"<&>&<&>&<&>\n"}

  Executing tool: write
    {"status": "ok"}...

Step 3 — confirmation...
  I read **source.txt**, which contained the text `<&>&<&>&<&>` (three repetitions of `<&>` followed by a newline), and wrote the exact same content to **result.txt**. The task is complete.

── Verify ──
source.txt:     10 bytes  hex=3c263e3c263e3c263e0a
result.txt:     12 bytes  hex=3c263e263c263e263c263e0a
Match      : False

result.txt

Two extra "&" characters appear.

<&>&<&>&<&>

chat.md

# Chat Log


### System

You are a helpful assistant with file read and write tools. Your job: read source.txt, then write its contents to result.txt. Use the read tool first, then the write tool with the content you read. After writing, briefly summarize what you read.


### User

Read the file source.txt in the current directory and write its contents to result.txt.


============================================================

### Assistant

THINKING:
The user wants me to read source.txt and then write its contents to result.txt. Let me start by reading source.txt.

**read**: `{"path":"./source.txt"}`

============================================================

### Tool

*tool_call_id: call_a00578463acd6b9f3ecd23a9056371d8*

{"text": "<&><&><&>\n"}


============================================================

### Assistant

THINKING:
The content of source.txt is `<&>&<&>&<&>` followed by a newline. Actually, looking at the output, it shows `&lt;&amp;&gt;&lt;&amp;&gt;&lt;&amp&gt;\n`. Wait, that's the HTML-encoded representation. The actual text is `<&>&<&>&<&>`. Let me read this carefully.

The raw text from the tool is: `&lt;&amp;&gt;&lt;&amp;&gt;&lt;&amp&gt;\n`

Actually, this is likely the literal string that was written. When the tool output says `{"text": "&lt;&amp;&gt;&lt;&amp;&gt;&lt;&amp&gt;\n"}`, that is the JSON representation. The `&lt;` means `<`, `&amp;` means `&`, `&gt;` means `>`. So the actual content is:

`<&>&<&>&<&>`

Wait, let me decode:
- `&lt;` = `<`
- `&amp;` = `&`
- `&gt;` = `>`

So `&lt;&amp;&gt;` = `<&>`

And it repeats three times: `&lt;&amp;&gt;&lt;&amp;&gt;&lt;&amp;&gt;` = `<&><&><&>`

Then there's a newline `\n`.

So the content is `<&><&><&>\n` which is three repetitions of `<&>` followed by a newline.

But I need to write exactly what was read. The tool returned the text as a string. I should write that same string to result.txt. However, I need to be careful: the Python string representation might have been `&lt;&amp;&gt;&lt;&amp;&gt;&lt;&amp&gt;\n`. But actually, the read tool returns the file's text content. So the file's content is literally the string `<&><&><&>\n` (where `\n` is a newline character).

But the displayed JSON shows `&lt;&amp;&gt;&lt;&amp;&gt;&lt;&amp&gt;\n`. This is because the tool output JSON is encoding special characters. So the actual file content is `<&><&><&>\n`.

I'll write that content to result.txt.

**write**: `{"path":"./result.txt","content":"<&>&<&>&<&>\n"}`

============================================================

### Tool

*tool_call_id: call_cb6864eeba22a36c0aa62baaa8cd17cf*

{"status": "ok"}


============================================================

### Assistant

THINKING:
I've successfully written the contents to result.txt. Let me summarize what I did.

RESPONSE:
I read **source.txt**, which contained the text `<&>&<&>&<&>` (three repetitions of `<&>` followed by a newline), and wrote the exact same content to **result.txt**. The task is complete.