"""Bounded local NDJSON object-record parser; no TCP/SSE/model service."""
import json
def pairs(values):
    d={}
    for k,v in values:
        if k in d:raise ValueError('duplicate decoded key')
        d[k]=v
    return d
def bad_constant(x):raise ValueError('non-JSON numeric constant')
class Parser:
    def __init__(self,limit=256,ignore_empty=True):self.limit=limit;self.ignore_empty=ignore_empty;self.buffer=b'';self.records=[];self.closed=False
    def feed(self,chunk):
        if self.closed or not isinstance(chunk,bytes)or len(chunk)>4096:raise ValueError('open parser and bounded bytes chunk required')
        self.buffer+=chunk
        while b'\n'in self.buffer:
            line,self.buffer=self.buffer.split(b'\n',1)
            if line.endswith(b'\r'):line=line[:-1]
            if b'\r'in line:raise ValueError('CR only allowed once immediately before delimiter LF')
            if len(line)>self.limit:raise ValueError('record byte budget exceeded')
            if not line:
                if self.ignore_empty:continue
                raise ValueError('empty line disabled by contract')
            text=line.decode('utf-8','strict');obj=json.loads(text,object_pairs_hook=pairs,parse_constant=bad_constant)
            if not isinstance(obj,dict):raise ValueError('local object-only subset, not all NDJSON values')
            self.records.append(obj)
        if len(self.buffer)>self.limit+1:raise ValueError('unfinished record budget exceeded')
    def finish(self):
        if self.buffer:raise ValueError('EOF before required LF record terminator')
        self.closed=True;return self.records
def read(chunks,**kwargs):
    p=Parser(**kwargs)
    for chunk in chunks:p.feed(chunk)
    return p.finish()
def demo():
    rows=[{'id':1,'text':'سلام 🙂'},{'id':2,'text':'خط اول\nخط دوم'},{'id':3,'data':{'ok':True}}]
    stream=(json.dumps(rows[0],ensure_ascii=False)+'\n\n'+json.dumps(rows[1],ensure_ascii=False)+'\r\n'+json.dumps(rows[2],ensure_ascii=False)+'\n').encode('utf-8')
    for i in range(len(stream)+1):assert read([stream[:i],stream[i:]])==rows
    assert read([bytes([b])for b in stream])==rows and read([stream])==rows
    errors={}
    cases={'invalid_utf8':b'{"x":"\xff"}\n','malformed_json':b'{"x":}\n','nested_duplicate':b'{"a":{"x":1,"x":2}}\n','missing_lf':b'{"x":1}','oversized':b'{"x":"'+b'a'*300+b'"}\n','unescaped_physical_lf':b'{"x":"a\nb"}\n','nan_constant':b'{"x":NaN}\n','internal_cr':b'{"x":\r1}\n','double_cr_before_lf':b'{"x":1}\r\r\n'}
    for name,value in cases.items():
        try:read([value])
        except(ValueError,UnicodeError)as e:errors[name]=type(e).__name__+': '+str(e)
        else:raise AssertionError('invalid accepted')
    try:read([b'\n'],ignore_empty=False)
    except ValueError:empty_rejected=True
    else:raise AssertionError('empty contract ignored')
    cr_splits=0
    for name in ['internal_cr','double_cr_before_lf']:
        value=cases[name]
        for i in range(len(value)+1):
            try:read([value[:i],value[i:]])
            except ValueError:cr_splits+=1
            else:raise AssertionError('invalid CR accepted under different chunk boundary')
    return {'authored_rows':rows,'authored_stream_utf8_hex':stream.hex(),'stream_bytes':len(stream),'all_two_part_split_positions_tested':len(stream)+1,'byte_at_a_time_and_all_records_one_chunk_equal_oracle':True,'parsed':read([stream]),'invalid_cases':errors,'invalid_cr_all_split_positions_rejected':cr_splits,'blank_line_ignore_configurable':empty_rejected,'scope':'Object-only bounded records256bytes and inputchunk4096bytes, LF/CRLF with exactly optional singleCR immediately beforeLF; any other rawCR rejected. ignoreempty defaulttrue configurable, strictEOF requires finalLF. UTF8 decoded after complete line; JSON escaped newline stays inside value. Duplicate keys rejected recursively. All local bytes, no transport throughput, SSE/ASR/API compatibility claim.'}
if __name__=='__main__':print(json.dumps(demo(),ensure_ascii=False,indent=2))