tegwick
5fec7de4cb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
183 lines
12 KiB
Python
183 lines
12 KiB
Python
"""Human workspace adapter: forms and rendering, with no storage or policy rules."""
|
|
|
|
from dataclasses import replace
|
|
import re
|
|
from urllib.parse import urlencode
|
|
import uuid
|
|
|
|
from .lifecycle_commands import decode_memo
|
|
from .lifecycle_service import Caller, LifecycleError
|
|
from .memo import Memo, BindingLevel, BindingSlice, Principal, Scope, StepKind
|
|
from .records import dumps
|
|
from .store import EvidenceUnavailable
|
|
from .ui import document, hidden, text
|
|
import json
|
|
|
|
|
|
def _form(action, session, fields, button):
|
|
return (f'<form method="post" action="{text(action)}">'
|
|
+ hidden("csrf", session.csrf) + hidden("key", str(uuid.uuid4()))
|
|
+ fields + f'<button type="submit">{text(button)}</button></form>')
|
|
|
|
|
|
def _textarea(name, label, value=""):
|
|
return f'<label>{text(label)}<textarea name="{text(name)}" required>{text(value)}</textarea></label>'
|
|
|
|
|
|
def _lines(value):
|
|
return [line.strip() for line in value.splitlines() if line.strip()]
|
|
|
|
|
|
class LifecycleWorkspace:
|
|
def __init__(self, service):
|
|
self.service = service
|
|
|
|
def dispatch(self, method, path, params, session, form=None):
|
|
caller = Caller.from_human_session(session)
|
|
self.service._caller(caller)
|
|
if method == "GET" and path == "/decisions":
|
|
if set(params) - {"q", "cursor"}:
|
|
raise ValueError()
|
|
query = params.get("q", "")
|
|
page = self.service.search(caller, query=query, cursor=params.get("cursor"))
|
|
content = ('<h1>Decision workspace</h1><p>Prepare a decision, examine evidence, and invite discussion.</p>'
|
|
'<form method="get" action="/decisions"><label>Search decisions '
|
|
f'<input name="q" value="{text(query)}" maxlength="200"></label><button>Search</button></form>'
|
|
'<p><a href="/decisions/new/prepare">Prepare a decision</a></p><ul>')
|
|
for row in page["decisions"]:
|
|
content += (f'<li><a href="/decisions/{text(row["memo_id"])}">{text(row["question"])}</a>'
|
|
f' · revision {row["version"]}</li>')
|
|
content += '</ul>'
|
|
if not page["decisions"]:
|
|
content += '<p>No decisions to display on this page.</p>'
|
|
if page["next_cursor"]:
|
|
content += f'<p><a href="/decisions?{text(urlencode({"q": query, "cursor": page["next_cursor"]}))}">Next page</a></p>'
|
|
return 200, document("Decision workspace", content, session.subject), []
|
|
if path == "/decisions/new/prepare":
|
|
if params:
|
|
raise ValueError()
|
|
if method == "GET":
|
|
fields = (_textarea("question", "Question") + _textarea("brief", "Context, alternatives and recommendation")
|
|
+ '<label>Subject or scope identifier <input name="scope_id" required maxlength="200"></label>'
|
|
'<label>Subject or scope label <input name="scope_label" required maxlength="200"></label>')
|
|
content = '<h1>Prepare a decision</h1>' + _form(path, session, fields, "Save draft")
|
|
return 200, document("Prepare a decision", content, session.subject), []
|
|
if method == "POST":
|
|
self._fields(form, {"question", "brief", "scope_id", "scope_label"})
|
|
memo = Memo(id=str(uuid.uuid4()), version=1, question=form["question"], brief=form["brief"],
|
|
requested_act="accept", binding_level=BindingLevel.ORGANIZATIONAL,
|
|
step_kind=StepKind.MITZEICHNUNG,
|
|
binding=BindingSlice(Principal("pending-human-session", "person", "To be assigned"),
|
|
Scope("resource", form["scope_id"], form["scope_label"])))
|
|
# Stable create identity across an HTTP retry of the same rendered form.
|
|
memo = replace(memo, id=str(uuid.uuid5(uuid.NAMESPACE_URL,
|
|
dumps([caller.attribution.tenant, session.subject, form["key"]]))))
|
|
self.service.register(caller, decode_memo(json.loads(dumps(memo))), key=form["key"])
|
|
return 303, "", [("Location", "/decisions/" + memo.id)]
|
|
match = re.fullmatch(r"/decisions/([A-Za-z0-9._:-]{1,256})(?:/(revise|comment|publish|source|evaluate|packet/([0-9]{1,3})))?", path)
|
|
if not match:
|
|
return 404, "Not found.", []
|
|
memo_id, action, index = match.groups()
|
|
if method == "GET" and (action is None or index is not None):
|
|
if set(params) - {"version", "after"}:
|
|
raise ValueError()
|
|
version = int(params["version"]) if "version" in params else None
|
|
if index is not None:
|
|
if version is None or "after" in params:
|
|
raise ValueError()
|
|
return 200, self.service.packet(caller, memo_id, version, int(index)), [
|
|
("Content-Disposition", f'attachment; filename="decision-document-{int(index)+1}.bin"')]
|
|
memo = self.service.retrieve(caller, memo_id, version)
|
|
events = self.service.events(caller, memo_id, after=int(params.get("after", "0")))
|
|
previous = None
|
|
if memo.version > 1:
|
|
try:
|
|
previous = self.service.retrieve(caller, memo_id, memo.version - 1)
|
|
except EvidenceUnavailable:
|
|
pass
|
|
except LifecycleError as error:
|
|
if error.status != 403:
|
|
raise
|
|
return 200, self._page(memo, events, session, previous), []
|
|
if method == "POST" and action in {"revise", "comment", "publish", "source", "evaluate"}:
|
|
if params:
|
|
raise ValueError()
|
|
fields = {"revise": {"version", "question", "brief", "summary"},
|
|
"comment": {"version", "text"}, "publish": {"version", "subjects", "groups"},
|
|
"source": {"version", "reference", "retrieved_at", "digest", "summary"},
|
|
"evaluate": {"version", "assessment", "source_ids", "uncertainty"}}[action]
|
|
self._fields(form, fields)
|
|
version = int(form["version"])
|
|
if action == "revise":
|
|
previous = self.service.retrieve(caller, memo_id, version)
|
|
memo = previous.next_version(question=form["question"], brief=form["brief"])
|
|
self.service.revise(caller, decode_memo(json.loads(dumps(memo))), expected_version=version,
|
|
summary=form["summary"], key=form["key"])
|
|
elif action == "publish":
|
|
self.service.publish(caller, memo_id, version, key=form["key"],
|
|
audience={"subjects": _lines(form["subjects"]), "groups": _lines(form["groups"])})
|
|
else:
|
|
body = {key: form[key] for key in fields - {"version"}}
|
|
if action == "evaluate":
|
|
body["source_ids"] = _lines(body["source_ids"])
|
|
self.service.annotate(caller, memo_id, version, key=form["key"],
|
|
kind="evaluation" if action == "evaluate" else action, body=body)
|
|
return 303, "", [("Location", "/decisions/" + memo_id)]
|
|
return 405, "Method not allowed.", []
|
|
|
|
@staticmethod
|
|
def _fields(form, required):
|
|
if not isinstance(form, dict) or set(form) != required | {"csrf", "key"}:
|
|
raise ValueError()
|
|
uuid.UUID(form["key"])
|
|
|
|
def _page(self, memo, events, session, previous=None):
|
|
base = "/decisions/" + memo.id
|
|
revision = hidden("version", memo.version)
|
|
content = (f'<p><a href="/decisions">Decision workspace</a></p><h1>{text(memo.question)}</h1>'
|
|
f'<p>Revision {memo.version} · {text(memo.binding.target.label)}</p>'
|
|
f'<section class="card"><h2>Context and recommendation</h2><p class="brief">{text(memo.brief)}</p></section>'
|
|
'<p>Human acceptance is not available for this draft. Publishing invites discussion; it does not record acceptance.</p>')
|
|
if previous is not None:
|
|
content += f'<details class="card"><summary>Compare with revision {previous.version}</summary>'
|
|
for field, label in (("question", "Question"), ("brief", "Context and recommendation")):
|
|
before, after = getattr(previous, field), getattr(memo, field)
|
|
if before != after:
|
|
content += (f'<h3>{label}</h3><p>Before</p><p class="brief">{text(before)}</p>'
|
|
f'<p>After</p><p class="brief">{text(after)}</p>')
|
|
content += '</details>'
|
|
if memo.packet:
|
|
content += '<section class="card"><h2>Supporting documents</h2><ul>'
|
|
for index, item in enumerate(memo.packet):
|
|
content += f'<li><a href="{text(base)}/packet/{index}?version={memo.version}">{text(item.label)}</a></li>'
|
|
content += '</ul></section>'
|
|
content += '<section class="card"><h2>Discussion and history</h2>'
|
|
for event in events:
|
|
body = {key: value for key, value in event["body"].items() if key != "policy_observation_id"}
|
|
content += (f'<article class="record"><h3>{text(event["kind"])} · revision {event["version"]}</h3>'
|
|
f'<p>{text(event["actor"]["sub"])} ({text(event["actor"]["kind"])}) · {text(event["at"])}</p>')
|
|
if event["kind"] == "source":
|
|
content += f'<p>Source reference: <code>{text(event["id"])}</code></p>'
|
|
for key, value in body.items():
|
|
content += f'<p><strong>{text(key.replace("_", " "))}:</strong> {text(dumps(value) if isinstance(value, (dict, list)) else value)}</p>'
|
|
content += '</article>'
|
|
if len(events) == 100:
|
|
content += f'<p><a href="{text(base)}?version={memo.version}&after={events[-1]["sequence"]}">More history</a></p>'
|
|
content += '</section><section class="card"><h2>Add a comment</h2>'
|
|
content += _form(base + "/comment", session, revision + _textarea("text", "Comment"), "Add comment") + '</section>'
|
|
content += '<details class="card"><summary>Refine this decision</summary>'
|
|
content += _form(base + "/revise", session, revision + _textarea("question", "Question", memo.question)
|
|
+ _textarea("brief", "Context and recommendation", memo.brief)
|
|
+ _textarea("summary", "What changed"), "Save new revision") + '</details>'
|
|
content += '<details class="card"><summary>Publish for discussion</summary><p>Enter explicit member or group identifiers, one per line.</p>'
|
|
content += _form(base + "/publish", session, revision + '<label>Members<textarea name="subjects"></textarea></label>'
|
|
'<label>Groups<textarea name="groups"></textarea></label>', "Publish this revision") + '</details>'
|
|
content += '<details class="card"><summary>Add a research source</summary>'
|
|
content += _form(base + "/source", session, revision + ''.join(_textarea(name, label) for name, label in (
|
|
("reference", "Source location or reference"), ("retrieved_at", "Retrieved at (date, time and timezone)"),
|
|
("digest", "Source content digest (sha256:...)"), ("summary", "What the source supports"))), "Add source") + '</details>'
|
|
content += '<details class="card"><summary>Evaluate the evidence</summary>'
|
|
content += _form(base + "/evaluate", session, revision + _textarea("assessment", "Assessment")
|
|
+ '<label>Source references (one per line)<textarea name="source_ids"></textarea></label>'
|
|
+ _textarea("uncertainty", "Uncertainty and open questions"), "Add evaluation") + '</details>'
|
|
return document("Decision workspace", content, session.subject)
|