sse (2f0efe33) · Commits · jan.koester / blogi

editor/src/webedit_api.cpp

+262 −0

Changes for editor/src/webedit_api.cpp: 262 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -3496,6 +3496,25 @@ std::string generateAiAgentSystemPrompt() {
		"layout; include @media rules for responsiveness (max-width: 768px and 480px).\n";
		}

		// Pulls the "data: <payload>" line out of one SSE event block (which may
		// have other lines, e.g. "event: ..."). Returns "" if there's no data line.
		std::string extractSseDataLine(const std::string &eventBlock) {
		size_t start = 0;
		while (start < eventBlock.size()) {
		size_t eol = eventBlock.find('\n', start);
		std::string line = eventBlock.substr(start, eol == std::string::npos
		? std::string::npos : eol - start);
		if (line.rfind("data:", 0) == 0) {
		std::string payload = line.substr(5);
		size_t p = payload.find_first_not_of(' ');
		return p == std::string::npos ? "" : payload.substr(p);
		}
		if (eol == std::string::npos) break;
		start = eol + 1;
		}
		return std::string();
		}

		} // namespace

		json_object *webedit::Api::callAiTool(const std::string &sessionid, const std::string &name,
		@@ -3700,6 +3719,14 @@ void webedit::Api::handleAiPrompt(libhttppp::HttpRequest &curreq,
		std::string apiPath = "/v1/chat/completions";
		std::string fullUrl = aiUrl + apiPath;

		// True streaming (progressive SSE to the browser) needs outbound chunked-
		// encoding, which this codebase only implements for HTTP/1.1 connections
		// (see HttpResponse::sendChunkedHeaders/sendChunk/endChunked). HTTP/2/3
		// connections fall back to the original single-shot buffered loop below.
		const std::string &reqVersion = curreq.getRequestVersion();
		bool canStream = (reqVersion == "HTTP/1.1" \|\| reqVersion == "HTTP/1.0");

		if (!canStream) {
		std::string finalResult;
		bool anyMutation = false;
		int toolCallCount = 0;
		@@ -3846,6 +3873,241 @@ void webedit::Api::handleAiPrompt(libhttppp::HttpRequest &curreq,
		json_object_object_add(resp, "tool_calls", json_object_new_int(toolCallCount));
		sendJson(curreq, resp);
		json_object_put(resp);
		return;
		}

		// ---------------- Streaming path (HTTP/1.1) ----------------
		// Sends Server-Sent Events as soon as they're produced: content_delta for
		// partial assistant text, tool_call right before each tool runs, and a
		// final done event with the same {result, applied, tool_calls} shape the
		// non-streaming path returns as one JSON blob. A failed write (browser
		// disconnected / Stop clicked, which tears down the TCP connection) is
		// treated as "client gone" and ends the loop early -- a real cancel,
		// unlike the old design where the server always ran to completion
		// regardless of whether anyone was still waiting for the response.
		libhttppp::HttpResponse sseResp;
		sseResp.setContentType("text/event-stream; charset=utf-8");
		sseResp.sendChunkedHeaders(curreq);
		curreq.flushSendData();

		bool clientGone = false;
		auto pushEvent = [&](json_object *payload) {
		const char *s = json_object_to_json_string_ext(payload, JSON_C_TO_STRING_NOSLASHESCAPE);
		std::string frame = "data: " + std::string(s) + "\n\n";
		json_object_put(payload);
		if (clientGone) return;
		try {
		sseResp.sendChunk(curreq, frame.data(), frame.size());
		curreq.flushSendData();
		} catch (const std::exception &) {
		clientGone = true;
		}
		};

		std::string finalResult;
		bool anyMutation = false;
		int toolCallCount = 0;
		bool haveFinal = false;

		try {
		for (int round = 0; round < kMaxAiToolRounds && !haveFinal && !clientGone; ++round) {
		json_object *aiReq = json_object_new_object();
		json_object_object_add(aiReq, "model", json_object_new_string(aiModel.c_str()));
		json_object_object_add(aiReq, "max_tokens", json_object_new_int(maxTokens));
		json_object_object_add(aiReq, "stream", json_object_new_boolean(true));
		json_object_object_add(aiReq, "messages", json_object_get(messages));
		json_object_object_add(aiReq, "tools", json_object_get(toolDefs));
		json_object_object_add(aiReq, "tool_choice", json_object_new_string("auto"));

		const char *aiReqStr = json_object_to_json_string_ext(aiReq, JSON_C_TO_STRING_NOSLASHESCAPE);
		std::vector<char> postData(aiReqStr, aiReqStr + strlen(aiReqStr));
		json_object_put(aiReq);

		libhttppp::HttpUrl url(fullUrl);
		libhttppp::HttpClient client(url);
		client.setTimeout(120);

		libhttppp::HttpRequest apiReq;
		apiReq.setRequestType(POSTREQUEST);
		apiReq.setRequestURL(apiPath);
		apiReq.setHeaderData("content-type")->push_back("application/json");
		if (!aiApiKey.empty())
		apiReq.setHeaderData("authorization")->push_back("Bearer " + aiApiKey);

		client.PostStream(apiReq, postData);

		std::string sseBuf;
		std::string roundContent;

		struct ToolAccum { std::string id, name, args; };
		std::map<int, ToolAccum> toolAccum;
		std::vector<int> toolOrder;

		char buf[4096];
		size_t n;
		while (!clientGone && (n = client.readBodyChunk(buf, sizeof(buf))) > 0) {
		sseBuf.append(buf, n);
		size_t pos;
		while ((pos = sseBuf.find("\n\n")) != std::string::npos) {
		std::string block = sseBuf.substr(0, pos);
		sseBuf.erase(0, pos + 2);

		std::string payload = extractSseDataLine(block);
		if (payload.empty() \|\| payload == "[DONE]")
		continue;

		json_object *chunkObj = json_tokener_parse(payload.c_str());
		if (!chunkObj) continue;

		json_object *choices = nullptr;
		if (json_object_object_get_ex(chunkObj, "choices", &choices) &&
		json_object_array_length(choices) > 0) {
		json_object *choice = json_object_array_get_idx(choices, 0);

		json_object *delta = nullptr;
		json_object_object_get_ex(choice, "delta", &delta);
		if (delta) {
		json_object *jContent = nullptr;
		if (json_object_object_get_ex(delta, "content", &jContent) && jContent &&
		json_object_get_type(jContent) == json_type_string) {
		std::string piece = json_object_get_string(jContent);
		if (!piece.empty()) {
		roundContent += piece;
		json_object *ev = json_object_new_object();
		json_object_object_add(ev, "type", json_object_new_string("content_delta"));
		json_object_object_add(ev, "text", json_object_new_string(piece.c_str()));
		pushEvent(ev);
		}
		}

		json_object *jToolCalls = nullptr;
		if (json_object_object_get_ex(delta, "tool_calls", &jToolCalls) && jToolCalls &&
		json_object_get_type(jToolCalls) == json_type_array) {
		int tn = json_object_array_length(jToolCalls);
		for (int ti = 0; ti < tn; ++ti) {
		json_object *tcFrag = json_object_array_get_idx(jToolCalls, ti);
		json_object *jIdx = nullptr;
		int idx = 0;
		if (json_object_object_get_ex(tcFrag, "index", &jIdx) && jIdx)
		idx = json_object_get_int(jIdx);

		if (toolAccum.find(idx) == toolAccum.end())
		toolOrder.push_back(idx);
		ToolAccum &acc = toolAccum[idx];

		std::string fragId = jsonStringField(tcFrag, "id");
		if (!fragId.empty()) acc.id = fragId;

		json_object *fn = nullptr;
		if (json_object_object_get_ex(tcFrag, "function", &fn) && fn) {
		std::string fragName = jsonStringField(fn, "name");
		if (!fragName.empty()) acc.name = fragName;
		std::string fragArgs = jsonStringField(fn, "arguments");
		if (!fragArgs.empty()) acc.args += fragArgs;
		}
		}
		}
		}
		}
		json_object_put(chunkObj);
		}
		}

		if (clientGone) break;

		if (!toolOrder.empty()) {
		json_object *asstMsg = json_object_new_object();
		json_object_object_add(asstMsg, "role", json_object_new_string("assistant"));
		json_object *tcArr = json_object_new_array();
		for (int idx : toolOrder) {
		ToolAccum &acc = toolAccum[idx];
		json_object *tcObj = json_object_new_object();
		json_object_object_add(tcObj, "id", json_object_new_string(acc.id.c_str()));
		json_object_object_add(tcObj, "type", json_object_new_string("function"));
		json_object *fnObj = json_object_new_object();
		json_object_object_add(fnObj, "name", json_object_new_string(acc.name.c_str()));
		json_object_object_add(fnObj, "arguments", json_object_new_string(acc.args.c_str()));
		json_object_object_add(tcObj, "function", fnObj);
		json_object_array_add(tcArr, tcObj);
		}
		json_object_object_add(asstMsg, "tool_calls", tcArr);
		json_object_array_add(messages, asstMsg);

		for (int idx : toolOrder) {
		ToolAccum &acc = toolAccum[idx];

		json_object *progressEv = json_object_new_object();
		json_object_object_add(progressEv, "type", json_object_new_string("tool_call"));
		json_object_object_add(progressEv, "name", json_object_new_string(acc.name.c_str()));
		pushEvent(progressEv);

		json_object *argsObj = acc.args.empty() ? nullptr : json_tokener_parse(acc.args.c_str());
		bool ownsArgsObj = (argsObj != nullptr);
		if (!argsObj) { argsObj = json_object_new_object(); ownsArgsObj = true; }

		json_object *toolResultPayload = nullptr;
		bool mutated = false;
		try {
		toolResultPayload = callAiTool(sessionid, acc.name, argsObj, mutated);
		} catch (const McpError &e) {
		toolResultPayload = json_object_new_object();
		json_object_object_add(toolResultPayload, "error", json_object_new_string(e.what()));
		} catch (const std::exception &e) {
		toolResultPayload = json_object_new_object();
		json_object_object_add(toolResultPayload, "error", json_object_new_string(e.what()));
		}
		if (ownsArgsObj) json_object_put(argsObj);

		anyMutation = anyMutation \|\| mutated;
		++toolCallCount;

		const char *resultStr = json_object_to_json_string_ext(
		toolResultPayload, JSON_C_TO_STRING_NOSLASHESCAPE);

		json_object *toolMsg = json_object_new_object();
		json_object_object_add(toolMsg, "role", json_object_new_string("tool"));
		if (!acc.id.empty())
		json_object_object_add(toolMsg, "tool_call_id", json_object_new_string(acc.id.c_str()));
		json_object_object_add(toolMsg, "content", json_object_new_string(resultStr));
		json_object_array_add(messages, toolMsg);

		json_object_put(toolResultPayload);
		}
		continue;
		}

		// No tool calls this round -- final answer.
		finalResult = roundContent;
		haveFinal = true;
		}
		} catch (const std::exception &e) {
		if (!clientGone) {
		json_object *errEv = json_object_new_object();
		json_object_object_add(errEv, "type", json_object_new_string("error"));
		json_object_object_add(errEv, "message", json_object_new_string(e.what()));
		pushEvent(errEv);
		}
		}

		json_object_put(messages);
		json_object_put(toolDefs);

		if (!haveFinal && finalResult.empty() && !clientGone)
		finalResult = "(Antwort abgebrochen: maximale Anzahl an Werkzeugaufrufen erreicht.)";

		if (!clientGone) {
		json_object *doneEv = json_object_new_object();
		json_object_object_add(doneEv, "type", json_object_new_string("done"));
		json_object_object_add(doneEv, "result", json_object_new_string(finalResult.c_str()));
		json_object_object_add(doneEv, "applied", json_object_new_boolean(anyMutation));
		json_object_object_add(doneEv, "tool_calls", json_object_new_int(toolCallCount));
		pushEvent(doneEv);
		}

		try {
		sseResp.endChunked(curreq);
		curreq.flushSendData();
		} catch (const std::exception &) {}
		}

		// --- AI Chat persistence handler ---