T5.4 - CR-005: a per-project location taxonomy, stored as codes

CLAUDE.md lists CR-005 among the change requests that get "silently half-built if
you treat them as frontend-only". This is the server half and the wizard half
together: a new table, four routes, an Alembic revision, and step 11.

CODES, NOT DISPLAY STRINGS, because CR-018 rolls cost up by these values and a
rollup keyed on a label breaks the day somebody fixes a typo in it. Two columns
carry that: `code` is a node's own slug, derived once at import and never
recomputed; `path` is the full slug path, unique per project, and is what a work
package will store. Renaming a value changes `name` alone - the probe renames a
floor and demands its path comes back byte-identical, with its children's paths
intact.

DEACTIVATE, NEVER DELETE. There is no DELETE route, and the probe checks for its
absence (405) rather than trusting that nobody added one. Deactivating hides a
value from new work packages and cascades DOWN, because a floor nobody can pick
must not keep offering its sectors. Reactivating walks UP only - a sector may
have been switched off for its own reasons, and silently resurrecting it would
undo a decision nobody made twice. That asymmetry is deliberate and is pinned by
a named check so it does not get "fixed" into a surprise.

Import reports rather than merges. Rejected rows come back with the SOURCE line
number and a reason; duplicates are listed as duplicates, separated into "already
in this project" and "already on line N of this import". Reusing a parent is not
a duplicate - B1/L2/1P and B1/L2/2P share a building and a floor by design, and
only the full path repeating counts. Re-importing a deactivated value brings the
same row back rather than creating a second one; the probe checks the id.

One parser, on the server. A CSV is read in the browser and posted as text
exactly as a paste is, so "what does a blank column mean" has one answer.
Comma, semicolon and tab all work - a paste out of a spreadsheet is tab
separated and a saved CSV is not, and which one somebody has is a question the
machine can answer.

No guessed floor names. IMPLEMENTATION.md section 8 says the B100 list has not
been supplied. The seeded sample has "Sample" inside every string, and the probe
greps html/ and server/ for a location-shaped assignment containing any of the
review's real names.

  server/models.py                    LocationNode
  server/alembic/versions/e2a4c7d91b30_location_taxonomy.py
  server/app.py                       GET/POST/PATCH + import, parser, slug
  html/work-package-suite.html        step 11, an 11th rail button
  html/work-package-suite-app.js      the step's logic; LAST_STEP replaces 10
  html/work-package-suite-styles.css  the list, the report
  html/theme-light.css                .field-error, now declared once
  tests/locations_check.py            new - 58 checks
  tests/stepper_check.py              STEP_COUNT 10 -> 11

Done when
  [x] CSV upload and paste both work and report rejected rows with reasons
  [x] duplicates are detected and reported rather than silently merged
  [x] values are editable after import - rename, add, deactivate
  [x] deactivating hides it from new work packages; an existing package
      referencing it still resolves, because the row is retained
  [x] values are stored as codes suitable for grouping
  [x] no guessed real-world floor names exist anywhere in the code

Two decisions worth disagreeing with

  Step 11, appended, not step 2, inserted. Locations belong beside Project by
  subject. Renumbering 2-10 would touch every sop-step-N id, every
  collectStepData case, every gate key and the analytics history - a large
  silent-mismatch surface for an ordering change. The count now lives in one
  place (LAST_STEP), so reordering later is cheap.

  Any project member may edit the list, not only a Project Admin. It matches how
  the SOP baseline itself is authored: the Project Admin gate is on CHANGING a
  completed SOP, not on writing one. If the location list should be tighter than
  the SOP it belongs to, that is a product call.

Verified one at a time
  locations_check  58/58  new
  stepper_check    70/70  (11 steps)
  browser_check    71/71
  a11y             22/22  sop now rings 38 focusable elements
  url_state        23/23
  autosave         34/34
  aggregates       16/16
  pipeline         43/43
  launcher         58/58
  f_items          F1-F5 FIXED, F6 REPRODUCES (T7.2)
  alembic          upgrade / downgrade / upgrade all clean on a throwaway SQLite
                   file, and the migrated schema matches Base.metadata.create_all
                   column for column - dev auto-creates and production migrates,
                   so a divergence between the two is invisible until it ships

.field-error was declared in two page sheets by the end of T5.2 and would have
been three by T5.8, so it moved to theme-light.css. No colour literal added
anywhere: still 0 across all page sheets and inline blocks.

