[Bug]: Mid-sentence Claude skill chips split the original request during native expansion #13256

#!/usr/bin/env python3
"""Capture Claude's actual model request using a loopback-only fake Messages API."""
import argparse
import http.server
import json
import os
from pathlib import Path
import subprocess
import tempfile
import threading
import time

parser = argparse.ArgumentParser()
parser.add_argument('--repo', default=str(Path.cwd()))
parser.add_argument('--case', choices=['original', 'minimal', 'start', 'prefix-only', 'no-skill'], default='minimal')
parser.add_argument('--mode', choices=['t3', 'plain', 'split', 'split-newline'], default='split')
args = parser.parse_args()
prompts = {
    'original': 'please $implement the fixes\n\nThen report the result',
    'minimal': 'you will $implement fixes',
    'start': '$implement fixes',
    'prefix-only': 'before $implement',
    'no-skill': 'you will implement fixes',
}
prompt = prompts[args.case]
if args.mode == 't3':
    module = Path(args.repo) / 'apps/server/src/provider/Drivers/ClaudeSkillDispatch.ts'
    js = '''import { pathToFileURL } from "node:url";
const { planClaudeSkillDispatch } = await import(pathToFileURL(process.argv[1]).href);
console.log(JSON.stringify(planClaudeSkillDispatch(process.argv[2], new Set(["implement"])) ?? null));'''
    plan = json.loads(subprocess.check_output(['node', '--input-type=module', '-e', js, str(module), prompt], text=True))
else:
    before, _, after = prompt.partition('$implement')
    plan = {'leadingText':before.rstrip() or None, 'commandText':'/implement'+after} if '$implement' in prompt else None
if args.mode == 'plain':
    blocks = [{'type': 'text', 'text': prompt.replace('$implement', '/implement')}]
else:
    blocks = []
    if plan is None:
        blocks.append({'type': 'text', 'text': prompt})
    else:
        if plan.get('leadingText'):
            blocks.append({'type': 'text', 'text': plan['leadingText'] + ('\n\n' if args.mode == 'split-newline' else '')})
        blocks.append({'type': 'text', 'text': plan['commandText']})
captured = []
class Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, *unused):
        pass
    def do_POST(self):
        data = json.loads(self.rfile.read(int(self.headers.get('Content-Length', 0))) or '{}')
        if 'messages' not in data:
            self.send_response(200); self.end_headers(); self.wfile.write(b'{}'); return
        captured.append(data)
        response = {'id': 'msg_fixture', 'type': 'message', 'role': 'assistant', 'model': data.get('model'), 'content': [{'type':'text','text':'fixture complete'}], 'stop_reason':'end_turn', 'stop_sequence':None, 'usage':{'input_tokens':1,'output_tokens':1}}
        if not data.get('stream'):
            self.send_response(200); self.send_header('Content-Type','application/json'); self.end_headers(); self.wfile.write(json.dumps(response).encode()); return
        events = [
            ('message_start', {'type':'message_start','message':{**response,'content':[],'stop_reason':None}}),
            ('content_block_start', {'type':'content_block_start','index':0,'content_block':{'type':'text','text':''}}),
            ('content_block_delta', {'type':'content_block_delta','index':0,'delta':{'type':'text_delta','text':'fixture complete'}}),
            ('content_block_stop', {'type':'content_block_stop','index':0}),
            ('message_delta', {'type':'message_delta','delta':{'stop_reason':'end_turn','stop_sequence':None},'usage':{'output_tokens':1}}),
            ('message_stop', {'type':'message_stop'}),
        ]
        self.send_response(200); self.send_header('Content-Type','text/event-stream'); self.end_headers()
        for event, body in events:
            self.wfile.write(('event: '+event+'\ndata: '+json.dumps(body)+'\n\n').encode())

started = time.monotonic()
server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
with tempfile.TemporaryDirectory(prefix='t3-skill-fixture-') as tmp:
    root = Path(tmp)
    home = root / 'config'
    cwd = root / 'project'; cwd.mkdir()
    skill = cwd / '.claude' / 'skills' / 'implement' / 'SKILL.md'
    skill.parent.mkdir(parents=True)
    skill.write_text('---\ndescription: Minimal serialization fixture\n---\nSKILL_BODY_SENTINEL\n\nARGUMENTS: $ARGUMENTS\n')
    env = {k:v for k,v in os.environ.items() if not k.startswith(('ANTHROPIC_', 'CLAUDE_', 'CLAUDECODE', 'AWS_', 'GOOGLE_', 'OTEL_'))}
    env.update(ANTHROPIC_BASE_URL=f'http://127.0.0.1:{server.server_port}', ANTHROPIC_API_KEY='local-fixture-only', CLAUDE_CONFIG_DIR=str(home), DISABLE_TELEMETRY='1', DISABLE_ERROR_REPORTING='1', DISABLE_AUTOUPDATER='1', CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC='1')
    command = ['claude', '-p', '--input-format', 'stream-json', '--output-format', 'stream-json', '--verbose', '--no-session-persistence', '--strict-mcp-config', '--mcp-config', '{"mcpServers":{}}', '--settings', '{"disableAllHooks":true}', '--setting-sources', 'project', '--tools', '', '--model', 'claude-sonnet-4-6', '--system-prompt', 'Serialization fixture. Respond with fixture complete.']
    message = {'type':'user','session_id':'','parent_tool_use_id':None,'message':{'role':'user','content':blocks}}
    try:
        result = subprocess.run(command, input=json.dumps(message)+'\n', text=True, capture_output=True, cwd=cwd, env=env, timeout=25)
    finally:
        server.shutdown(); server.server_close(); thread.join()
    if not captured:
        print('HARNESS ERROR: no Messages request', result.returncode, result.stderr[:1500], result.stdout[-2500:])
        raise SystemExit(2)
    texts = [block.get('text','') for msg in captured[-1]['messages'] if msg.get('role')=='user' for block in (msg.get('content') if isinstance(msg.get('content'),list) else [{'text':msg.get('content','')}]) if block.get('type','text')=='text']
    texts = [s.replace(str(root.resolve()), '<FIXTURE>').replace(tmp, '<FIXTURE>') for s in texts if not s.startswith('<system-reminder>')]
    intended = prompt.replace('$implement', '/implement')
    expanded = any('SKILL_BODY_SENTINEL' in t for t in texts)
    if not expanded and args.mode != 'plain' and args.case != 'no-skill':
        print('HARNESS ERROR: fixture skill did not expand', json.dumps(texts, indent=2))
        raise SystemExit(2)
    preserved = any(intended in t for t in texts)
    verdict = ((preserved and not expanded) if args.mode == 'plain' or args.case == 'no-skill' else expanded and (preserved if args.case != 'start' else any('ARGUMENTS: fixes' in t for t in texts)))
    report = {'case':args.case,'mode':args.mode,'input':prompt,'sdk_blocks':blocks,'model_text_blocks':texts,'skill_expanded':expanded,'original_sentence_preserved':preserved,'cli_exit':result.returncode,'elapsed_seconds':round(time.monotonic()-started,2),'verdict':'PASS' if verdict else 'FAIL'}
    output = Path(__file__).with_name(f'{args.case}-{args.mode}.json')
    output.write_text(json.dumps(report, indent=2)+'\n')
    print(json.dumps(report,indent=2))
    raise SystemExit(0 if verdict else 1)