Question for the PR, per CLAUDE.md: the levels are fixed at building / floor /
sector. Micron's floors behave like buildings, which this handles by letting a
project use whichever levels it needs - but a job that wants a fourth level, or
different names for the three, cannot say so. Whether that is worth a
per-project level vocabulary is a product question; the schema would take it
without a migration.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-16 11:06:27 -05:00
parent 6088ef17e8
commit 2081c1ad3c
11 changed files with 1397 additions and 25 deletions

View File

@@ -0,0 +1,65 @@
"""per-project Building / Floor / Sector taxonomy (CR-005)
The location taxonomy differs per project — on Micron, floors within B100 behave
like separate buildings — so it is configured once per SOP instead of hard-coded.
CR-018 rolls cost up by these values, which is why the table stores CODES
(`code`, `path`) beside the display `name`: a rollup keyed on a label breaks the
day somebody fixes a typo in it.
`active` rather than a delete. Deactivating hides a value from new work packages
while every package already referencing it still resolves its label, which is the
same rule CR-002 and CR-016 apply to fields.
Additive only: a new table, no change to any existing one, so nothing to backfill
and nothing to migrate.
Revision ID: e2a4c7d91b30
Revises: a7c31f9e5b02
Create Date: 2026-08-16 09:41:02.118307
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'e2a4c7d91b30'
down_revision = 'a7c31f9e5b02'
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
'location_nodes',
sa.Column('id', sa.String(length=40), nullable=False),
sa.Column('project_id', sa.String(length=40), nullable=False),
sa.Column('parent_id', sa.String(length=40), nullable=True),
sa.Column('level', sa.String(length=20), nullable=False, server_default='building'),
sa.Column('code', sa.String(length=60), nullable=False, server_default=''),
sa.Column('path', sa.String(length=200), nullable=False, server_default=''),
sa.Column('name', sa.String(length=200), nullable=False, server_default=''),
sa.Column('active', sa.Boolean(), nullable=False, server_default=sa.true()),
sa.Column('sort', sa.Integer(), nullable=False, server_default='0'),
sa.Column('created_by', sa.String(length=200), nullable=False, server_default=''),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False,
server_default=sa.func.now()),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False,
server_default=sa.func.now()),
sa.ForeignKeyConstraint(['project_id'], ['projects.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id'),
# One row per (project, path). This is what makes a re-import report a
# duplicate instead of quietly creating a second B100/L2/1P.
sa.UniqueConstraint('project_id', 'path', name='uq_location_path'),
)
op.create_index(op.f('ix_location_nodes_project_id'), 'location_nodes',
['project_id'], unique=False)
op.create_index(op.f('ix_location_nodes_parent_id'), 'location_nodes',
['parent_id'], unique=False)
op.create_index(op.f('ix_location_nodes_path'), 'location_nodes', ['path'], unique=False)
def downgrade() -> None:
op.drop_index(op.f('ix_location_nodes_path'), table_name='location_nodes')
op.drop_index(op.f('ix_location_nodes_parent_id'), table_name='location_nodes')
op.drop_index(op.f('ix_location_nodes_project_id'), table_name='location_nodes')
op.drop_table('location_nodes')

View File

@@ -1868,6 +1868,350 @@ def wp_metrics(project_id: Optional[str] = Query(None), sop_id: Optional[str] =
}
# ── Location taxonomy — CR-005 ────────────────────────────────────────────────
# Building / Floor / Sector, configured once per project rather than hard-coded.
# See models.LocationNode for why this stores codes as well as names, and why
# nothing here deletes.
LOCATION_LEVELS = ("building", "floor", "sector")
_SLUG_STRIP = re.compile(r"[^A-Z0-9]+")
def location_slug(name: str) -> str:
"""A stable code from a display name: upper-cased, non-alphanumerics collapsed
to a single dash, trimmed. `Level 2` -> `LEVEL-2`, `1P (chase)` -> `1P-CHASE`.
Derived ONCE, at import, and never recomputed — see LocationNode. Returns ''
when there is nothing to make a code from, which the caller turns into a
rejected row with a reason rather than a silently-skipped one."""
return _SLUG_STRIP.sub("-", (name or "").strip().upper()).strip("-")[:60]
def parse_location_rows(text: str) -> tuple[list[tuple[int, list[str]]], list[dict]]:
"""Split pasted or uploaded text into (rows, rejected).
Rows come back paired with their SOURCE line number, not their index among the
accepted ones. "Duplicate on row 12" has to mean row 12 of the file somebody is
looking at, or the report sends them to the wrong line.
Accepts comma, tab or semicolon separators — a paste out of Excel is
tab-separated and a saved CSV is not, and asking which one somebody has is a
question the machine can answer. A header line naming the levels is skipped.
Every rejection carries the line number and a reason. A silent skip is the
failure mode this endpoint exists to avoid: an import that says "42 rows" over
a file with 50 in it has lost eight and told nobody."""
rows: list[tuple[int, list[str]]] = []
rejected: list[dict] = []
for i, raw in enumerate((text or "").splitlines(), start=1):
line = raw.strip()
if not line:
continue
if "\t" in raw:
parts = [c.strip() for c in raw.split("\t")]
elif ";" in line and "," not in line:
parts = [c.strip() for c in line.split(";")]
else:
parts = [c.strip() for c in line.split(",")]
parts = [p.strip().strip('"').strip() for p in parts]
while parts and not parts[-1]:
parts.pop()
if not parts:
continue
low = [p.lower() for p in parts]
if i == 1 and low[:1] in (["building"], ["bldg"]):
continue # header row
if len(parts) > len(LOCATION_LEVELS):
rejected.append({"line": i, "text": line,
"reason": "more than 3 columns — expected building, floor, sector"})
continue
if not parts[0]:
rejected.append({"line": i, "text": line,
"reason": "no building — a floor or sector needs one above it"})
continue
# A gap in the middle ("B100,,1P") would attach a sector to nothing.
gap = next((n for n, p in enumerate(parts) if not p), None)
if gap is not None and any(parts[gap + 1:]):
rejected.append({"line": i, "text": line,
"reason": "a %s is named with no %s above it"
% (LOCATION_LEVELS[len(parts) - 1], LOCATION_LEVELS[gap])})
continue
parts = [p for p in parts if p]
if any(not location_slug(p) for p in parts):
rejected.append({"line": i, "text": line,
"reason": "no letters or digits to make a code from"})
continue
rows.append((i, parts))
return rows, rejected
def _location_tree(db: Session, project_id: str, include_inactive: bool):
stmt = select(models.LocationNode).where(models.LocationNode.project_id == project_id)
if not include_inactive:
stmt = stmt.where(models.LocationNode.active.is_(True))
nodes = db.scalars(stmt.order_by(models.LocationNode.path)).all()
return nodes
@app.get("/api/projects/{project_id}/locations")
def list_locations(project_id: str,
include_inactive: bool = Query(False),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""The project's taxonomy, flat, ordered by path so a caller can rebuild the
tree without a second query.
`include_inactive` defaults to FALSE, which is what makes a deactivated value
disappear from new work packages. It is available as true because a package
that already references a deactivated value still has to render its label —
deactivating hides a choice, it does not rewrite history."""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, project_id)
nodes = _location_tree(db, project_id, include_inactive)
return {
"project_id": project_id,
"levels": list(LOCATION_LEVELS),
"nodes": [n.to_dict() for n in nodes],
"counts": {lvl: sum(1 for n in nodes if n.level == lvl) for lvl in LOCATION_LEVELS},
"generated_at": models.utcnow().isoformat(),
}
class LocationImportIn(BaseModel):
text: str = ""
dry_run: bool = False
@app.post("/api/projects/{project_id}/locations/import")
def import_locations(project_id: str, body: LocationImportIn,
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""Bulk import from CSV or a paste. Same code path for both: a file is read in
the browser and posted as text, because two parsers would be two sets of rules
about what a blank column means.
Reports rather than merges. A row naming a building/floor/sector combination
that already exists — in this file or in the project — comes back in
`duplicates` with its line number. Reusing a PARENT is not a duplicate:
`B100,L2,1P` and `B100,L2,2P` share a building and a floor by design, and it
is only the full path repeating that is a duplicate.
`dry_run` parses and reports without writing, which is what lets the wizard
show what an import will do before it does it."""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
# Any project member, matching how the SOP baseline itself is authored — the
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The location list cannot be changed")
rows, rejected = parse_location_rows(body.text)
existing = {n.path: n for n in db.scalars(
select(models.LocationNode).where(models.LocationNode.project_id == project_id)
).all()}
before = set(existing)
created: list[dict] = []
duplicates: list[dict] = []
reactivated: list[dict] = []
seen_in_file: dict[str, int] = {}
next_sort = max((n.sort for n in existing.values()), default=0)
for line_no, parts in rows:
segs = [location_slug(p) for p in parts]
full = "/".join(segs)
if full in seen_in_file:
duplicates.append({"line": line_no, "path": full, "names": parts,
"reason": "already on line %d of this import" % seen_in_file[full]})
continue
seen_in_file[full] = line_no
if full in before:
node = existing[full]
if not node.active:
# Re-importing a value somebody deactivated is a request to bring it
# back, not a duplicate — and it must reuse the SAME row, or every
# work package pointing at the old path is orphaned.
if not body.dry_run:
node.active = True
reactivated.append({"path": full, "names": parts})
else:
duplicates.append({"line": line_no, "path": full, "names": parts,
"reason": "already in this project"})
continue
# Create any missing ancestors, then the leaf. Sharing a parent is the
# normal case, not a collision.
parent_id = None
for depth, seg in enumerate(segs):
sub = "/".join(segs[:depth + 1])
node = existing.get(sub)
if node is None:
next_sort += 1
node = models.LocationNode(
id=gen_id("loc"), project_id=project_id, parent_id=parent_id,
level=LOCATION_LEVELS[depth], code=seg, path=sub,
name=parts[depth], active=True, sort=next_sort,
created_by=user.username,
)
existing[sub] = node
if not body.dry_run:
db.add(node)
if sub not in before:
created.append({"path": sub, "level": LOCATION_LEVELS[depth],
"code": seg, "name": parts[depth]})
parent_id = node.id
result = {
"project_id": project_id, "dry_run": bool(body.dry_run),
"read": len(rows) + len(rejected),
"created": created, "duplicates": duplicates,
"reactivated": reactivated, "rejected": rejected,
}
if body.dry_run:
db.rollback()
return result
log_event(db, user, "locations_imported", "project", project_id, project_id,
summary="%d location value(s) added" % len(created),
detail={"created": len(created), "duplicates": len(duplicates),
"rejected": len(rejected), "reactivated": len(reactivated)})
db.commit()
return result
class LocationIn(BaseModel):
level: str = "building"
parent_id: Optional[str] = None
name: str = ""
@app.post("/api/projects/{project_id}/locations")
def add_location(project_id: str, body: LocationIn,
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""Add one value by hand. Same rules as the import — an import you cannot
correct afterwards is an import nobody trusts enough to run."""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
# Any project member, matching how the SOP baseline itself is authored — the
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The location list cannot be changed")
name = (body.name or "").strip()
code = location_slug(name)
if not code:
raise HTTPException(status_code=400,
detail="That name has no letters or digits to make a code from")
level = (body.level or "").strip().lower()
if level not in LOCATION_LEVELS:
raise HTTPException(status_code=400,
detail="Level must be one of: " + ", ".join(LOCATION_LEVELS))
depth = LOCATION_LEVELS.index(level)
parent = None
if body.parent_id:
parent = db.get(models.LocationNode, body.parent_id)
if not parent or parent.project_id != project_id:
raise HTTPException(status_code=400, detail="Parent is not on this project")
if depth == 0 and parent is not None:
raise HTTPException(status_code=400, detail="A building has nothing above it")
if depth > 0 and parent is None:
raise HTTPException(status_code=400,
detail="A %s needs a %s above it" % (level, LOCATION_LEVELS[depth - 1]))
if parent is not None and LOCATION_LEVELS.index(parent.level) != depth - 1:
raise HTTPException(status_code=400,
detail="A %s cannot sit under a %s" % (level, parent.level))
path = (parent.path + "/" + code) if parent else code
clash = db.scalars(select(models.LocationNode).where(
(models.LocationNode.project_id == project_id) & (models.LocationNode.path == path)
)).first()
if clash:
if not clash.active:
clash.active = True
log_event(db, user, "location_reactivated", "project", project_id, project_id,
summary=path)
db.commit()
return clash.to_dict()
raise HTTPException(status_code=409, detail="%s” is already on this project" % name)
top = db.scalar(select(func.max(models.LocationNode.sort)).where(
models.LocationNode.project_id == project_id)) or 0
node = models.LocationNode(
id=gen_id("loc"), project_id=project_id, parent_id=(parent.id if parent else None),
level=level, code=code, path=path, name=name, active=True, sort=top + 1,
created_by=user.username,
)
db.add(node)
log_event(db, user, "location_added", "project", project_id, project_id, summary=path)
db.commit()
return node.to_dict()
class LocationPatch(BaseModel):
name: Optional[str] = None
active: Optional[bool] = None
@app.patch("/api/projects/{project_id}/locations/{node_id}")
def update_location(project_id: str, node_id: str, body: LocationPatch,
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""Rename or deactivate. There is no DELETE, and that is the design:
rename changes `name` only. `code` and `path` are untouched, so every
work package pointing at this value keeps pointing at it. That
is the whole reason the two are separate columns.
deactivate hides the value from new work packages. Cascades DOWN — a floor
nobody can pick makes its sectors unpickable too, and leaving
them offered would be offering a path to nowhere. Reactivating
a child reactivates its ancestors for the same reason.
"""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
# Any project member, matching how the SOP baseline itself is authored — the
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The location list cannot be changed")
node = db.get(models.LocationNode, node_id)
if not node or node.project_id != project_id:
raise HTTPException(status_code=404, detail="Location not found on this project")
changed = {}
if body.name is not None:
name = body.name.strip()
if not name:
raise HTTPException(status_code=400, detail="A name cannot be empty")
if name != node.name:
changed["name"] = {"from": node.name, "to": name}
node.name = name
if body.active is not None and bool(body.active) != bool(node.active):
changed["active"] = {"from": bool(node.active), "to": bool(body.active)}
node.active = bool(body.active)
if not node.active:
for child in db.scalars(select(models.LocationNode).where(
(models.LocationNode.project_id == project_id)
& (models.LocationNode.path.like(node.path + "/%"))
)).all():
child.active = False
else:
# Walk up by path rather than by parent_id: one query, and it cannot
# loop on a malformed chain.
segs = node.path.split("/")
ancestors = ["/".join(segs[:n]) for n in range(1, len(segs))]
if ancestors:
for anc in db.scalars(select(models.LocationNode).where(
(models.LocationNode.project_id == project_id)
& (models.LocationNode.path.in_(ancestors))
)).all():
anc.active = True
if changed:
log_event(db, user, "location_updated", "project", project_id, project_id,
summary=node.path, detail=changed)
db.commit()
return node.to_dict()
@app.get("/api/projects/{project_id}/summary")
def project_summary(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""What the launcher needs to describe a project without asking the browser

View File

@@ -224,6 +224,68 @@ class ProjectMember(Base):
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
class LocationNode(Base):
"""One value in a project's Building / Floor / Sector taxonomy — CR-005.
The taxonomy differs per project. On Micron, floors within B100 behave like
separate buildings, so floor and sector are the unit of both execution and
cost tracking; on another job "building" may be the only level that means
anything. So it is configured once per SOP rather than hard-coded, and no
real-world floor name appears anywhere in this repository.
CODES, NOT DISPLAY STRINGS. `CR-018` rolls cost up by these, and a rollup
keyed on a label breaks the day somebody fixes a typo in it. Two columns
carry that:
code this node's own slug among its siblings, derived once from the name
it was imported with and then NEVER recomputed — renaming a node is
a display change, which is exactly what makes rename safe for the
work packages already pointing at it.
path the full slug path from the root, '/'-joined and unique per project
(`B100/L2/1P`). This is the grouping key and the value a work
package stores.
DEACTIVATE, NEVER DELETE. `active=False` hides a value from new work
packages; every existing package referencing it still resolves its label,
because the row is still there. Same rule as `CR-002`/`CR-016`: removal is
expressed as a toggle, and the data is retained.
"""
__tablename__ = "location_nodes"
__table_args__ = (
UniqueConstraint("project_id", "path", name="uq_location_path"),
)
LEVELS = ("building", "floor", "sector")
id: Mapped[str] = mapped_column(String(40), primary_key=True)
project_id: Mapped[str] = mapped_column(
String(40), ForeignKey("projects.id", ondelete="CASCADE"), index=True
)
# Self-reference by id. No ForeignKey to its own table for the same reason the
# rest of this file declares none — see the module docstring — and because a
# self-referential FK plus SQLite's deferred-constraint behaviour makes a bulk
# import fiddly for no gain. Orphans are prevented in the API, which is the
# only writer.
parent_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True)
level: Mapped[str] = mapped_column(String(20), default="building") # building | floor | sector
code: Mapped[str] = mapped_column(String(60), default="") # own slug
path: Mapped[str] = mapped_column(String(200), default="", index=True) # full slug path
name: Mapped[str] = mapped_column(String(200), default="") # display label
active: Mapped[bool] = mapped_column(Boolean, default=True)
sort: Mapped[int] = mapped_column(Integer, default=0)
created_by: Mapped[str] = mapped_column(String(200), default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)
def to_dict(self) -> dict:
return {
"id": self.id, "project_id": self.project_id, "parent_id": self.parent_id,
"level": self.level, "code": self.code, "path": self.path, "name": self.name,
"active": bool(self.active), "sort": self.sort,
"created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at),
}
class Comment(Base):
__tablename__ = "comments"